Retry et dead letter topic avec Spring Kafka

1. Aperçu général

Un consumer Kafka rencontrera certainement des messages en erreur lors du traitement. Il y a des erreurs temporaires, par exemple une passerelle de paiement timeout. Il y a des erreurs permanentes, par exemple des données au mauvais format. Si vous ne les traitez pas correctement, vous rencontrerez l’une de ces deux situations : une perte de message ou un consumer bloqué indéfiniment sur un message qui ne réussit jamais.

Cet article explique comment configurer retry dans Spring Kafka de deux façons : en mode blocking avec DefaultErrorHandler et en mode non-blocking avec @RetryableTopic. Les messages qui restent en erreur après épuisement des tentatives de retry sont envoyés vers un dead letter topic (DLT). Après la lecture, vous saurez quand choisir l’une ou l’autre approche et comment écrire des tests pour vérifier le comportement.

2. Configuration

Ajoutez la dépendance Spring Kafka. Spring Boot 3.x gère lui-même la version, donc vous n’avez pas besoin de la déclarer :

<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>

Configuration minimale dans application.properties :

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

Boot utilise par défaut StringSerializer et StringDeserializer. Pour simplifier l’exemple, la valeur du message est l’identifiant de commande sous forme de String.

3. Consumer susceptible de défaillance

Voici un service de traitement des paiements. Un identifiant de commande mal formaté est une erreur permanente. Une erreur lors de l’appel à la passerelle de paiement est une erreur temporaire :

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("Identifiant de commande invalide : " + orderId);
        }
        // Appel de la passerelle de paiement ; peut lever PaymentGatewayException en cas de timeout
    }
}

Le listener appelle simplement le service :

@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("Traitement de la commande {}", orderId);
        paymentService.charge(orderId);
    }
}

Si vous ne configurez rien, Spring Kafka utilise DefaultErrorHandler avec FixedBackOff(0, 9). Le message d’erreur est réessayé 9 fois de suite, soit 10 fois au total. Ensuite, le handler se contente de journaliser et d’ignorer. Le message est considéré comme perdu. C’est pourquoi vous devez reconfigurer.

4. Retry blocking avec DefaultErrorHandler

4.1. Configuration du backoff et du DLT

Quand le listener lève une exception, DefaultErrorHandler cherche à nouveau l’offset du message en erreur pour que le prochain poll le relise. Après épuisement des tentatives de retry, le handler appelle un ConsumerRecordRecoverer. Ici, le recoverer est 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;
    }
}

Explication de chaque partie :

• DeadLetterPublishingRecoverer(template) envoie le message d’erreur vers le topic <nom du topic>.DLT, dans la même partition que le message original. Pour le topic orders, le DLT est orders.DLT.
• ExponentialBackOffWithMaxRetries(3) réessaie au maximum 3 fois. Les délais d’attente sont respectivement 1s, 2s, 4s et ne dépassent pas 10s.
• addNotRetryableExceptions déclare les exceptions qui ne doivent pas être réessayées. IllegalArgumentException va directement au DLT, car même en réessayant plusieurs fois, des données mal formatées resteront mal formatées.

Spring Boot attache automatiquement le bean CommonErrorHandler à la container factory par défaut. Vous n’avez pas besoin de créer vous-même ConcurrentKafkaListenerContainerFactory.

Lors de l’envoi de ORD-1 et du timeout continu de la passerelle de paiement, le log ressemblera à ceci :

10:00:00.010 INFO  Traitement de la commande ORD-1
10:00:01.020 INFO  Traitement de la commande ORD-1
10:00:03.030 INFO  Traitement de la commande ORD-1
10:00:07.040 INFO  Traitement de la commande ORD-1
10:00:07.060 ERROR Backoff ... exhausted for orders-0@42

Au total, il y a 4 tentatives de traitement : 1 tentative initiale et 3 retries. Ensuite, le message se trouve dans orders.DLT.

4.2. Headers dans le DLT

Le message dans le DLT conserve sa clé et sa valeur. Le recoverer ajoute des headers pour vous aider à enquêter sur la cause :

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 method ... threw exception
kafka_dlt-exception-stacktrace = ...

Votre exception d’origine se trouve dans kafka_dlt-exception-cause-fqcn. Si vous devez lire les headers via du code, utilisez les constantes dans KafkaHeaders, par exemple KafkaHeaders.DLT_ORIGINAL_TOPIC.

Note : Comme le recoverer envoie par défaut vers la même partition, orders.DLT doit avoir au moins autant de partitions que orders. Sinon, la publication échouera. Vous pouvez créer le DLT avec un bean NewTopic ayant le même nombre de partitions. Une autre approche consiste à transmettre un destination resolver qui retourne new TopicPartition(r.topic() + ".DLT", -1) pour que Kafka choisisse lui-même la partition.

4.3. Quand le retry blocking est approprié

Le retry blocking maintient l’ordre des messages dans la partition. Un message suivant n’est traité que si le message précédent a réussi ou a été envoyé au DLT. C’est le bon choix quand l’ordre est important, par exemple pour les événements OrderCreated → OrderPaid → OrderShipped d’une même commande.

Le coût est que toute la partition est bloquée pendant le traitement. À partir de Spring Kafka 2.9, le handler attend le backoff sans dépasser max.poll.interval.ms, grâce aux mécanismes de seek et pause. Néanmoins, si vous définissez un backoff total de plusieurs minutes, le lag augmentera considérablement.

5. Retry non-blocking avec @RetryableTopic

5.1. Configuration via annotation

Avec le retry non-blocking, les messages d’erreur sont transférés vers un retry topic séparé. Cela permet au consumer de continuer à traiter le message suivant immédiatement. Un autre consumer lit le retry topic après un délai de backoff défini :

