Kafka-Nachrichten in Spring Boot senden und empfangen

1. Übersicht

Apache Kafka ist eine beliebte Event-Streaming-Plattform für asynchrone Kommunikation zwischen Services. Spring Boot unterstützt Kafka durch die Spring-Kafka-Bibliothek, wodurch Sie KafkaProducer und die Polling-Schleife poll() nicht selbst verwalten müssen.

In diesem Artikel bauen Sie ein kleines Beispiel auf: Ein Bestellservice sendet das Event OrderCreatedEvent, ein Consumer empfängt es, um Bestände im Lager zu reservieren. Nach diesem Artikel wissen Sie, wie Sie konfigurieren, JSON-Nachrichten versenden und empfangen, Fehler mit dem Dead Letter Topic behandeln und mit Embedded Kafka testen.

2. Vorbereitung

2.1. Maven-Abhängigkeiten

Sie benötigen spring-kafka. Spring Boot verwaltet die Version bereits, daher müssen Sie die Version nicht angeben:

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

<!-- Für Tests -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-test</artifactId>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka-test</artifactId>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.awaitility</groupId>
    <artifactId>awaitility</artifactId>
    <scope>test</scope>
</dependency>

spring-kafka-test stellt einen In-Memory Kafka-Broker für Tests bereit. Awaitility hilft beim Warten auf asynchrone Verarbeitung ohne Thread.sleep() zu verwenden.

2.2. Kafka lokal ausführen

Die schnellste Methode ist, das offizielle Apache-Kafka-Image im KRaft-Modus zu verwenden, d.h. ohne ZooKeeper:

docker run -d --name kafka -p 9092:9092 apache/kafka:latest

Der Broker wird auf localhost:9092 lauschen.

3. Datenmodell und Konfiguration

Das zu sendende Event ist ein einfacher record:

package com.nonla.shop.order;

import java.math.BigDecimal;

public record OrderCreatedEvent(String orderId, String customerId, BigDecimal amount) {
}

Dann folgt die Konfiguration in application.yml:

spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      acks: all
    consumer:
      group-id: inventory-service
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        spring.json.trusted.packages: "com.nonla.shop.order"

Bedeutung der einzelnen Teile:

• bootstrap-servers ist die Broker-Adresse.
• Der Producer verwendet StringSerializer für den Schlüssel und JsonSerializer für den Wert. JsonSerializer verwendet Jackson, um Objekte in JSON zu konvertieren.
• acks: all zwingt den Broker, nur zu bestätigen, nachdem alle synchronisierten Repliken geschrieben haben. Dies ist sicherer, aber etwas langsamer.
• group-id definiert die Consumer-Gruppe. Instanzen in derselben Gruppe teilen sich die Partitionen.
• auto-offset-reset: earliest ermöglicht dem Consumer, von Anfang an zu lesen, wenn der Gruppe noch kein Offset committed wurde.
• spring.json.trusted.packages deklariert die Pakete, aus denen JsonDeserializer Objekte erstellen darf. Ohne diese Zeile erhalten Sie die Fehlermeldung “The class … is not in the trusted packages”.

3.1. Topic automatisch erstellen

Spring Boot konfiguriert automatisch einen KafkaAdmin. Dieser Bean erstellt alle NewTopic-Beans, die im Context deklariert sind:

package com.nonla.shop.config;

import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.TopicBuilder;

@Configuration
public class KafkaTopicConfig {

    public static final String ORDERS_TOPIC = "orders";

    @Bean
    public NewTopic ordersTopic() {
        return TopicBuilder.name(ORDERS_TOPIC)
                .partitions(3)
                .replicas(1)
                .build();
    }
}

Beim Starten der Anwendung wird das Topic orders mit 3 Partitionen erstellt, wenn es noch nicht vorhanden ist. Wenn das Topic bereits vorhanden ist, überspringt KafkaAdmin es und überschreibt die aktuelle Konfiguration nicht.

4. Nachrichten mit KafkaTemplate versenden

Spring Boot erstellt automatisch einen KafkaTemplate-Bean basierend auf der Producer-Konfiguration. Sie müssen ihn nur injizieren:

package com.nonla.shop.order;

import com.nonla.shop.config.KafkaTopicConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;

import java.util.concurrent.CompletableFuture;

@Service
public class OrderEventPublisher {

    private static final Logger log = LoggerFactory.getLogger(OrderEventPublisher.class);

