Spring Cloud Stream 4.1：StreamBridge发送Kafka消息失败异常处理问题

问题场景 #

使用Spring Cloud Stream 4.1版本，通过StreamBridge发送消息到Kafka：

streamBridge.send("myTopic", "payload");

期望Kafka宕机时该方法抛出异常，但实际始终返回true。尝试以下配置未达到预期：

cloud:
  stream:
    kafka:
      binder:
        brokers: localhost:54442
        producer-properties:
          ack: all # 仅配置此项时，Kafka宕机线程会无限阻塞
          delivery.timeout.ms: 10000 # 配置此项时，streamBridge.send()总是抛出异常

解决方案 #

1. 核心原理说明 #

StreamBridge默认异步发送，返回的true仅代表消息成功进入生产者本地缓存，不代表Kafka已确认接收。要实现“Kafka不可用时抛出异常”，必须开启同步发送模式，让方法阻塞等待Kafka的ACK结果。

2. 正确配置组合 #

需要结合Spring Cloud Stream生产者同步属性与Kafka原生配置，调整如下：

cloud:
  stream:
    kafka:
      binder:
        brokers: localhost:54442
        producer-properties:
          acks: all
          delivery.timeout.ms: 10000
          request.timeout.ms: 8000  # 必须小于delivery.timeout.ms
          retries: 1                 # 避免无意义的无限重试
    bindings:
      myTopic-out-0:  # 绑定名称格式为{topic}-out-{index}，需与发送目标匹配
        producer:
          sync: true       # 开启同步发送，send方法将等待ACK结果
          fail-fast: true  # 遇到异常直接抛出，不静默处理

3. 代码调整提示 #

调用StreamBridge时，建议指定绑定名称而非直接用topic名（若绑定名称与topic一致可忽略）：

// 使用绑定名称发送，确保配置生效
streamBridge.send("myTopic-out-0", "payload");

4. 关键配置详解 #

• sync: true：强制生产者同步发送，streamBridge.send()不再直接返回true，而是阻塞等待Kafka的ACK或超时。
• acks: all：要求所有ISR副本确认接收，保证消息可靠性的同时，在Kafka不可用时能触发超时异常。
• delivery.timeout.ms：消息从发送到最终判定失败的总时长，需大于request.timeout.ms + retries * request.timeout.ms。
• request.timeout.ms：单次请求的超时时间，需小于delivery.timeout.ms，避免提前触发总超时。
• retries: 1：设置合理重试次数，应对Kafka短暂不可用场景，同时防止无限阻塞。

验证效果 #

配置完成后，当Kafka宕机时：

1. 生产者尝试发送消息，等待ACK超时
2. 触发1次重试（若配置retries>0）
3. 总时长超过delivery.timeout.ms后，streamBridge.send()会抛出KafkaProducerException异常，符合预期。

内容的提问来源于stack exchange，提问作者Half Blood Prince