@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("Traitement de {} depuis {}", orderId, topic);
        paymentService.charge(orderId);
    }

    @DltHandler
    public void onDlt(String orderId,
                      @Header(KafkaHeaders.EXCEPTION_MESSAGE) String error) {
        log.error("Commande {} envoyée au DLT : {}", orderId, error);
    }
}

Points clés :

• attempts = "4" signifie un total de 4 tentatives de traitement : 1 au topic original et 3 aux retry topics.
• La valeur est une String pour supporter les placeholders comme "${app.retry.attempts}".
• @Backoff provient du package org.springframework.retry.annotation. Spring Retry est inclus avec spring-kafka.
• exclude énumère les exceptions qui vont directement au DLT, équivalent à addNotRetryableExceptions.
• @DltHandler est placé dans la même classe. C’est ici que vous pouvez journaliser, envoyer une alerte ou sauvegarder dans une table pour un traitement manuel.

5.2. Topics créés automatiquement

Avec des backoff croissants, par défaut chaque retry a son propre topic avec un suffixe basé sur le délai :

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

Spring Kafka crée automatiquement ces topics au démarrage (autoCreateTopics = "true" par défaut). Les environnements de production désactivent généralement l’auto-création sur le broker. Dans ce cas, vous devez créer à l’avance les topics avec les bons noms et nombres de partitions. Si vous préférez utiliser un index à la place du délai, définissez topicSuffixingStrategy = TopicSuffixingStrategy.SUFFIX_WITH_INDEX_VALUE.

Notez bien le suffixe : @RetryableTopic utilise -dlt, tandis que DeadLetterPublishingRecoverer utilise .DLT. Gardez cela à l’esprit lors de la configuration du monitoring.

5.3. Compromis du non-blocking

Le retry non-blocking casse l’ordre. Supposez que ORD-1 échoue et est transféré au retry topic, tandis que ORD-2 réussit immédiatement. Alors ORD-2 est traité avant ORD-1. Si votre flux métier dépend de l’ordre des événements, n’utilisez pas cette approche.

De plus, le nombre de topics augmente et il y a aussi plus de consumers. En contrepartie, un unique message problématique ne bloque pas toute la partition.

6. Test du flux DLT

Le test suivant utilise @EmbeddedKafka. Il envoie un identifiant de commande mal formaté et vérifie que le message apparaît dans orders.DLT sans retry :

@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");
        }
    }
}

Le test fonctionne comme suit :

• ${spring.embedded.kafka.brokers} pointe l’application vers le broker embarqué.
• KafkaTestUtils.consumerProps crée une configuration consumer avec auto.offset.reset=earliest, donc le test ne manque aucun message publié avant que le consumer ne s’abonne.
• Puisque IllegalArgumentException n’est pas retryable, le message arrive au DLT quasi instantanément. Le test n’a pas besoin d’attendre l’exhaustion de la chaîne de backoff.
• Les assertions vérifient trois choses : la clé et la valeur sont préservées, les headers enregistrent correctement l’exception d’origine et les headers enregistrent correctement le topic source.

Si vous voulez tester le flux de retry, extrayez les valeurs de backoff dans les properties. Dans le test, vous définiriez les délais à quelques dizaines de millisecondes pour que le test s’exécute rapidement.

7. Erreurs courantes

• Ne pas catégoriser les exceptions. Faire un retry sur NullPointerException ou une erreur de validation ne fait que perdre du temps. Déclarez clairement quelles exceptions doivent être réessayées et lesquelles vont directement au DLT. Les erreurs de désérialisation comme DeserializationException sont déjà non-retryable par défaut. Cependant, pour que le handler reçoive cette erreur, vous devez envelopper le désérialisateur avec ErrorHandlingDeserializer.
• Le serializer de KafkaTemplate ne correspond pas. Le recoverer republish la valeur désérialisée. Si le consumer lit du JSON en objet mais que le template utilise StringSerializer, l’envoi au DLT échouera. Utilisez un template avec le serializer approprié. Pour les erreurs de désérialisation, la valeur est byte[], vous avez donc besoin de ByteArraySerializer.
• Le listener n’est pas idempotent. Retry signifie que le même message peut être traité plusieurs fois. Si charge a déjà débité l’argent puis timeout, un retry débiterait à nouveau. Protégez-vous contre les doublons avec une clé d’idempotence, par exemple en sauvegardant les orderId déjà traités.
• Personne ne lit le DLT. Le DLT n’est pas une poubelle. Vous avez besoin d’une alerte quand le DLT reçoit un nouveau message, et d’un processus de replay après correction de l’erreur.
• Backoff total trop long avec retry blocking. La partition reste bloquée pendant tout ce temps. Si le temps d’attente total dépasse quelques dizaines de secondes, envisagez de basculer vers le non-blocking.

Note : Ne pas utiliser @RetryableTopic pour un flux qui nécessite un ordre strict par clé. Inversement, n’utilisez pas le retry blocking avec un backoff long pour un topic à haut débit. Un seul message en erreur pourrait augmenter le lag de dizaines de milliers.

8. Conclusion

Cet article a présenté deux façons de traiter les messages en erreur dans Spring Kafka. DefaultErrorHandler combiné avec DeadLetterPublishingRecoverer pour le retry blocking, qui préserve l’ordre mais bloque la partition. @RetryableTopic avec @DltHandler pour le retry non-blocking, qui ne bloque pas le consumer mais sacrifice l’ordre.

Quelle que soit votre choix, vous devriez faire quatre choses : catégoriser les exceptions retryable et non-retryable, assurer que le listener est idempotent, créer le DLT avec un nombre approprié de partitions, et surveiller régulièrement le DLT. Un test avec @EmbeddedKafka comme à la section 6 vous aide à vous assurer que la configuration fonctionne correctement avant de passer en production.