    private final KafkaTemplate<String, OrderCreatedEvent> kafkaTemplate;

    public OrderEventPublisher(KafkaTemplate<String, OrderCreatedEvent> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public CompletableFuture<SendResult<String, OrderCreatedEvent>> publish(OrderCreatedEvent event) {
        return kafkaTemplate.send(KafkaTopicConfig.ORDERS_TOPIC, event.orderId(), event)
                .whenComplete((result, ex) -> {
                    if (ex != null) {
                        log.error("Order {} konnte nicht gesendet werden", event.orderId(), ex);
                        return;
                    }
                    var metadata = result.getRecordMetadata();
                    log.info("Order {} an Partition {} Offset {} gesendet",
                            event.orderId(), metadata.partition(), metadata.offset());
                });
    }
}

Punkte zu beachten:

• send(topic, key, value) verwendet orderId als Schlüssel. Kafka hasht den Schlüssel, um eine Partition zu wählen, daher gehen alle Nachrichten für eine Bestellung in dieselbe Partition. Dies bewahrt die Reihenfolge zwischen ihnen.
• Ab Spring Kafka 3.0 gibt send() CompletableFuture<SendResult> statt des veralteten ListenableFuture zurück.
• send() ist asynchron. Der Aufruf kehrt sofort zurück, der Callback in whenComplete läuft, wenn der Broker antwortet.

Bei erfolgreichem Versand sieht das Log so aus:

INFO  c.n.s.order.OrderEventPublisher : Order ORD-1001 an Partition 2 Offset 0 gesendet

Wenn Sie einen Thread blockieren müssen, bis der Versand abgeschlossen ist, z.B. in einem Batch-Job, können Sie publish(event).get(5, TimeUnit.SECONDS) aufrufen. Machen Sie dies nicht im Request-Verarbeitungs-Thread bei hoher Last, da jeder Aufruf einen Thread blockieren würde, der auf den Broker wartet.

Hinweis: Das Schreiben in die Datenbank und das Versenden an Kafka in derselben Methode ist keine atomare Transaktion. Wenn das Datenbank-Commit erfolgreich ist, aber das Versenden an Kafka fehlschlägt, sind die Daten nicht konsistent. Bei kritischen Daten sollten Sie das Transactional Outbox Pattern in Betracht ziehen.

5. Nachrichten mit @KafkaListener empfangen

Auf der Consumer-Seite brauchen Sie nur eine Methode mit der Annotation @KafkaListener:

package com.nonla.shop.inventory;

import com.nonla.shop.config.KafkaTopicConfig;
import com.nonla.shop.order.OrderCreatedEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

@Component
public class OrderCreatedListener {

    private static final Logger log = LoggerFactory.getLogger(OrderCreatedListener.class);

    private final InventoryService inventoryService;

    public OrderCreatedListener(InventoryService inventoryService) {
        this.inventoryService = inventoryService;
    }

    @KafkaListener(topics = KafkaTopicConfig.ORDERS_TOPIC, concurrency = "3")
    public void onOrderCreated(@Payload OrderCreatedEvent event,
                               @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
                               @Header(KafkaHeaders.OFFSET) long offset) {
        log.info("Order {} empfangen von Partition {} Offset {}", event.orderId(), partition, offset);
        inventoryService.reserve(event);
    }
}

InventoryService ist nur ein normaler Service:

package com.nonla.shop.inventory;

import com.nonla.shop.order.OrderCreatedEvent;
import org.springframework.stereotype.Service;

@Service
public class InventoryService {

    public void reserve(OrderCreatedEvent event) {
        if (event.amount().signum() <= 0) {
            throw new IllegalArgumentException("Ungültiger Betrag: " + event.amount());
        }
        // Bestand im Lager reservieren...
    }
}

Erklärung:

• topics gibt das Topic an, das überwacht werden soll. groupId kommt aus spring.kafka.consumer.group-id, oder Sie können es direkt in der Annotation überschreiben.
• concurrency = "3" erstellt 3 Consumer-Threads, jeder verwaltet eine Partition. Ein Wert größer als die Anzahl der Partitionen ist nutzlos, da überschüssige Threads nichts erhalten.
• @Payload empfängt das deserialisierte Objekt. @Header erhält Metadaten wie Partition und Offset.
• Standardmäßig wird der Offset nach Abschluss der Methode ohne Exception committed (AckMode.BATCH). Wenn die Anwendung abstürzt, wird die Nachricht daher erneut verarbeitet. Dies ist at-least-once-Semantik, also muss Ihre Logik idempotent sein: das zweimalige Verarbeiten derselben orderId darf den Bestand nicht zweimal reservieren.

Wenn Sie auch den Schlüssel, Zeitstempel oder benutzerdefinierte Header benötigen, können Sie den gesamten ConsumerRecord<String, OrderCreatedEvent> als Parameter empfangen.

6. Fehlerbehandlung und Dead Letter Topic

Wenn der Listener eine Exception wirft, versucht DefaultErrorHandler standardmäßig 9-mal erneut, protokolliert dann und überspringt die Nachricht. In der Praxis möchten Sie fehlerhafte Nachrichten normalerweise in ein separates Topic zur späteren Überprüfung verschieben:

package com.nonla.shop.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
import org.springframework.kafka.listener.DefaultErrorHandler;
import org.springframework.util.backoff.FixedBackOff;

@Configuration
public class KafkaErrorConfig {

