Spring Cloud Stream Kafka生产者Broker宕机时无异常抛出问题及解决方案

问题描述 #

当Kafka Broker宕机时，Spring Cloud Stream（Kafka）生产者不会抛出异常。例如以下代码：

boolean sent = streamBridge.send("sometopic", deliver);
log.info("-------------> message sent? {}",sent);

即使Kafka Broker已完全宕机，sent布尔值仍会返回true，这会直接导致消息丢失（即便使用Outbox模式也无法避免）。

由于Spring Cloud Stream天生是异步非阻塞的，直接依赖返回值无法判断真实发送状态。目前使用AdminClient检查Broker可用性，但该方案存在竞态问题（检查时可用、发送时不可用），属于较为初级的实现。计划采用事务型生产者，需要明确如何在Broker宕机时触发异常，确保发送状态可感知。

附带相关日志：

2023-10-06T00:06:37.372+01:00  INFO 58262 --- [           main] o.a.kafka.common.utils.AppInfoParser     : Kafka startTimeMs: 1696547197372
2023-10-06T00:06:37.496+01:00  INFO 58262 --- [| adminclient-1] org.apache.kafka.clients.NetworkClient   : [AdminClient clientId=adminclient-1] Node -1 disconnected.
2023-10-06T00:06:37.497+01:00  INFO 58262 --- [| adminclient-1] org.apache.kafka.clients.NetworkClient   : [AdminClient clientId=adminclient-1] Cancelled in-flight API_VERSIONS request with correlation id 0 due to node -1 being disconnected (elapsed time since creation: 114ms, elapsed time since send: 114ms, request timeout: 3600000ms)
2023-10-06T00:06:37.601+01:00  INFO 58262 --- [| adminclient-1] org.apache.kafka.clients.NetworkClient   : [AdminClient clientId=adminclient-1] Node -1 disconnected.
2023-10-06T00:06:37.601+01:00  INFO 58262 --- [| adminclient-1] org.apache.kafka.clients.NetworkClient   : [AdminClient clientId=adminclient-1] Cancelled in-flight API_VERSIONS request with correlation id 1 due to node -1 being disconnected (elapsed time since creation: 2ms, elapsed time since send: 2ms, request timeout: 3600000ms)
2023-10-06T00:06:37.805+01:00  INFO 58262 --- [| adminclient-1] org.apache.kafka.clients.NetworkClient   : [AdminClient clientId=adminclient-1] Node -1 disconnected.
2023-10-06T00:06:37.805+01:00  INFO 58262 --- [| adminclient-1] org.apache.kafka.clients.NetworkClient   : [AdminClient clientId=adminclient-1] Cancelled in-flight API_VERSIONS request with correlation id 2 due to node -1 being disconnected (elapsed time since creation: 1ms, elapsed time since send: 1ms, request timeout: 3600000ms)
2023-10-06T00:06:38.009+01:00  INFO 58262 --- [| adminclient-1] org.apache.kafka.clients.NetworkClient   : [AdminClient clientId=adminclient-1] Node -1 disconnected.
2023-10-06T00:06:38.009+01:00  INFO 58262 --- [| adminclient-1] org.apache.kafka.clients.NetworkClient   : [AdminClient clientId=adminclient-1] Cancelled in-flight API_VERSIONS request with correlation id 3 due to node -1 being disconnected (elapsed time since creation: 1ms, elapsed time since send: 1ms, request timeout: 3600000ms)
2023-10-06T00:06:38.418+01:00  INFO 58262 --- [| adminclient-1] org.apache.kafka.clients.NetworkClient   : [AdminClient clientId=adminclient-1] Node -1 disconnected.
2023-10-06T00:06:38.418+01:00  INFO 58262 --- [| adminclient-1] org.apache.kafka.clients.NetworkClient   : [AdminClient clientId=adminclient-1] Cancelled in-flight API_VERSIONS request with correlation id 4 due to node -1 being disconnected (elapsed time since creation: 2ms, elapsed time since send: 2ms, request timeout: 3600000ms)
2023-10-06T00:06:39.129+01:00  INFO 58262 --- [| adminclient-1] org.apache.kafka.clients.NetworkClient   : [AdminClient clientId=adminclient-1] Node -1 disconnected.
2023-10-06T00:06:39.129+01:00  INFO 58262 --- [| adminclient-1] org.apache.kafka.clients.NetworkClient   : [AdminClient clientId=adminclient-1] Cancelled in-flight API_VERSIONS request with correlation id 5 due to node -1 being disconnected (elapsed time since creation: 2ms, elapsed time since send: 2ms, request timeout: 3600000ms)
2023-10-06T00:06:40.246+01:00  INFO 58262 --- [| adminclient-1] org.apache.kafka.clients.NetworkClient   : [AdminClient clientId=adminclient-1] Node -1 disconnected.
2023-10-06T00:06:40.246+01:00  INFO 58262 --- [| adminclient-1] org.apache.kafka.clients.NetworkClient   : [AdminClient clientId=adminclient-1] Cancelled in-flight API_VERSIONS request with correlation id 6 due to node -1 being disconnected (elapsed time since creation: 1ms, elapsed time since send: 1ms, request timeout: 3600000ms)

