如何在Spring Boot中用事务管理器实现Kafka事务的提交与回滚？

需要实现一个Kafka生产者方法，基于最新版Spring Boot，要求：

• 批量发送多条Kafka记录时，全部成功则提交事务，任意一条失败则回滚所有记录
• 计划采用Spring Kafka的事务方案，但缺少示例代码

现有代码如下：

自定义ProducerListener接口 #

public interface ProducerListener<K, V> {

    void onSuccess(ProducerRecord<K, V> producerRecord, RecordMetadata recordMetadata);

    void onError(ProducerRecord<K, V> producerRecord, RecordMetadata recordMetadata,
            Exception exception);

}

待完善的produceData方法 #

public ProducerResult produceData(String topic, List<Data> data) {

    data.forEach(
            d -> {
                final ProducerRecord<String, Data> record = createRecord(topic, d);
                CompletableFuture<SendResult<Integer, Data>> future = template.send(record);

                future.whenComplete((result, ex) -> {
                    if (ex != null) {
                        // 需要回滚事务
                        return new ProducerResult(false);
                    }
                });
            }
    );
    // 需要提交事务
    return new ProducerResult(true);
}

其中createRecord(topic, d)方法返回new ProducerRecord<>(topic, d)，请问如何结合KafkaTemplate.executeInTransaction实现需求？

1. 完成Spring Boot Kafka事务基础配置 #

在application.yml（或application.properties）中添加事务前缀配置，Spring Boot会自动创建事务管理器：

spring:
  kafka:
    producer:
      transaction-id-prefix: tx-kafka-  # 必须配置，用于生成唯一事务ID
      bootstrap-servers: your-kafka-broker:9092
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: com.yourpackage.DataSerializer  # 替换为你的Data序列化器

2. 重构produceData方法，结合executeInTransaction实现事务控制 #

executeInTransaction会自动封装事务逻辑：回调内代码正常执行则提交事务，抛出异常则自动回滚所有已发送记录。需要注意：

• 原异步发送逻辑会导致事务提前提交，需改为同步等待所有发送完成
• 统一处理异常，让异常抛出触发事务回滚

重构后的代码：

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;

public ProducerResult produceData(String topic, List<Data> data) {
    try {
        // 用executeInTransaction包裹所有发送逻辑
        template.executeInTransaction(operations -> {
            // 批量发送并收集所有异步结果
            var futures = data.stream()
                    .map(d -> {
                        ProducerRecord<String, Data> record = createRecord(topic, d);
                        return operations.send(record);
                    })
                    .collect(Collectors.toList());

            // 同步等待所有发送完成，确保所有消息都被处理
            for (CompletableFuture<SendResult<String, Data>> future : futures) {
                // get()会抛出异常，一旦有失败就触发事务回滚
                future.get();
            }
            return null; // executeInTransaction允许返回任意值，此处无需返回结果
        });
        return new ProducerResult(true);
    } catch (Exception e) {
        // 捕获所有异常，返回失败结果
        return new ProducerResult(false);
    }
}

3. 结合自定义ProducerListener（可选） #

如果需要监听单条消息的成功/失败细节，可以给KafkaTemplate配置你的ProducerListener实现：

// 在配置类中注入
@Bean
public KafkaTemplate<String, Data> kafkaTemplate(ProducerFactory<String, Data> producerFactory) {
    KafkaTemplate<String, Data> template = new KafkaTemplate<>(producerFactory);
    template.setProducerListener(new YourProducerListenerImpl()); // 替换为你的实现类
    return template;
}

注意：ProducerListener的onError不会触发事务回滚，事务回滚完全由executeInTransaction回调内的异常决定，Listener仅用于日志或监控。

关键逻辑说明 #

• executeInTransaction会绑定当前线程到一个Kafka事务，所有通过operations.send发送的消息都会加入该事务
• 只要回调内抛出任何异常，Spring Kafka会自动发送回滚指令，Kafka broker会丢弃所有未提交的事务内消息
• 必须同步等待所有send的Future完成，否则事务会在异步操作未完成时提前提交，导致部分消息丢失或事务失效

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