Kafka retry e dead letter topic com Spring Kafka

1. Visão geral

Um consumer Kafka certamente encontrará mensagens com erro no processamento. Existem erros temporários, como timeout de gateway de pagamento. Existem erros permanentes, como dados com formato inválido. Se não tratado corretamente, você enfrentará uma de duas situações: perder a mensagem ou o consumer ficar travado em uma mensagem que nunca terá sucesso.

Este artigo orienta você sobre como configurar retry no Spring Kafka de duas formas: blocking com DefaultErrorHandler e non-blocking com @RetryableTopic. Mensagens que continuam falhando após esgotar as tentativas de retry são encaminhadas para o dead letter topic (DLT). Ao finalizar, você saberá quando escolher cada abordagem e como escrever testes para verificar o comportamento.

2. Configuração

Adicione a dependência Spring Kafka. O Spring Boot 3.x gerencia automaticamente a versão, então você não precisa declará-la:

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

Configuração mínima em application.properties:

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

Boot usa StringSerializer e StringDeserializer por padrão. Para simplificar o exemplo, o valor da mensagem é o ID da ordem em formato String.

3. Consumer Propenso a Erros

Este é um serviço de processamento de pagamento. ID de ordem com formato inválido é um erro permanente. Erro ao chamar o gateway de pagamento é um erro temporário:

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("ID da ordem inválida: " + orderId);
        }
        // Chamar o gateway de pagamento; pode lançar PaymentGatewayException quando timeout
    }
}

O listener apenas chama o serviço:

@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("Processando ordem {}", orderId);
        paymentService.charge(orderId);
    }
}

Se você não configurar nada, Spring Kafka usa DefaultErrorHandler com FixedBackOff(0, 9). A mensagem com erro é retentada 9 vezes consecutivas, num total de 10 tentativas. Depois disso, o handler apenas registra em log e descarta. A mensagem é perdida. Por isso você precisa reconfigurar.

4. Retry Blocking com DefaultErrorHandler

4.1. Configurar Backoff e DLT

Quando o listener lança uma exception, DefaultErrorHandler faz seek para o offset da mensagem com erro para que a próxima sondagem a leia novamente. Após esgotar as tentativas de retry, o handler chama um ConsumerRecordRecoverer. Aqui, o 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;
    }
}

Explicação de cada parte:

• DeadLetterPublishingRecoverer(template) envia a mensagem com erro para o topic <nome-do-topic>.DLT, na mesma partition da mensagem original. Para o topic orders, o DLT é orders.DLT.
• ExponentialBackOffWithMaxRetries(3) faz retry no máximo 3 vezes. Os intervalos de espera são 1s, 2s, 4s e não excedem 10s.
• addNotRetryableExceptions declara as exceptions que não devem ser retentadas. IllegalArgumentException vai direto para o DLT, pois tentar novamente não ajudará se os dados estão mal formatados.

Spring Boot automaticamente atribui o bean CommonErrorHandler à factory padrão do container. Você não precisa criar manualmente ConcurrentKafkaListenerContainerFactory.

Quando envia ORD-1 e o gateway de pagamento sempre faz timeout, o log parecerá assim:

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

No total, há 4 processamentos: 1 inicial e 3 retries. Depois disso, a mensagem fica em orders.DLT.

4.2. Headers no DLT

A mensagem no DLT mantém sua key e value originais. O recoverer adiciona headers para você investigar a causa:

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

Sua exception original está em kafka_dlt-exception-cause-fqcn. Se precisar ler headers via código, use as constantes em KafkaHeaders, por exemplo KafkaHeaders.DLT_ORIGINAL_TOPIC.

Nota: Como o recoverer padrão envia para a mesma partition, orders.DLT deve ter pelo menos o mesmo número de partitions que orders. Se não, a publicação falhará. Você pode criar o DLT usando um bean NewTopic com o mesmo número de partitions. Outra forma é passar um destination resolver que retorna new TopicPartition(r.topic() + ".DLT", -1) para deixar o Kafka escolher a partition automaticamente.

4.3. Quando o Retry Blocking é Apropriado

Retry blocking mantém a ordem das mensagens na partition. Uma mensagem posterior é processada apenas quando a anterior foi bem-sucedida ou encaminhada para o DLT. Esta é a escolha correta quando a ordem é importante, por exemplo, os eventos OrderCreated → OrderPaid → OrderShipped do mesmo pedido.

O preço a pagar é que toda a partition fica bloqueada durante a espera. A partir do Spring Kafka 2.9, o handler aguarda o backoff sem exceder max.poll.interval.ms, graças aos mecanismos de seek e pause. Mesmo assim, se você definir o backoff total por alguns minutos, o lag aumentará significativamente.

5. Retry Non-Blocking com @RetryableTopic

5.1. Configuração com Annotation

Com retry non-blocking, mensagens com erro são encaminhadas para um retry topic separado. Assim, o consumer continua processando a próxima mensagem imediatamente. Um outro consumer lê o retry topic após o intervalo de backoff definido:

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

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