解决方案 #

1. 配置同步发送与强确认机制 #

Spring Cloud Stream默认异步发送，streamBridge.send返回true仅代表消息成功进入本地生产者队列，而非Broker已确认接收。开启同步发送+强确认可直接感知Broker状态：

在application.yml中添加配置：

spring:
  cloud:
    stream:
      kafka:
        binder:
          producers:
            sync: true          # 开启同步发送，阻塞等待Broker响应
            acks: all           # 要求所有同步副本确认接收
            retries: 0          # 关闭重试，避免超时前无限等待
            delivery-timeout-ms: 5000  # 设置发送超时时间，超时则抛出异常

配置后，若Broker宕机无法响应，streamBridge.send会在超时后抛出KafkaProducerException，此时返回值能准确反映真实发送状态。

2. 事务型生产者配置 #

结合事务机制，可确保发送动作与本地事务绑定，Broker宕机时事务提交失败会触发异常：

第一步：添加事务配置

spring:
  cloud:
    stream:
      kafka:
        binder:
          transaction:
            transaction-id-prefix: tx-  # 必须配置事务ID前缀，用于生成唯一事务ID
          producers:
            sync: true
            acks: all
            retries: 0
            delivery-timeout-ms: 5000

第二步：事务包裹发送逻辑

使用@Transactional注解包裹发送代码，Broker宕机时事务提交会失败并抛出TransactionSystemException：

@Transactional
public void sendTransactionalMessage(String topic, Object payload) {
    boolean sent = streamBridge.send(topic, payload);
    if (!sent) {
        throw new RuntimeException("消息发送失败，事务回滚");
    }
}

3. 异步监听生产者错误事件 #

若不想阻塞主线程，可通过ProducerListener异步监听发送错误，捕获Broker不可用场景：

第一步：实现错误监听器

@Component
public class KafkaProducerErrorListener implements ProducerListener<Object, Object> {

    private static final Logger log = LoggerFactory.getLogger(KafkaProducerErrorListener.class);

    @Override
    public void onSendError(Message<?> message, String destination, Exception exception) {
        log.error("发送消息到{}失败: {}", destination, exception.getMessage(), exception);
        // 自定义处理逻辑：如告警、持久化消息到本地存储、触发重试等
        throw new RuntimeException("消息发送失败", exception);
    }
}

第二步：启用错误通道

spring:
  cloud:
    stream:
      kafka:
        binder:
          producers:
            error-channel-enabled: true  # 启用错误通道，触发监听器逻辑

4. 为什么不推荐AdminClient检查 #

AdminClient的Broker可用性检查是瞬时状态，存在检查时可用、发送时不可用的竞态条件，无法保证发送动作的可靠性。上述基于生产者自身的确认机制和事务配置，能直接感知发送时刻的Broker状态，可靠性更高。

内容的提问来源于stack exchange，提问作者Pascoal Eddy Bayonne