How to use @Transactional with @KafkaListener?

user_10523025 0
Viewed 3307

Is there any possibility to use declarative tx management (via @Transactional) with @KafkaListener annotated method ? I would like to use it in order to, for example, define separate tx timeout per listener. My setup is as follows:

TransactionManager:

@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){
}

The problem is - KafkaMessageListenerContainer creates it's own transaction before such method is invoked - it uses its own TransactionTemplate:

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

TransactionInterceptor is not used. So how to set specific tx timeout for concrete @KafkaListener method ?

spring-kafka spring-boot spring-transactions java
2 Answers

The aspects of transaction management through plain @Transactional don't work with Kafka listeners as you intent - there is no in-built interaction.

There is a dedicated chapter how to use application events with within a transaction: Transaction Bound Event using the annotation @TransactionalEventListener helps to assure that the transaction has completed successfully.

You can register a regular event listener by using the @EventListener annotation. If you need to bind it to the transaction, use @TransactionalEventListener. When you do so, the listener is bound to the commit phase of the transaction by default.

user_37649650

It can be done, but it's a bit complicated because you have to send the consumed offset(s) to the Kafka transaction.

Instead of using a ChainedKafkaTransactionManager, you can use a KafkaTransactionManager for the container and @Transactional for the HibernateTransactionManager.

This will give similar results, since the hibernate TX will commit just before the Kafka transaction (and if the hibernate commit fails, the Kafka TX will roll back).

EDIT

To configure a different chained TM into each listener container you can do something like this.

@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);
                    }
                });
    }

}

Where each chained TM has a Hibernate TM with a different default timeout.

The groupid is populated from the @KafkaListener id or groupId property.

user_12407630
Related