如何将@Transactional 与@KafkaListener 一起使用？

Question

是否有可能将声明性 tx 管理（通过 @Transactional）与 @KafkaListener 注释方法一起使用？ 例如，我想使用它来为每个侦听器定义单独的 tx 超时。 我的设置如下：

事务管理器：

@Bean
@ConditionalOnBean(value = {HibernateTransactionManager.class})
public ChainedKafkaTransactionManager<Object, Object> chainedHibernateTm(KafkaTransactionManager<String, String> kafkaTransactionManager,
                                                                                 org.springframework.orm.hibernate5.HibernateTransactionManager hibernateTransactionManager) {
  return new ChainedKafkaTransactionManager<>(
    kafkaTransactionManager, 
    hibernateTransactionManager);
}

KafkaListener：

@KafkaListener(topic = "my_topic")
@Transactional(timeout = 5)
public void handleMessage(SomeMessage message){
}

问题是 - KafkaMessageListenerContainer 在调用此类方法之前创建自己的事务 - 它使用自己的 TransactionTemplate：

@Nullable
private TransactionTemplate determineTransactionTemplate() {
  return this.transactionManager != null
    ? new TransactionTemplate(this.transactionManager)
    : null;
}

不使用 TransactionInterceptor。那么如何为具体的@KafkaListener 方法设置特定的 tx 超时呢？

Answer 1

通过普通的@Transactional 进行事务管理的各个方面无法与您所希望的那样与 Kafka 侦听器一起使用 - 没有内置交互。

有一个专门的章节如何在事务中使用应用程序事件：Transaction Bound Event 使用注释 @TransactionalEventListener 有助于确保事务已成功完成。

您可以使用@EventListener 注解注册一个常规事件监听器。如果需要将其绑定到事务，请使用@TransactionalEventListener。这样做时，监听器默认绑定到事务的提交阶段。

Answer 2

可以做到，但有点复杂，因为你必须将消耗的偏移量发送到 Kafka 事务。

您可以将KafkaTransactionManager 用于容器，将@Transactional 用于HibernateTransactionManager，而不是使用ChainedKafkaTransactionManager。

这将产生类似的结果，因为休眠 TX 将在 Kafka 事务之前提交（如果休眠提交失败，Kafka TX 将回滚）。

编辑

要将不同的链式 TM 配置到每个侦听器容器中，您可以执行以下操作。

@Component
class ContainerFactoryCustomizer {

    ContainerFactoryCustomizer(AbstractKafkaListenerContainerFactory<?, ?, ?> factory,
            ChainedKafkaTransactionManager<?, ?> chainedOne,
            ChainedKafkaTransactionManager<?, ?> chainedTwo) {
        factory.setContainerCustomizer(
                container -> {
                    String groupId = container.getContainerProperties().getGroupId();
                    if (groupId.equals("foo")) {
                        container.getContainerProperties().setTransactionManager(chainedOne);
                    }
                    else {
                        container.getContainerProperties().setTransactionManager(chainedTwo);
                    }
                });
    }

}

每个链接的 TM 都有一个具有不同默认超时的 Hibernate TM。

groupid 由 @KafkaListener id 或 groupId 属性填充。