Spring Kafka生产者事务未按文档同步，无事务日志记录求助

我正在使用Spring Boot 3.3和Spring Kafka 3.2，想要测试Spring文档中的事务同步功能。

生产者配置 #

@Configuration
public class KafkaProducerConfig {

    @Value(value = "${spring.kafka.bootstrap-servers}")
    private String bootstrapAddress;

    @Bean
    public ProducerFactory<Long, PersonDto> personProducerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(
                ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
                bootstrapAddress);
        configProps.put(
                ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                LongSerializer.class);
        configProps.put(
                ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                JsonSerializer.class);
        configProps.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "tx-person");
        return new DefaultKafkaProducerFactory<>(configProps);
    }

    @Bean
    public KafkaTemplate<Long, PersonDto> personKafkaTemplate() {
        return new KafkaTemplate<>(personProducerFactory());
    }

    @Bean
    public KafkaTransactionManager personKafkaTransactionManager() {
        KafkaTransactionManager manager = new KafkaTransactionManager(personProducerFactory());
        return manager;
    }

 
    @Bean
    @Primary
    public JpaTransactionManager transactionManager(EntityManagerFactory entityManagerFactory) {
        return new JpaTransactionManager(entityManagerFactory);
    }
}

如上述代码所示，生产者已配置为事务型。

服务代码 #

@Service
public class PersonKafkaService {

    @Autowired
    private PersonJpaService personJpaService;

    @Autowired
    private KafkaTemplate<Long, PersonDto> personKafkaTemplate;

    @Transactional
    public void processPersonDetails(PersonDetailsDto personDetails) {
        PersonDto person = personJpaService.save(personDetails);
        personKafkaTemplate.send(OUT_TOPIC, person.id(), person);
    }
}

日志配置（application.yml） #

logging:
  level.root: error
  level:
    org.apache.kafka: ERROR
    org.springframework.orm.jpa.JpaTransactionManager: trace
    org.springframework.kafka.transaction: trace

日志输出 #

2024-06-14T14:03:14.595+03:00 DEBUG 30914 --- [nio-8080-exec-2] o.s.orm.jpa.JpaTransactionManager        : Found thread-bound EntityManager [SessionImpl(571954782<open>)] for JPA transaction
2024-06-14T14:03:14.595+03:00 DEBUG 30914 --- [nio-8080-exec-2] o.s.orm.jpa.JpaTransactionManager        : Creating new transaction with name [ing.hub.ingHub.service.PersonKafkaService.processPersonDetails]: PROPAGATION_REQUIRED,ISOLATION_DEFAULT
2024-06-14T14:03:14.597+03:00 DEBUG 30914 --- [nio-8080-exec-2] o.s.orm.jpa.JpaTransactionManager        : Exposing JPA transaction as JDBC [org.springframework.orm.jpa.vendor.HibernateJpaDialect$HibernateConnectionHandle@31ec2652]
2024-06-14T14:03:14.597+03:00 DEBUG 30914 --- [nio-8080-exec-2] o.s.orm.jpa.JpaTransactionManager        : Found thread-bound EntityManager [SessionImpl(571954782<open>)] for JPA transaction
2024-06-14T14:03:14.597+03:00 DEBUG 30914 --- [nio-8080-exec-2] o.s.orm.jpa.JpaTransactionManager        : Participating in existing transaction
2024-06-14T14:03:14.599+03:00 DEBUG 30914 --- [nio-8080-exec-2] o.s.orm.jpa.JpaTransactionManager        : Found thread-bound EntityManager [SessionImpl(571954782<open>)] for JPA transaction
2024-06-14T14:03:14.599+03:00 DEBUG 30914 --- [nio-8080-exec-2] o.s.orm.jpa.JpaTransactionManager        : Participating in existing transaction
2024-06-14T14:03:14.799+03:00 DEBUG 30914 --- [nio-8080-exec-2] o.s.orm.jpa.JpaTransactionManager        : Initiating transaction commit
2024-06-14T14:03:14.800+03:00 DEBUG 30914 --- [nio-8080-exec-2] o.s.orm.jpa.JpaTransactionManager        : Committing JPA transaction on EntityManager [SessionImpl(571954782<open>)]
2024-06-14T14:03:14.816+03:00 DEBUG 30914 --- [nio-8080-exec-2] o.s.orm.jpa.JpaTransactionManager        : Not closing pre-bound JPA EntityManager after transaction

