Kafka Retry und Dead Letter Topic mit Spring Kafka

1. Übersicht

Ein Kafka Consumer wird zwangsläufig auf fehlerhafte Nachrichten stoßen. Es gibt vorübergehende Fehler, zum Beispiel wenn das Zahlungsgateway Timeout aufweist. Und es gibt permanente Fehler, zum Beispiel wenn die Daten falsch formatiert sind. Ohne richtige Behandlung landest du in einer von zwei Situationen: Nachrichten gehen verloren, oder der Consumer hängt bleibend bei einer Nachricht fest, die niemals erfolgreich wird.

Dieser Beitrag zeigt dir, wie du Retry in Spring Kafka auf zwei Arten konfigurierst: Blocking mit DefaultErrorHandler und Non-Blocking mit @RetryableTopic. Nachrichten, die weiterhin fehlschlagen, nachdem die Wiederholungsversuche aufgebraucht sind, werden zu einem dead letter topic (DLT) weitergeleitet. Nach dem Lesen weißt du, wann du welche Variante wählst und kannst Tests schreiben, um das Verhalten zu überprüfen.

2. Einrichtung

Füge die Spring Kafka Dependency hinzu. Spring Boot 3.x verwaltet Versionen automatisch, daher musst du keine Version deklarieren:

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

Minimale Konfiguration in application.properties:

spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.consumer.group-id=order-service
spring.kafka.consumer.auto-offset-reset=earliest

Boot verwendet standardmäßig StringSerializer und StringDeserializer. Um das Beispiel kompakt zu halten, ist der Message-Value eine Bestellnummer als String.

3. Consumer mit möglichen Fehlern

Das ist ein Service für die Zahlungsverarbeitung. Eine ungültig formatierte Bestellnummer ist ein permanenter Fehler. Ein Fehler beim Aufruf des Zahlungsgateways ist ein vorübergehender Fehler:

public class PaymentGatewayException extends RuntimeException {
    public PaymentGatewayException(String message) {
        super(message);
    }
}

@Service
public class PaymentService {

    public void charge(String orderId) {
        if (orderId == null || !orderId.startsWith("ORD-")) {
            throw new IllegalArgumentException("Ungültige Bestellnummer: " + orderId);
        }
        // Ruft das Zahlungsgateway auf; kann PaymentGatewayException werfen, wenn Timeout auftritt
    }
}

Der Listener ruft nur den Service auf:

@Component
public class OrderListener {

    private static final Logger log = LoggerFactory.getLogger(OrderListener.class);
    private final PaymentService paymentService;

    public OrderListener(PaymentService paymentService) {
        this.paymentService = paymentService;
    }

    @KafkaListener(topics = "orders")
    public void onOrder(String orderId) {
        log.info("Bearbeite Bestellung {}", orderId);
        paymentService.charge(orderId);
    }
}

Wenn du nichts konfigurierst, verwendet Spring Kafka DefaultErrorHandler mit FixedBackOff(0, 9). Fehlgeschlagene Nachrichten werden 9 Mal hintereinander wiederholt, insgesamt 10 Mal. Danach protokolliert der Handler nur und ignoriert die Nachricht. Sie gilt als verloren. Das ist der Grund, warum du die Konfiguration anpassen musst.

4. Retry Blocking mit DefaultErrorHandler

4.1. Backoff- und DLT-Konfiguration

Wenn der Listener eine Exception wirft, sucht DefaultErrorHandler den Offset der fehlgeschlagenen Nachricht erneut auf, damit sie bei der nächsten Abfrage erneut gelesen wird. Nach Ablauf der Wiederholungsversuche ruft der Handler einen ConsumerRecordRecoverer auf. Hier ist der Recoverer DeadLetterPublishingRecoverer:

@Configuration
public class KafkaErrorConfig {

    @Bean
    public DefaultErrorHandler errorHandler(KafkaTemplate<Object, Object> template) {
        var recoverer = new DeadLetterPublishingRecoverer(template);

        var backOff = new ExponentialBackOffWithMaxRetries(3);
        backOff.setInitialInterval(1_000L);
        backOff.setMultiplier(2.0);
        backOff.setMaxInterval(10_000L);

        var handler = new DefaultErrorHandler(recoverer, backOff);
        handler.addNotRetryableExceptions(IllegalArgumentException.class);
        return handler;
    }
}