    @Bean
    public DefaultErrorHandler errorHandler(KafkaTemplate<Object, Object> kafkaTemplate) {
        var recoverer = new DeadLetterPublishingRecoverer(kafkaTemplate);
        var handler = new DefaultErrorHandler(recoverer, new FixedBackOff(1000L, 2));
        handler.addNotRetryableExceptions(IllegalArgumentException.class);
        return handler;
    }
}

Spring Boot bindet diesen CommonErrorHandler-Bean automatisch an die Standard-Container-Factory. Die obige Konfiguration funktioniert wie folgt:

• FixedBackOff(1000L, 2) versucht maximal 2-mal mit jeweils 1 Sekunde Abstand erneut.
• Nach Abschluss der Versuche sendet DeadLetterPublishingRecoverer die Nachricht an das Topic orders.DLT. Dieser Name ist der ursprüngliche Name plus Suffix .DLT, und die Nachricht behält ihre Partition. Das DLT-Topic benötigt daher mindestens so viele Partitionen wie das Original-Topic.
• IllegalArgumentException ist ein Datenfehler, daher ist ein erneuter Versuch nutzlos. Der Handler verschiebt die Nachricht direkt in das DLT.

Eine Nachricht mit amount = -5 ergibt ähnliche Logs:

ERROR o.s.k.l.DefaultErrorHandler : Backoff none exhausted for orders-1@4
INFO  c.n.s.inventory.DltListener  : Fehlernachricht im DLT: ORD-1002

Hinweis: Fehler während der Deserialisierung, z.B. ungültige JSON, erreichen den Listener nicht. Ohne Behandlung kann der Consumer bei dieser Nachricht steckenbleiben – das nennt man “Poison Pill”. Wickeln Sie den Deserializer mit ErrorHandlingDeserializer ein:

spring:
  kafka:
    consumer:
      value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
      properties:
        spring.deserializer.value.delegate.class: org.springframework.kafka.support.serializer.JsonDeserializer

Mit dieser Konfiguration werden Deserialisierungsfehler an DefaultErrorHandler weitergeleitet und die fehlerhafte Nachricht geht ebenfalls ins DLT.

7. Testen mit Embedded Kafka

@EmbeddedKafka startet einen In-Memory-Broker, daher benötigen Sie Docker nicht beim Ausführen von Tests. Das folgende Beispiel testet den kompletten Kreislauf: Versand über OrderEventPublisher, dann Bestätigung, dass InventoryService das richtige Event erhält.

package com.nonla.shop.order;

import com.nonla.shop.inventory.InventoryService;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.test.context.EmbeddedKafka;
import org.springframework.test.context.bean.override.mockito.MockitoBean;

import java.math.BigDecimal;
import java.time.Duration;

import static org.assertj.core.api.Assertions.assertThat;
import static org.awaitility.Awaitility.await;
import static org.mockito.Mockito.verify;

@SpringBootTest(properties = "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}")
@EmbeddedKafka(partitions = 3, topics = "orders")
class OrderKafkaIntegrationTest {

    @Autowired
    private OrderEventPublisher publisher;

    @MockitoBean
    private InventoryService inventoryService;

