Kafka retry 和 dead letter topic 配合 Spring Kafka

1. 概述

Kafka consumer 肯定会遇到消息处理失败的情况。有临时错误，比如支付网关超时。也有永久错误，比如数据格式不正确。如果处理不当，你会遭遇两种情况之一：消息丢失，或者 consumer 永远卡在某条消息上，无法继续。

本文介绍如何在 Spring Kafka 中配置重试，分为两种方式：使用 DefaultErrorHandler 的 blocking 重试和使用 @RetryableTopic 的 non-blocking 重试。重试次数耗尽后仍然失败的消息将被推送到 dead letter topic（DLT）。读完本文，你将了解何时选择哪种方式，以及如何编写测试来验证重试行为。

2. 设置

添加 Spring Kafka 依赖。Spring Boot 3.x 自动管理版本，所以你不需要声明版本号：

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

在 application.properties 中的最小配置：

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

Boot 默认使用 StringSerializer 和 StringDeserializer。为了示例简洁，消息的值是字符串格式的订单 ID。

3. 可能失败的 Consumer

这是一个处理支付的服务。格式错误的订单 ID 是永久错误。调用支付网关的错误是临时错误：

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：" + orderId);
        }
        // 调用支付网关；超时时可能抛出 PaymentGatewayException
    }
}

Listener 只是调用这个服务：

@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("处理订单 {}", orderId);
        paymentService.charge(orderId);
    }
}

如果你不配置任何东西，Spring Kafka 使用 DefaultErrorHandler，配有 FixedBackOff(0, 9)。失败的消息被连续重试 9 次，总共 10 次。之后 handler 只记录日志并忽略消息。消息被视为丢失。这就是你需要重新配置的原因。

4. 使用 DefaultErrorHandler 进行 Blocking 重试

4.1. 配置退避和 DLT

当 listener 抛出异常时，DefaultErrorHandler 会查找失败消息的 offset，以便下次 poll 时重新读取它。重试次数耗尽后，handler 调用 ConsumerRecordRecoverer。这里的 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;
    }
}

逐部分解释：

• DeadLetterPublishingRecoverer(template) 将失败的消息发送到话题 <原话题名>.DLT，在相同的分区中。对于话题 orders，DLT 是 orders.DLT。
• ExponentialBackOffWithMaxRetries(3) 最多重试 3 次。等待间隔分别是 1 秒、2 秒、4 秒，不超过 10 秒。
• addNotRetryableExceptions 声明无需重试的异常。IllegalArgumentException 直接进入 DLT，因为无论重试多少次，数据格式错误都是错的。

Spring Boot 自动将 CommonErrorHandler bean 附加到默认的 container factory。你不需要自己创建 ConcurrentKafkaListenerContainerFactory。

当发送 ORD-1 并且支付网关持续超时时，日志会如下所示：

10:00:00.010 INFO  处理订单 ORD-1
10:00:01.020 INFO  处理订单 ORD-1
10:00:03.030 INFO  处理订单 ORD-1
10:00:07.040 INFO  处理订单 ORD-1
10:00:07.060 ERROR Backoff ... exhausted for orders-0@42

总共有 4 次处理：1 次初始处理和 3 次重试。之后消息位于 orders.DLT。

4.2. DLT 中的 Header

DLT 中的消息保留原始的 key 和 value。Recoverer 添加了 header 供你调查原因：

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

你的原始异常位于 kafka_dlt-exception-cause-fqcn。如果需要用代码读取 header，你可以使用 KafkaHeaders 中的常量，例如 KafkaHeaders.DLT_ORIGINAL_TOPIC。

注意： 由于 recoverer 默认发送到同一分区，orders.DLT 必须有至少与 orders 相同数量的分区。如果没有，发布操作将失败。你可以通过 NewTopic bean 创建 DLT，使其拥有相同数量的分区。另一种方法是传递目标解析器，返回 new TopicPartition(r.topic() + ".DLT", -1) 让 Kafka 自动选择分区。

4.3. 何时 Blocking 重试合适

Blocking 重试保持消息顺序。只有当前一条消息成功或已进入 DLT 时，下一条消息才会被处理。这是当顺序很重要时的正确选择，例如同一订单的事件序列 OrderCreated → OrderPaid → OrderShipped。

代价是整个分区在等待期间被阻塞。从 Spring Kafka 2.9 开始，handler 在等待退避时不会超过 max.poll.interval.ms，这得益于 seek 和 pause 机制。尽管如此，如果你设置总退避时间为几分钟，lag 将急剧上升。

5. 使用 @RetryableTopic 进行 Non-Blocking 重试

5.1. 通过注解配置

使用 non-blocking 重试时，失败的消息被转移到一个单独的重试话题。因此，consumer 可以立即处理下一条消息。另一个 consumer 在定义的退避间隔后读取重试话题：

