Spring Cloud Stream Kafka Binder异步生产者Broker宕机丢消息排查

问题场景 #

我使用Spring Cloud Stream Kafka Binder向Kafka生产消息，配置如下：

spring.cloud.stream.kafka.bindings.outputChannelName.producer.sync: false
error-channel-enabled: true
unclean.leader.election.enabled: true

已订阅errorChannel来记录发送失败的消息。发现当Broker宕机时，异步生产者会针对UNKNOWN_TOPIC_OR_PARTITION、NOT_LEADER_FOR_PARTITION、NETWORK_EXCEPTION错误在delivery.timeout.ms期限内重试，但并非所有错误都会重试。超过delivery.timeout.ms后，消息过期并进入errorChannel被记录，但仍有部分消息丢失，既未出现在成功链路也未出现在errorChannel中，即使设置Debug日志级别也未找到相关异常或错误。

问题解答 #

1. 如何追踪丢失的消息？ #

• 启用生产者元数据通道：配置spring.cloud.stream.kafka.bindings.outputChannelName.producer.record-metadata-channel，指定专属通道接收每条消息的发送回执（成功/失败状态），异步模式下也能完整捕获消息最终流向
• 自定义生产者拦截器：实现ProducerInterceptor，在onSend方法记录消息唯一ID、发送时间，在onAcknowledgement方法记录消息的确认状态，全程追踪单条消息的流转路径
• 监控生产者缓冲区状态：通过kafka.producer.buffer-available-bytes指标检查内存缓冲区是否溢出，若缓冲区满，异步发送的新消息可能被直接丢弃，无异常抛出
• 本地消息暂存校验：发送前为每条消息生成唯一ID并暂存（如内存缓存、临时表），定期对比已确认成功/失败的消息ID，定位缺失的消息条目

2. 是否遗漏了特定配置？ #

• 强制设置acks=all：配置spring.cloud.stream.kafka.bindings.outputChannelName.producer.acks=all，确保消息被所有同步副本持久化后才算发送成功，避免Broker宕机时消息未落地就被标记为成功
• 调整重试策略：设置retries=10（根据业务容错能力调整）、retry.backoff.ms=1000，保证可重试错误有足够的重试次数和间隔，减少因重试次数不足导致的消息丢失
• 开启失败路由强制触发：配置spring.cloud.stream.kafka.bindings.outputChannelName.producer.fail-on-error=true，确保异步发送失败的消息能被正确路由到errorChannel，避免内部吞掉异常
• 合理设置delivery.timeout.ms：该值必须大于retries * retry.backoff.ms，否则重试未完成就触发超时，导致消息提前进入错误通道或直接丢失
• 启用幂等性：配置spring.cloud.stream.kafka.bindings.outputChannelName.producer.enable.idempotence=true，配合acks=all使用，既防止消息重复，又确保重试过程中消息不丢失
• 检查事务配置：若使用事务，需确保spring.cloud.stream.kafka.bindings.outputChannelName.producer.transactional=true，同时配置事务超时时间，避免事务提交失败导致消息丢失

3. 需要在日志中排查哪些特定异常？ #

• *BufferExhaustedException*：生产者内存缓冲区已满，无法接收新消息，异步模式下该异常可能被内部吞掉，导致消息直接丢失
• *TimeoutException*：消息发送超时，需确认超时是否与delivery.timeout.ms配置匹配，排查是重试超时还是初始发送超时
• *ProducerFencedException*：生产者被栅栏隔离（常见于事务场景），会导致消息发送失败但可能不触发errorChannel路由
• *InterruptedException*：生产者线程被中断，可能导致消息发送流程中途终止，未触发错误回调
• *KafkaProducerClosedException*：生产者被意外关闭，此时正在发送的消息可能丢失，需排查关闭触发原因
• 搜索errorChannel相关日志：确认是否有消息被路由到错误通道但未被记录（如日志级别被过滤）
• 排查KafkaMessageChannelBinder日志：搜索该类的Debug级日志，查看消息发送时的内部处理逻辑，是否有异常被静默处理

额外建议 #

• 监控核心指标：重点监控record-send-rate、record-error-rate、buffer-available-bytes、retries-per-request等指标，通过指标波动快速定位消息丢失的时间点和诱因
• 模拟异常场景：通过测试工具模拟Broker宕机、分区Leader切换、网络抖动等场景，验证消息重试和错误路由逻辑的完整性
• 按需切换同步发送：若业务对消息可靠性要求极高，可临时将sync设为true，牺牲部分性能换取消息状态的强可追踪性

内容的提问来源于stack exchange，提问作者Ranjit Meher