    @Test
    void whenOrderPublished_thenListenerReservesInventory() throws Exception {
        var event = new OrderCreatedEvent("ORD-1001", "CUS-42", new BigDecimal("250000"));

        var result = publisher.publish(event).get();
        assertThat(result.getRecordMetadata().topic()).isEqualTo("orders");

        var captor = ArgumentCaptor.forClass(OrderCreatedEvent.class);
        await().atMost(Duration.ofSeconds(10))
                .untilAsserted(() -> verify(inventoryService).reserve(captor.capture()));

        assertThat(captor.getValue()).isEqualTo(event);
    }
}

Was jeder Teil tut:

• ${spring.embedded.kafka.brokers} ist eine Property, die @EmbeddedKafka bereitstellt. Sie enthält die Adresse des eingebetteten Brokers, die wir verwenden, um bootstrap-servers zu überschreiben.
• @MockitoBean (verfügbar ab Spring Boot 3.4) ersetzt den echten InventoryService durch einen Mock zur leichteren Überprüfung. In älteren Versionen verwenden Sie @MockBean.
• publish(event).get() wartet auf die Broker-Bestätigung, sodass der Test sicherstellt, dass die Nachricht geschrieben wurde.
• Der Listener läuft auf einem anderen Thread, daher wiederholt await().untilAsserted(...) die verify-Aussage für maximal 10 Sekunden.
• Da OrderCreatedEvent ein record ist, vergleicht equals() nach Wert. assertThat(...).isEqualTo(event) bestätigt also, dass JSON korrekt serialisiert und deserialisiert wurde.

Genau auto-offset-reset: earliest in der Konfiguration macht den Test stabil. Wenn der Consumer der Gruppe nach dem Versand beitritt, kann er die Nachricht immer noch lesen.

8. Häufige Fehler und Fälle, in denen Kafka nicht geeignet ist

Einige Fehler, die Sie leicht antreffen:

• Fehlende trusted.packages: JsonDeserializer weigert sich, unbekannte Klassen zu erstellen. Deklarieren Sie korrekt die Pakete. Verwenden Sie "*" nur, wenn Sie vollständig alle Producer kontrollieren.
• Producer und Consumer in verschiedenen Paketen: JsonSerializer schreibt standardmäßig den vollständig qualifizierten Klassennamen in den Header __TypeId__. Wenn der Consumer-Service eine Klasse mit demselben Namen, aber in einem anderen Paket hat, schlägt die Deserialisierung fehl. Dann setzen Sie spring.json.use.type.headers: false und spring.json.value.default.type: com.nonla.inventory.OrderCreatedEvent auf der Consumer-Seite.
• Listener läuft lange: Wenn die Verarbeitung eines poll() max.poll.interval.ms überschreitet (Standard 5 Minuten), betrachtet der Broker den Consumer als tot und startet ein Rebalancing. Das Ergebnis ist doppelte Nachrichtenverarbeitung. Reduzieren Sie max.poll.records oder optimieren Sie die Logik.
• concurrency größer als Partitionszahl: Überschüssige Threads bleiben inaktiv. Um den Durchsatz zu erhöhen, müssen Sie zunächst die Anzahl der Partitionen erhöhen.
• Versand mit null key: Nachrichten werden gleichmäßig über Partitionen verteilt, daher verlieren Sie die Reihenfolgegarantie pro Entität.

Kafka ist auch nicht immer die richtige Wahl:

• Sie benötigen synchrone Request/Response mit sofortigen Ergebnissen. Dann verwenden Sie REST oder gRPC.
• Kleines System, niedriger Durchsatz und keine Notwendigkeit, Event-Historie zu speichern. Der Betrieb eines Kafka-Clusters ist dann ein Mehraufwand, und RabbitMQ oder eine Queue-Tabelle in der Datenbank könnte ausreichen.
• Sie benötigen komplexe nachrichtenbasierte Routing oder spezifische Latenzen und Prioritäten pro Nachricht. Das ist die Stärke traditioneller Message Broker.

9. Fazit

In diesem Artikel haben Sie Spring Kafka mit JSON-Serializer konfiguriert und ein Topic mit NewTopic erstellt. Sie haben Nachrichten über KafkaTemplate mit Schlüssel zum Erhalt der Reihenfolge versendet und Nachrichten mit @KafkaListener mit concurrency empfangen. Sie haben auch Fehler mit DefaultErrorHandler in Kombination mit Dead Letter Topic behandelt und Integration-Tests mit @EmbeddedKafka und Awaitility geschrieben.

Denken Sie beim Einsatz in der Produktion an drei Punkte: Der Listener muss idempotent sein, Deserialisierungsfehler müssen mit ErrorHandlingDeserializer behandelt werden, und das Event-Versenden muss mit Datenbankdaten konsistent sein.