我原本期望能看到KafkaTransactionManager相关的日志（比如事务启动、提交等），但日志中完全没有相关记录，这说明Kafka事务似乎并未参与其中。请问我是否遗漏了什么配置，还是功能未按预期工作？

问题出在事务同步的自动触发条件以及KafkaTemplate的事务绑定逻辑上，以下是具体修复步骤：

1. 开启Kafka事务同步支持 #

Spring Kafka的事务同步不会自动开启，需要显式配置KafkaTransactionManager的同步策略：

@Bean
public KafkaTransactionManager personKafkaTransactionManager() {
    KafkaTransactionManager manager = new KafkaTransactionManager(personProducerFactory());
    // 强制开启事务同步，绑定到当前事务上下文
    manager.setTransactionSynchronization(KafkaTransactionManager.SYNCHRONIZATION_ALWAYS);
    return manager;
}

2. 显式绑定KafkaTemplate到事务管理器 #

默认情况下KafkaTemplate会关联同名事务管理器，但存在多个事务管理器时，最好显式指定避免歧义：

@Bean
public KafkaTemplate<Long, PersonDto> personKafkaTemplate() {
    KafkaTemplate<Long, PersonDto> template = new KafkaTemplate<>(personProducerFactory());
    // 绑定到指定的Kafka事务管理器
    template.setTransactionManager(personKafkaTransactionManager());
    return template;
}

3. 整合JPA与Kafka事务（两种可选方式） #

由于你配置了@Primary的JPA事务管理器，默认事务只会使用它，需要让Kafka事务参与到同一事务上下文：

方式一：显式指定多事务管理器 #

在服务方法上添加两个@Transactional注解，分别指定JPA和Kafka事务管理器：

@Transactional(transactionManager = "transactionManager") // JPA事务
@Transactional(transactionManager = "personKafkaTransactionManager") // Kafka事务
public void processPersonDetails(PersonDetailsDto personDetails) {
    PersonDto person = personJpaService.save(personDetails);
    personKafkaTemplate.send(OUT_TOPIC, person.id(), person);
}

方式二：使用链式事务管理器（推荐） #

创建ChainedTransactionManager整合两个事务管理器，保证两者原子性（同时提交/回滚）：

@Bean
public ChainedTransactionManager chainedTransactionManager(JpaTransactionManager jpaTm, KafkaTransactionManager kafkaTm) {
    return new ChainedTransactionManager(jpaTm, kafkaTm);
}

然后修改服务方法的事务注解：

@Transactional(transactionManager = "chainedTransactionManager")
public void processPersonDetails(PersonDetailsDto personDetails) {
    PersonDto person = personJpaService.save(personDetails);
    personKafkaTemplate.send(OUT_TOPIC, person.id(), person);
}

4. 优化日志配置 #

新增事务同步相关日志打印，确保能看到Kafka事务的执行细节：

logging:
  level.root: error
  level:
    org.apache.kafka: ERROR
    org.springframework.orm.jpa.JpaTransactionManager: trace
    org.springframework.kafka.transaction: trace
    org.springframework.transaction.support: trace # 打印事务同步的核心逻辑

验证结果 #

完成配置后重新运行，你应该能看到类似以下的Kafka事务日志：

o.s.k.t.KafkaTransactionManager : Creating new transaction with name [...]: PROPAGATION_REQUIRED,ISOLATION_DEFAULT
o.s.k.t.KafkaTransactionManager : Participating in existing transaction
o.s.k.t.KafkaTransactionManager : Initiating transaction commit

关键原理 #

• Spring Kafka的事务同步依赖TransactionSynchronizationManager，只有当Kafka事务管理器被激活并绑定到当前事务上下文时，才会触发同步逻辑。
• 使用ChainedTransactionManager时，会按顺序调用各个事务管理器的提交/回滚方法，保证跨资源事务的原子性。
• 单事务管理器场景下，Kafka事务会作为同步器绑定到JPA事务，在JPA事务提交后再执行Kafka事务的提交（或回滚）。

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