Giải thích từng phần:

• DeadLetterPublishingRecoverer(template) sendet die fehlerhafte Nachricht an topic <topic-name>.DLT, in der gleichen Partition wie die ursprüngliche Nachricht. Für topic orders ist das DLT orders.DLT.
• ExponentialBackOffWithMaxRetries(3) wiederholt maximal 3 Mal. Die Wartezeiten betragen jeweils 1s, 2s, 4s und überschreiten nicht 10s.
• addNotRetryableExceptions deklariert Exceptions, die nicht wiederholt werden müssen. IllegalArgumentException geht direkt zum DLT, denn egal wie oft wiederholt wird - falsche Daten bleiben falsch.

Spring Boot bindet das Bean CommonErrorHandler automatisch in die Standard-Container-Factory ein. Du brauchst keine ConcurrentKafkaListenerContainerFactory selbst zu erstellen.

Wenn du ORD-1 sendest und das Zahlungsgateway ständig Timeout aufweist, sieht das Log ungefähr so aus:

10:00:00.010 INFO  Bearbeite Bestellung ORD-1
10:00:01.020 INFO  Bearbeite Bestellung ORD-1
10:00:03.030 INFO  Bearbeite Bestellung ORD-1
10:00:07.040 INFO  Bearbeite Bestellung ORD-1
10:00:07.060 ERROR Backoff ... exhausted for orders-0@42

Insgesamt gibt es 4 Verarbeitungen: 1 Initial und 3 Wiederholungen. Danach liegt die Nachricht in orders.DLT.

4.2. Header im DLT

Nachrichten im DLT behalten ihren Key und Value. Der Recoverer fügt Header hinzu, damit du die Ursache ermitteln kannst:

kafka_dlt-original-topic      = orders
kafka_dlt-original-partition  = 0
kafka_dlt-original-offset     = 42
kafka_dlt-exception-fqcn      = org.springframework.kafka.listener.ListenerExecutionFailedException
kafka_dlt-exception-cause-fqcn = com.example.PaymentGatewayException
kafka_dlt-exception-message   = Listener-Methode ... hat Exception geworfen
kafka_dlt-exception-stacktrace = ...

Deine ursprüngliche Exception ist in kafka_dlt-exception-cause-fqcn. Wenn du Header per Code lesen musst, verwendest du die Konstanten in KafkaHeaders, z.B. KafkaHeaders.DLT_ORIGINAL_TOPIC.

Hinweis: Da der Recoverer standardmäßig in die gleiche Partition sendet, muss orders.DLT mindestens so viele Partitionen wie orders haben. Andernfalls schlägt die Veröffentlichung fehl. Du kannst das DLT mit einem NewTopic-Bean erstellen, das die gleiche Anzahl Partitionen hat. Alternativ kannst du einen Destination-Resolver übergeben, der new TopicPartition(r.topic() + ".DLT", -1) zurückgibt, damit Kafka die Partition selbst wählt.

4.3. Wann ist Retry Blocking sinnvoll

Retry Blocking wahrt die Reihenfolge der Nachrichten in der Partition. Eine Nachricht wird erst verarbeitet, wenn die vorherige erfolgreich war oder zum DLT verschoben wurde. Das ist die richtige Wahl, wenn die Reihenfolge wichtig ist, z.B. bei Events wie OrderCreated → OrderPaid → OrderShipped derselben Bestellung.

Der Preis ist, dass die gesamte Partition während des Wartens blockiert ist. Ab Spring Kafka 2.9 wartet der Handler auf das Backoff, ohne max.poll.interval.ms zu überschreiten, dank Seek- und Pause-Mechanismus. Wenn du jedoch das Backoff auf einige Minuten insgesamt setzt, wird der Lag stark ansteigen.

5. Retry Non-Blocking mit @RetryableTopic

5.1. Konfiguration per Annotation

Bei Non-Blocking-Retry wird die fehlgeschlagene Nachricht in ein separates Retry-Topic verschoben. Dadurch kann der Consumer sofort mit der nächsten Nachricht fortfahren. Ein anderer Consumer liest das Retry-Topic nach der definierten Backoff-Zeit:

@Component
public class OrderRetryListener {

    private static final Logger log = LoggerFactory.getLogger(OrderRetryListener.class);
    private final PaymentService paymentService;