Os pontos principais:

• attempts = "4" significa um total de 4 processamentos: 1 no topic original e 3 no retry topic. O valor é uma String para suportar placeholders como "${app.retry.attempts}".
• @Backoff é do package org.springframework.retry.annotation. Spring Retry já vem com spring-kafka.
• exclude lista as exceptions que vão direto para o DLT, equivalente a addNotRetryableExceptions.
• @DltHandler é colocado na mesma classe. Este é o lugar para registrar em log, enviar alertas ou salvar em uma tabela para processamento manual.

5.2. Topics Criados

Com backoff crescente, por padrão cada tentativa de retry tem um topic separado, com sufixo baseado no intervalo de delay:

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

Spring Kafka cria automaticamente esses topics na inicialização (autoCreateTopics = "true" é o padrão). Ambientes de produção normalmente desabilitam auto-create no broker. Nesses casos, você precisa criar antecipadamente os topics com os nomes e números de partitions corretos. Se deseja usar um índice em vez do delay, defina topicSuffixingStrategy = TopicSuffixingStrategy.SUFFIX_WITH_INDEX_VALUE.

Atenção com os sufixos: @RetryableTopic usa -dlt, enquanto DeadLetterPublishingRecoverer usa .DLT. Lembre-se disso ao configurar monitoramento.

5.3. Compensações do Non-Blocking

Retry non-blocking quebra a ordem. Suponha que ORD-1 falha e é encaminhado para o retry topic, enquanto ORD-2 tem sucesso imediatamente. Nesse caso, ORD-2 é processado antes de ORD-1. Se o fluxo de negócio depender da ordem dos eventos, não use essa abordagem.

Além disso, o número de topics aumenta e também há mais consumers. Em compensação, uma mensagem problemática não bloqueia toda a partition.

6. Testando o Fluxo DLT

O teste abaixo usa @EmbeddedKafka. O teste envia um ID de ordem com formato inválido e verifica se a mensagem aparece em orders.DLT sem ser retentada:

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

O teste funciona da seguinte forma:

• ${spring.embedded.kafka.brokers} aponta a aplicação para o broker incorporado.
• KafkaTestUtils.consumerProps cria configuração do consumer com auto.offset.reset=earliest, então o teste não perde mensagens publicadas antes de o consumer se inscrever.
• Como IllegalArgumentException é non-retryable, a mensagem vai para o DLT quase imediatamente. O teste não precisa esperar toda a sequência de backoff.
• As asserções verificam três coisas: a key e value são preservadas, o header registra corretamente a exception original, e o header registra corretamente o topic de origem.

Se desejar testar o fluxo de retry, mova o valor de backoff para properties. No teste, defina o intervalo de espera para alguns milissegundos para o teste executar rapidamente.

7. Erros Comuns

• Não classificar exceptions. Fazer retry de NullPointerException ou erros de validação apenas desperdiça tempo. Declare claramente qual exception é retentada e qual vai direto para o DLT. Erros de desserialização como DeserializationException já são non-retryable por padrão. Mas para o handler receber esse erro, você precisa envolver o deserializer com ErrorHandlingDeserializer.
• Serializer de KafkaTemplate não corresponde. O recoverer publica novamente o value já desserializado. Se o consumer lê JSON como objeto mas o template usa StringSerializer, enviar para o DLT falhará. Use um template com um serializer apropriado. Para mensagens com erro de desserialização, o value é byte[], então você precisa de ByteArraySerializer.
• Listener não é idempotente. Retry significa que a mesma mensagem pode ser processada várias vezes. Se charge já debitou e depois fez timeout, o retry debitará novamente. Defenda-se contra duplicatas usando uma idempotency key, por exemplo salvando o orderId já processado.
• Ninguém lê o DLT. DLT não é uma lixeira. Você precisa alertar quando há novas mensagens no DLT e precisar de um processo de replay após corrigir o erro.
• Backoff total muito longo com retry blocking. A partition fica bloqueada durante esse tempo. Se o tempo total de espera exceder dezenas de segundos, considere mudar para non-blocking.

Nota: Não use @RetryableTopic para fluxos que exigem ordem rigorosa por key. Por outro lado, não use retry blocking com backoff longo para topics com alto throughput. Uma mensagem com erro pode fazer o lag aumentar em dezenas de milhares.

8. Conclusão

Este artigo apresentou duas abordagens para tratamento de mensagens com erro no Spring Kafka. DefaultErrorHandler combinado com DeadLetterPublishingRecoverer para retry blocking, preserva ordem mas bloqueia a partition. @RetryableTopic junto com @DltHandler para retry non-blocking, não bloqueia o consumer mas sacrifica a ordem.

Qualquer que seja sua escolha, você deve fazer quatro coisas: classificar exceptions retentáveis e non-retentáveis, garantir que o listener seja idempotente, criar DLT com um número apropriado de partitions e monitorar o DLT regularmente. Um teste com @EmbeddedKafka como na seção 6 ajuda você a ter certeza de que a configuração funciona corretamente antes de ir para produção.