事务模式下Kafka ConsumerRecordRecoverer工作机制及事务配置疑问

核心问题拆解 #

你的测试结果差异本质是Recoverer的执行上下文是否绑定事务、以及事务传播行为导致的，先明确几个关键前提：

1. 监听器抛出异常后，原事务会先触发回滚，AfterRollbackProcessor是在事务回滚完成后才调用Recoverer——这意味着默认情况下Recoverer不在监听器的事务上下文内。
2. 配置了transaction-id-prefix的KafkaTemplate，在无事务上下文时会自动创建临时事务，发送后立即提交；如果绑定到事务上下文，则会等待事务提交时才统一提交发送操作。

三种测试场景的原因分析 #

1. 无@Transactional注解 #

Recoverer执行时无事务上下文，kafkaTemplate.send()每次都会创建临时事务，发送完成后立即提交。即使后续抛出异常，已经完成的发送操作不会回滚，因此errors主题收到3条消息。

2. 无参数@Transactional #

此时Recoverer会尝试绑定当前线程的事务上下文，但监听器的事务已经回滚，线程的事务上下文处于失效状态。Spring事务管理器会根据当前状态尝试复用或新建事务，导致部分发送操作可能因事务回滚丢失，部分因临时事务提交保留，因此消息数不固定。

3. @Transactional(propagation = Propagation.REQUIRES_NEW) #

每次Recoverer执行都会创建独立的全新事务：

• 前两次执行：发送消息后抛出异常，事务回滚，发送操作被撤销，因此errors主题无消息；
• 第三次执行：未抛出异常，事务提交，发送操作生效，因此errors主题仅收到1条消息。
  这完全符合你“前两次回滚、第三次提交”的预期。

@Transactional(Propagation.REQUIRES_NEW)是否合理？ #

这取决于你的业务需求：

• 如果需要保证重试逻辑与错误消息发送的原子性（即重试失败时，错误消息也不能被提交），那么REQUIRES_NEW是合理的——它让Recoverer的操作完全独立于监听器的事务上下文，确保异常时能回滚发送操作。
• 如果不需要原子性（比如即使重试失败，也要把错误消息持久化到errors主题），则不需要加事务注解，让发送操作自动提交即可。

额外注意事项 #

• 你的Recoverer用ThreadLocal记录重试次数，虽然最后会重置为0，但listener配置了concurrency:3，线程池线程复用可能导致ThreadLocal值混乱，建议改用RetryContext传递重试次数（DefaultAfterRollbackProcessor的重试逻辑会维护RetryContext）。
• DefaultAfterRollbackProcessor构造参数的最后一个true表示commitRecovered，即Recoverer执行成功后会自动提交消费偏移量，避免重复消费。

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