Spring Cloud Stream中DB+Kafka生产者事务同步问题咨询

环境版本 #

• spring-kafka 2.8.11
• spring-boot 2.7.7
• spring-cloud 2021.0.5

问题现象 #

采用基于Spring Cloud Stream的事件驱动分布式架构，生产者与消费者微服务分离。需求为生产者先执行数据库增改操作，再向Kafka发送消息，但当前仅数据库事务生效：发生错误时数据库事务回滚，Kafka消息却仍被发送并被消费者消费。

已在Spring Boot启动类添加@EnableTransactionManagement注解开启事务，尝试过@Transactional等方案实现生产者事务，但均无效。测试时在发送Kafka消息后手动抛出RuntimeException，问题复现。

现有代码与配置 #

生产者示例代码 #

@Autowired
private final StreamBridge streamBridge;

@Transactional
public void sendDbAndKafkaUpdate() {
    // 数据库写入操作...
    
    // 发送Kafka消息
    sendKafkaMessage();
}

private void sendKafkaMessage() {
    streamBridge.send("topic-name", messageEvent);

    // 手动抛出RuntimeException测试
    throw new RuntimeException();
}

生产者事务配置（application.yaml） #

spring:
  cloud:
    stream:
      kafka:
        binder:
          transaction:
            transaction-id-prefix: ${kafka.unique.tx.id.per.instance}  # 每个服务实例配置唯一值
            producer:
              configuration:
                retries: 1
                acks: all
    
                key.serializer: org.apache.kafka.common.serialization.StringSerializer
                value.serializer: io.confluent.kafka.serializers.protobuf.KafkaProtobufSerializer
                schema.registry.url: ${kafka.schema.registry.url}

尝试过的官方配置 #

参考官方文档的生产者事务章节配置事务管理器，但无效：抛出异常后数据库回滚，Kafka消息依然发送成功。

@Bean
public PlatformTransactionManager transactionManager(BinderFactory binders,
        @Value("${kafka.unique.tx.id.per.instance}") String txId) {

    ProducerFactory<byte[], byte[]> pf = ((KafkaMessageChannelBinder) binders.getBinder(null,
            MessageChannel.class)).getTransactionalProducerFactory();
    KafkaTransactionManager tm = new KafkaTransactionManager<>(pf);
    tm.setTransactionId(txId);
    return tm;
}

疑问 #

1. 使用StreamBridge发送消息时，binder名称应如何设置？若仅使用Apache Kafka Binder，传入null是否可行？（当前未配置output binding）
2. 如何实现数据库更新与Kafka消息发送的生产者事务同步？需考虑：
   • 官方文档建议使用ChainedTransactionManager，但该类已被废弃；
   • 未直接使用KafkaTemplate（基于Spring Cloud Stream抽象）。

解决方案 #

不在消费者绑定或默认级别配置isolation.level，而是在Kafka Binder配置级别设置：

spring.cloud.stream.kafka.binder.configuration.isolation.level: read_committed

注：文档中有时将值写为"read-committed"，但此配置对当前场景无效。

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