Spring Kafka 的错误处理:DefaultErrorHandler 与 DLT

默认行为

实测 · 本机执行输出

  poison-1 被投递 10 次,时刻 [0.0, 0.5, 1.0, 1.5]…[4.6] 秒;ok-2 在 [5.1] 秒被处理;已提交偏移量 2(2 = 两条都已越过);Kafka 里的 topic ['t']

共 10 次投递,间隔约 0.5 秒,之后丢弃。

死信恢复器

实测 · 本机执行输出

  poison-1 被投递 3 次,时刻 [0.0, 0.5, 1.0] 秒;ok-2 在 [1.6] 秒被处理
  t-dlt 里的消息值 = poison-1;头信息的键 = ['kafka_dlt-exception-cause-fqcn', 'kafka_dlt-exception-fqcn', 'kafka_dlt-exception-message', 'kafka_dlt-exception-stacktrace', 'kafka_dlt-original-consumer-group', 'kafka_dlt-original-offset', 'kafka_dlt-original-partition', 'kafka_dlt-original-timestamp', 'kafka_dlt-original-timestamp-type', 'kafka_dlt-original-topic']
  其中 kafka_dlt-original-topic = t,kafka_dlt-original-offset 对应字节 = 8 字节,kafka_dlt-exception-fqcn = org.springframework.kafka.listener.ListenerExecutionFailedException,kafka_dlt-exception-message = Listener method 'void demo.Main$L.on(org.apache.kafka.clients.consumer.ConsumerRecord<java.lang.String, java.lang.String>)' threw exception; poison

默认发到原 topic-dlt,头里带原始位置与异常信息。

毒丸消息

实测 · 本机执行输出

  addNotRetryableExceptions(FatalEx):fatal-1 被投递 1 次;poison-2 被投递 3 次;t-dlt 里有 2 条
  值反序列化器直接用 IntegerDeserializer,发一条 "abc":监听方法被调用 0 次;6 秒内应用日志写了 20952037 行、3923 MB(一直在重复同一个反序列化错误);已提交偏移量 -1001
  换成 ErrorHandlingDeserializer 包着 IntegerDeserializer,并配了 DLT 恢复器(KafkaTemplate 的值序列化器还是默认的 StringSerializer):监听方法被调用 0 次;t-dlt 里有 0 条;已提交偏移量 -1001(-1001 表示一直没有提交);应用日志里的原因:Can't convert value of class [B to class org.apache.kafka.common.serialization.StringSerializer
  把生产者的值序列化器换成 ByteArraySerializer:监听方法被调用 0 次;t-dlt 里有 1 条(内容 b'abc',头里的键 ['kafka_dlt-exception-cause-fqcn', 'kafka_dlt-exception-fqcn', 'kafka_dlt-exception-message', 'kafka_dlt-exception-stacktrace', 'kafka_dlt-original-consumer-group', 'kafka_dlt-original-offset', 'kafka_dlt-original-partition', 'kafka_dlt-original-timestamp', 'kafka_dlt-original-timestamp-type', 'kafka_dlt-original-topic']);已提交偏移量 1

不可重试异常只投递 1 次;反序列化失败要配 ErrorHandlingDeserializer 和字节序列化器。

重试 topic

实测 · 本机执行输出

  阻塞重试(FixedBackOff 500 ms × 2 次):ok-2 在 [1.6] 秒才被处理(排在 poison-1 后面)
  只加 @EnableKafkaRetryTopic(没有 TaskScheduler):应用启动失败,原因:Either a RetryTopicSchedulerWrapper or TaskScheduler bean is required
  @RetryableTopic:Kafka 里自动出现的 topic ['t', 't-dlt', 't-retry'];ok-2 在 [0.6] 秒被处理;poison-1 的处理轨迹:0.0s t → 0.5s t-retry → 1.1s t-retry → 1.6s t-dlt

@RetryableTopic 不阻塞主 topic,需要 TaskScheduler。

分步演示

Kafka 错误处理

本步的对象

本文用到的工具与边界

本机 macOS 27 arm64 / OpenJDK 21.0.12(Homebrew) / Spring Boot 4.0.8(Spring Framework 7.0.9、Tomcat 11.0.24) / Spring Cloud 2025.1.3;用 javac 编译、java 直接运行,依赖由 Maven 从 Maven Central 解析;数字每次运行会有小幅抖动。Kafka 4.3.1 单节点(自己起的);topic t 1 个分区;每个场景之前删除并重建 topic、使用新的消费组;死信 topic 里的消息用 confluent-kafka 读取;时间是相对监听方法第一次收到消息的秒数;「日志 3.9 GB」是我把应用日志级别设为 ERROR 时跑 6 秒量到的
没有验证:默认重试次数 10 次与间隔 0.5 秒背后的配置(我只看到了行为);DeadLetterPublishingRecoverer 自定义 topic 名与分区选择;ErrorHandlingDeserializer 用在键上;批量监听的错误处理(BatchListenerFailedException);事务性生产者下的恢复器;@RetryableTopic 的多个重试 topic 命名策略(这里 3 次同间隔复用了一个 t-retry);DLT 消息的重放;重试 topic 对消息顺序的影响

下一篇 · SPRING 142

@TransactionalEventListener 做 outbox