使用 JPA 启动时的分区分配和 ChainedKafkaTransactionManager答案

【问题标题】：On Partitions Assignment and ChainedKafkaTransactionManager at startup with JPA使用 JPA 启动时的分区分配和 ChainedKafkaTransactionManager
【发布时间】：2020-05-14 10:54:46
【问题描述】：

我有很多事务型消费者，它们的 ChainedKafkaTransactionManager 基于 JpaTransactionManager 和 KafkaTransactionManager（全部为 @KafkaListener）。

JPA 需要设置一个 ThreadLocal 变量，以便能够知道要连接到哪个 DB（是租户 ID）。

在启动应用程序时，在onPartitionsAssignedlistener中，spring-kafka 正在尝试创建链式 txn，因此尝试创建 JPA txn，但没有设置租户，然后失败。

该租户是通过 http 过滤器和/或 kafka 拦截器（通过事件标头）设置的。

我尝试将自动连接的KafkaListenerEndpointRegistry 与setAutoStartup(false) 一起使用，但我看到消费者没有收到任何事件，可能是因为它们尚未初始化（我认为它们是按需初始化的）。

如果我设置一个模拟租户 ID 并在应用程序准备好时调用 registry.start()，初始化似乎是在其他线程中完成的（可能是因为我使用的是 ConcurrentKafkaListenerContainerFactory），所以它不起作用。

有没有办法避免初始 onPartitionsAssigned 侦听器上的 JPA 事务，这是消费者初始化的一部分？

【问题讨论】：

不清楚你的意思。收到记录时最终将调用侦听器的同一线程首先调用onPartitionsAssigned()。如果没有为此设置 TL，那么在调用侦听器时也不会设置。什么设置了 TL，什么时候？
我编辑了这个问题，试图更清楚。理想情况下，我希望在没有 JPA 的情况下完成消费者初始化，并使用链式 txn 管理器进行正常的 pub/sub。

标签： apache-kafka spring-transactions spring-kafka

【解决方案1】：

如果您的链式 TM 首先具有 KafkaTM，然后是 JPA TM（这是正常情况），您可以通过将 Kafka TM 注入容器并使用 @Transactional（仅使用 JPA TM）来实现类似的功能在侦听器上）以在调用侦听器时启动 JPA 事务。

事务提交之间的时间会略微增加，但会提供类似的功能。

如果这对您不起作用，请打开 GitHub 问题；我们可以禁用分配时的初始提交，或者根本不使用事务（可选）。

【讨论】：

您的意思是将@Transactional 添加到所有事务性@KafkaListener 对吗？相似还是相等？因为我们有一个非常密集的 EDA，并且我们依靠 jpa 和 kafka 之间的事务能力来实现本地一致性。我可以打开问题以获得其他选项吗？
是的；请打开问题；我们应该添加选项。在这种情况下使用@Transactional 几乎是相同的；使用 CKTM（假设首先是 KTM），我们启动 KTx，然后是 JPATx，然后调用侦听器；当方法退出时，提交 JPATx，然后是 KTx。只有容器中的 KTM 和@Transactional 中的 JPATM，我们启动 KTx，调用监听器，事务拦截器启动 JPATx，然后调用实际的监听器；当监听器正常退出时，拦截器提交JPA，当拦截器退出时，容器提交KTx。
话虽如此，如果您不写信给 Kafka（以及 JPA），Kafka 事务除了一些额外的开销之外并不会真正为您带来任何好处。
已创建问题：github.com/spring-projects/spring-kafka/issues/1370。谢谢！