Transaction Synchronization in Spring Kafka

spring-transactions spring-kafka
Eike Behrends picture Eike Behrends · Nov 17, 2017 · Viewed 13.1k times · Source

I want to synchronize a kafka transaction with a repository transaction:

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

Since the merge (https://github.com/spring-projects/spring-kafka/issues/373) and according to the doc this is possible. Nevertheless i have problems to understand and implement that feature. Looking at the example in https://docs.spring.io/spring-kafka/reference/html/#transaction-synchronization I have to create a MessageListenerContainer to listen to my own events. Do I still have to send my events using the KafkaTemplate? Does the MessageListenerContainer prohibit the sending to the broker?

And if i understand correctly the kafkaTemplate und the kafkaTransactionManager have to use the same producerFactory in which i have to enable Transaction setting a transactionIdPrefix. And in my example i have to set the TransactionManager of the messageListenerContainer to the DataSourceTransactionManager. Is that correct?

From my perspective it looks weird that I send an event via kafkaTemplate, listen to my own event and forward the event using the kafkaTemplate again.

I would really help me if i can get an example for a simple synchronization of a kafka transaction with a repository transaction and an explanation.

Answer

Gary Russell picture Gary Russell · Nov 17, 2017

If the listener container is provisioned with a KafkaTransactionManager, the container will create a producer which will be used by any downstream kafka template and the container will send the offsets to the transaction for you.

If the container has some other transaction manager, the container can't send the offsets since it doesn't have access to the producer (or template).

Another solution is to annotate your method with @Transactional (with the datasource TM) and configure the container with a kafka TM.

That way, your DB tx will commit just before the thread returns to the container which will then send the offsets to the kafka transaction and commit it.

See the framework test cases for examples.

Related questions

Synchronising transactions between database and Kafka producer

We have a micro-services architecture, with Kafka used as the communication mechanism between the services. Some of the services have their own databases. Say the user makes a call to Service A, which should result in a record (or set …

Read More
Spring transaction REQUIRED vs REQUIRES_NEW : Rollback Transaction

I have a method that has the propagation = Propagation.REQUIRES_NEW transactional property: @Transactional(propagation = Propagation.REQUIRES_NEW) public void createUser(final UserBean userBean) { //Some logic here that requires modification in DB } This method can be called multiple times simultaneously, …

Read More
The matching wildcard is strict, but no declaration can be found for element 'tx:annotation-driven'

I am trying to configure JSF+Spring+hibernate and I'm tying to run a test but when I use this "tx:annotation-driven" on my application-context.xml file, I get this error: The matching wildcard is strict, but no declaration can …

Read More