Sample spring transaction with jpa and kafka?

Question

After upgrading the spring boot to 2.2.5 from 2.1.11 version, the kafka client is producing the messages to the broker before the jpa transaction is committed. Without using the chained kafka transaction manager the transaction was working fine. Is there any backward compatibility changes introducing in transaction between 2.1.x and 2.2.x? Could someone provide any working transactions manager spanning JPA and Kafka?

I was using just the following transaction manager for JPA

  @Bean
  @Primary
  public PlatformTransactionManager transactionManager(final EntityManagerFactory emf) {
    final JpaTransactionManager txManager = new JpaTransactionManager();
    txManager.setEntityManagerFactory(emf);
    return txManager;
  }

I was using the following properties for kafka transaction:

spring.cloud.stream.kafka.default.consumer.configuration.isolation.level: read_committed spring.cloud.stream.kafka.binder.transaction.transaction-id-prefix: xyz-0 spring.kafka.producer.transaction-id-prefix: xyz-0

user_8715331 · Answer

We can create chained transaction with jpa and kafka. We can use BinderFactory to get the kafka binder and create the binder kafka transaction manager.

@Bean(name = "chainedTransactionManager")
  @Primary
  public PlatformTransactionManager chainedTransactionManager(JpaTransactionManager jpaTM,
      BinderFactory binders) {
    Binder binder = binders.getBinder("kafka", MessageChannel.class);
    if (binder instanceof KafkaMessageChannelBinder) {
      ProducerFactory pf =
          ((KafkaMessageChannelBinder) binder).getTransactionalProducerFactory();
      KafkaTransactionManager ktm = new KafkaTransactionManager<>(pf);
      ktm.setTransactionSynchronization(
          AbstractPlatformTransactionManager.SYNCHRONIZATION_ON_ACTUAL_TRANSACTION);
      return new ChainedKafkaTransactionManager(jpaTM, ktm);
    } else {
      return jpaTM;
    }
  }```

Sample spring transaction with jpa and kafka?

1 Answers