使用事务的响应式Kafka生产者遇Kafka不可达如何恢复？

根因说明 #

你遇到的三类异常本质是两个核心问题：

• 单例共享的ReactiveKafkaProducerTemplate一旦进入事务错误状态，就无法自动恢复，会污染所有后续请求
• 原有逻辑缺少对取消信号的处理，也没有做客户端侧的兜底超时，事务状态容易卡在中间态

配置修正 #

之前调整超时不生效是因为缺少联动配置，且需要关闭错误停止开关，正确配置如下：

// 抽离公共SenderOptions为单例Bean，避免重复构造配置
@Bean
public SenderOptions<String, String> kafkaSenderOptions() {
    Map<String, Object> props = new HashMap<>();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "http://localhost:9092");
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
    // 核心超时配置，和需求20s上限对齐
    props.put(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 20000); // 事务最大超时，不能超过Broker侧transaction.max.timeout.ms默认值
    props.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 18000); // 消息投递超时，略短于事务超时，避免时序冲突
    props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 10000); // 单次请求超时
    props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 5000); // 元数据拉取、缓冲区分配的最大阻塞时间
    return SenderOptions.create(props)
            .stopOnError(false) // 错误后不停止生产者实例
            .maxInFlightRequestsPerConnection(1); // 事务场景强制要求为1
}

代码逻辑修正 #

废弃单例ReactiveKafkaProducerTemplate的用法，改为每次请求新建独立的生产者实例，从根源避免状态污染，同时增加取消、异常场景的事务回滚逻辑和客户端兜底超时：

public Mono<Void> send(String message) {
    // 每次请求生成唯一事务ID，避免ID冲突
    String txId = "producer-tx-" + UUID.randomUUID();
    SenderOptions<String, String> txOptions = kafkaSenderOptions()
            .withProducerProperty(ProducerConfig.TRANSACTIONAL_ID_CONFIG, txId);
    
    // Reactor using语法自动管理生产者生命周期，请求结束后自动销毁实例
    return Mono.using(
            // 每次请求创建独立的生产者实例
            () -> new ReactiveKafkaProducerTemplate<String, String>(txOptions),
            kafkaTemplate -> kafkaTemplate.transactionManager()
                    .begin()
                    .then(kafkaTemplate.send("my.test.topic", message))
                    .then(kafkaTemplate.transactionManager().commit())
                    // 客户端侧兜底20s超时，到点直接终止当前请求
                    .timeout(Duration.ofSeconds(20))
                    // 异常场景自动回滚事务
                    .onErrorResume(e -> kafkaTemplate.transactionManager()
                            .abort()
                            .then(Mono.error(e)))
                    // 处理用户主动取消请求的场景，强制回滚事务
                    .doOnCancel(() -> kafkaTemplate.transactionManager()
                            .abort()
                            .subscribe()),
            // 请求结束自动关闭生产者，彻底清除状态残留
            ReactiveKafkaProducerTemplate::close
    );
}

适配效果 #

• 20s内Kafka恢复的场景：请求正常重试，事务提交成功
• 超过20s未恢复的场景：主动超时回滚，销毁当前生产者，后续请求新建实例不受影响
• 用户主动取消请求的场景：事务正常回滚，不会残留IN_TRANSACTION状态

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