    public OrderRetryListener(PaymentService paymentService) {
        this.paymentService = paymentService;
    }

    @RetryableTopic(
            attempts = "4",
            backoff = @Backoff(delay = 1_000, multiplier = 2.0, maxDelay = 10_000),
            exclude = IllegalArgumentException.class)
    @KafkaListener(topics = "payments")
    public void onPayment(String orderId,
                          @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
        log.info("Verarbeite {} von {}", orderId, topic);
        paymentService.charge(orderId);
    }

    @DltHandler
    public void onDlt(String orderId,
                      @Header(KafkaHeaders.EXCEPTION_MESSAGE) String error) {
        log.error("Bestellung {} geht zum DLT: {}", orderId, error);
    }
}

Die Hauptpunkte:

• attempts = "4" bedeutet insgesamt 4 Verarbeitungen: 1 Mal im ursprünglichen Topic und 3 Mal im Retry-Topic. Der Wert ist ein String, um Platzhalter wie "${app.retry.attempts}" zu unterstützen.
• @Backoff stammt aus dem Package org.springframework.retry.annotation. Spring Retry ist bereits in spring-kafka enthalten.
• exclude listet Exceptions auf, die direkt zum DLT gehen, gleichbedeutend mit addNotRetryableExceptions.
• @DltHandler wird in der gleichen Klasse platziert. Hier protokollierst du, sendest Warnungen oder speicherst für manuelle Verarbeitung.

5.2. Erstellte Topics

Bei schrittweise zunehmendem Backoff hat standardmäßig jeder Wiederholungsversuch ein eigenes Topic mit einem Suffix basierend auf der Verzögerung:

payments
payments-retry-1000
payments-retry-2000
payments-retry-4000
payments-dlt

Spring Kafka erstellt diese Topics beim Start automatisch (autoCreateTopics = "true" ist Standard). In der Produktionsumgebung ist Auto-Create auf dem Broker normalerweise deaktiviert. Dann musst du die Topics vorher mit den richtigen Namen und Partitionsanzahl erstellen. Wenn du einen Index statt eine Verzögerung als Suffix verwenden möchtest, setze topicSuffixingStrategy = TopicSuffixingStrategy.SUFFIX_WITH_INDEX_VALUE.

Beachte das Suffix: @RetryableTopic verwendet -dlt, während DeadLetterPublishingRecoverer .DLT verwendet. Merke dir das beim Konfigurieren des Monitoring.

5.3. Kompromisse von Non-Blocking

Non-Blocking-Retry bricht die Reihenfolge. Angenommen, ORD-1 schlägt fehl und wird zum Retry-Topic verschoben, während ORD-2 sofort erfolgreich ist. Dann wird ORD-2 vor ORD-1 verarbeitet. Wenn dein Geschäftsfluss von der Event-Reihenfolge abhängt, nutze diese Methode nicht.

Außerdem nimmt die Anzahl der Topics und Consumer zu. Im Gegenzug blockiert eine einzelne fehlerhafte Nachricht nicht die ganze Partition.

6. DLT-Fluss testen

Der folgende Test nutzt @EmbeddedKafka. Er sendet eine ungültig formatierte Bestellnummer und überprüft, dass die Nachricht in orders.DLT erscheint, ohne wiederholt zu werden:

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

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    private EmbeddedKafkaBroker broker;

    @Test
    void givenInvalidOrderId_whenConsumed_thenSentToDltWithoutRetry() {
        kafkaTemplate.send("orders", "order-key", "INVALID-99");

        Map<String, Object> props = KafkaTestUtils.consumerProps("dlt-test", "false", broker);
        try (Consumer<String, String> consumer = new DefaultKafkaConsumerFactory<>(
                props, new StringDeserializer(), new StringDeserializer()).createConsumer()) {

            broker.consumeFromAnEmbeddedTopic(consumer, "orders.DLT");
            ConsumerRecord<String, String> record =
                    KafkaTestUtils.getSingleRecord(consumer, "orders.DLT", Duration.ofSeconds(10));

            assertThat(record.value()).isEqualTo("INVALID-99");
            assertThat(record.key()).isEqualTo("order-key");

            String cause = new String(record.headers()
                    .lastHeader(KafkaHeaders.DLT_EXCEPTION_CAUSE_FQCN).value());
            assertThat(cause).isEqualTo(IllegalArgumentException.class.getName());

            String originalTopic = new String(record.headers()
                    .lastHeader(KafkaHeaders.DLT_ORIGINAL_TOPIC).value());
            assertThat(originalTopic).isEqualTo("orders");
        }
    }
}