@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("处理 {} 来自 {}", orderId, topic);
        paymentService.charge(orderId);
    }

    @DltHandler
    public void onDlt(String orderId,
                      @Header(KafkaHeaders.EXCEPTION_MESSAGE) String error) {
        log.error("订单 {} 进入 DLT：{}", orderId, error);
    }
}

关键点：

• attempts = "4" 表示总共 4 次处理：1 次在原始话题中，3 次在重试话题中。值是 String 以支持占位符，如 "${app.retry.attempts}"。
• @Backoff 来自 org.springframework.retry.annotation 包。Spring Retry 已经包含在 spring-kafka 中。
• exclude 列出直接进入 DLT 的异常，等同于 addNotRetryableExceptions。
• @DltHandler 在同一类中。这是你记录日志、发送警报或保存到表中以进行手动处理的地方。

5.2. 创建的话题

对于递增的退避，默认情况下每个重试都有一个单独的话题，使用延迟作为后缀：

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

Spring Kafka 在启动时自动创建这些话题（autoCreateTopics = "true" 是默认值）。生产环境通常在 broker 上禁用自动创建。在这种情况下，你需要提前创建具有正确名称和分区数的话题。如果你想使用索引而不是延迟，请设置 topicSuffixingStrategy = TopicSuffixingStrategy.SUFFIX_WITH_INDEX_VALUE。

请记住后缀：@RetryableTopic 使用 -dlt，而 DeadLetterPublishingRecoverer 使用 .DLT。配置监控时要记住这一点。

5.3. Non-Blocking 的权衡

Non-Blocking 重试破坏顺序。假设 ORD-1 失败并转移到重试话题，而 ORD-2 立即成功。那么 ORD-2 会在 ORD-1 之前被处理。如果你的业务流程依赖于事件顺序，请不要使用这种方法。

此外，话题数量增加，consumer 数量也增加。作为交换，一条“坏”消息不会阻塞整个 consumer。

6. 测试 DLT 流程

下面的测试使用 @EmbeddedKafka。它发送一个格式不正确的订单 ID，并验证该消息出现在 orders.DLT 中且无需重试：

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

测试的工作原理如下：

• ${spring.embedded.kafka.brokers} 指向应用的嵌入式 broker。
• KafkaTestUtils.consumerProps 创建消费者配置，带有 auto.offset.reset=earliest，所以测试不会错过在消费者订阅之前发布的消息。
• 由于 IllegalArgumentException 是 non-retryable，消息几乎立即进入 DLT。测试无需等待整个退避序列。
• 断言检查三件事：key 和 value 保持不变，header 记录正确的原始异常，header 记录正确的源话题。

如果你想测试重试流程，请将退避值提取到 properties 中。在测试中，你可以设置延迟为几十毫秒以快速运行。

7. 常见错误

• 不区分异常类型。 重试 NullPointerException 或验证错误只是浪费时间。清楚地声明哪些异常应该重试，哪些应该直接进入 DLT。反序列化错误如 DeserializationException 默认已经是 non-retryable。但为了让 handler 收到这个错误，你需要用 ErrorHandlingDeserializer 包装反序列化器。
• KafkaTemplate 的序列化器不匹配。 Recoverer 重新发布已反序列化的 value。如果 consumer 将 JSON 读取为对象，而 template 使用 StringSerializer，发送到 DLT 会失败。请使用具有匹配序列化器的 template。对于反序列化错误的消息，value 是 byte[]，所以需要 ByteArraySerializer。
• Listener 不是幂等的。 重试意味着同一消息可能被处理多次。如果 charge 已经扣款然后才超时，重试会再次扣款。使用幂等性密钥进行去重，例如保存已处理的 orderId。
• 没人读 DLT。 DLT 不是垃圾箱。你需要在 DLT 有新消息时发送警报，并在修复错误后有一个重放流程。
• Blocking 重试的退避时间太长。 分区在此期间被阻塞。如果总等待时间超过几十秒，请考虑切换到 non-blocking。

注意： 不要对需要严格按 key 维护顺序的流程使用 @RetryableTopic。反之，对于吞吐量高的话题，不要使用长退避的 blocking 重试。一条失败的消息可能导致 lag 增加数万。

8. 结论

本文介绍了在 Spring Kafka 中处理失败消息的两种方式。DefaultErrorHandler 结合 DeadLetterPublishingRecoverer 进行 blocking 重试，保持顺序但阻塞分区。@RetryableTopic 与 @DltHandler 进行 non-blocking 重试，不会让 consumer 停滞但牺牲顺序。

无论选择哪种方式，你应该做四件事：区分可重试和不可重试的异常，确保 listener 是幂等的，使用适当分区数创建 DLT，并定期监控 DLT。第 6 节中使用 @EmbeddedKafka 的测试可以帮助你在上线前确认配置是否正确工作。