spring-transactions - Spring Kafka 中的事务同步

Question

我想将 kafka 事务与存储库事务同步:

@Transactional
public void syncTransaction(){
  myRepository.save(someObject)
  kafkaTemplate.send(someEvent)
}

由于合并( https://github.com/spring-projects/spring-kafka/issues/373 )并且根据文档这是可能的。尽管如此，我在理解和实现该功能方面仍有问题。
查看 https://docs.spring.io/spring-kafka/reference/html/#transaction-synchronization 中的示例我必须创建一个 MessageListenerContainer 来监听我自己的事件。
我是否仍然需要使用 KafkaTemplate 发送我的事件？
MessageListenerContainer 是否禁止发送到代理？
如果我理解正确，kafkaTemplate 和 kafkaTransactionManager 必须使用相同的生产者工厂，我必须在其中启用事务设置 transactionIdPrefix。在我的示例中，我必须将 messageListenerContainer 的 TransactionManager 设置为 DataSourceTransactionManager。那是对的吗？
从我的角度来看，我通过 kafkaTemplate 发送事件，监听我自己的事件并再次使用 kafkaTemplate 转发事件看起来很奇怪。
如果我能得到一个简单的 kafka 事务与存储库事务的同步示例和解释，我真的会帮助我。

Answer 1

如果监听器容器配备了 KafkaTransactionManager ，容器将创建一个生产者，该生产者将被任何下游 kafka 模板使用，并且容器将为您发送偏移量到交易。

如果容器有其他事务管理器，则容器无法发送偏移量，因为它无权访问生产者(或模板)。

另一种解决方案是使用 @Transactional 注释您的方法(使用数据源 TM)并使用 kafka TM 配置容器。

这样，您的 DB tx 将在线程返回容器之前提交，然后将偏移量发送到 kafka 事务并提交。

见 the framework test cases举些例子。