Der Test funktioniert wie folgt:

• ${spring.embedded.kafka.brokers} verweist die Anwendung auf den eingebetteten Broker.
• KafkaTestUtils.consumerProps erstellt eine Consumer-Konfiguration mit auto.offset.reset=earliest, daher verpasst der Test keine Nachrichten, die vor dem Abonnement veröffentlicht wurden.
• Da IllegalArgumentException non-wiederholbar ist, erreicht die Nachricht das DLT fast sofort. Der Test muss nicht auf die gesamte Backoff-Sequenz warten.
• Die Assertions überprüfen drei Dinge: Key und Value bleiben erhalten, der Header verzeichnet die ursprüngliche Exception korrekt, und der Header verzeichnet das Quell-Topic korrekt.

Wenn du den Retry-Fluss testen möchtest, lagere die Backoff-Werte in Properties aus. Im Test kannst du die Wartezeit auf einige Dutzend Millisekunden setzen, damit der Test schnell läuft.

7. Häufige Fehler

• Exceptions nicht klassifizieren. Eine Wiederholung von NullPointerException oder Validierungsfehlern kostet nur Zeit. Deklariere klar, welche Exceptions wiederholt werden und welche direkt zum DLT gehen. Deserialisierungsfehler wie DeserializationException sind standardmäßig bereits non-wiederholbar. Damit der Handler diesen Fehler aber erhält, musst du den Deserializer mit ErrorHandlingDeserializer umhüllen.

• Serializer des KafkaTemplate passt nicht. Der Recoverer veröffentlicht den deserialiserten Value erneut. Wenn der Consumer JSON in ein Objekt liest, der Template aber StringSerializer nutzt, schlägt das Senden zum DLT fehl. Nutze einen Template mit passendem Serializer. Bei Deserialisierungsfehlern ist der Value ein byte[], daher benötigst du ByteArraySerializer.

• Listener ist nicht idempotent. Retry bedeutet, dass die gleiche Nachricht mehrfach verarbeitet werden kann. Wenn charge Geld bereits abgebucht hat und dann ein Timeout auftritt, zieht das Retry wieder Geld ab. Setze dich gegen Duplikate mit einem Idempotency-Key zur Wehr, z.B. indem du den verarbeiteten orderId speicherst.

• Niemand liest das DLT. Das DLT ist kein Müll. Du brauchst Benachrichtigungen, wenn neue Nachrichten im DLT ankommen, und einen Replay-Prozess nach der Fehlerbehebung.

• Backoff insgesamt zu lang mit Retry Blocking. Die Partition ist währenddessen blockiert. Wenn die Gesamtwartezeit mehr als einige Dutzend Sekunden überschreitet, erwäge den Wechsel zu Non-Blocking.

Hinweis: Verwende @RetryableTopic nicht für Flüsse, die strikte Reihenfolge nach Key benötigen. Umgekehrt: nutze kein Retry Blocking mit langer Backoff für Topics mit hohem Durchsatz. Eine einzelne fehlerhafte Nachricht kann den Lag um Zehntausende erhöhen.

8. Fazit

Dieser Artikel hat zwei Möglichkeiten zur Behandlung fehlgeschlagener Nachrichten in Spring Kafka gezeigt. DefaultErrorHandler mit DeadLetterPublishingRecoverer für Retry Blocking erhält die Reihenfolge, blockiert aber die Partition. @RetryableTopic mit @DltHandler für Non-Blocking-Retry blockiert den Consumer nicht, opfert aber die Reihenfolge.

Egal welche Variante du wählst, du solltest vier Dinge tun: Exceptions in wiederholbar und nicht wiederholbar klassifizieren, sicherstellen, dass der Listener idempotent ist, das DLT mit passender Partitionsanzahl erstellen und das DLT regelmäßig überwachen. Ein Test mit @EmbeddedKafka wie in Abschnitt 6 hilft dir zu überprüfen, dass die Konfiguration korrekt funktioniert, bevor du in die Produktion gehst.