title: Spring Kafka ChainedKafkaTransactionManager doesn't synchronize with JPA Spring-data transaction
body: <p>I read a ton of Gary Russell answers and posts, but didn't find actual solution for the common use-case for synchronization of the sequence below:</p>
  
  <p><code>recieve from topic A =&gt; save to DB via Spring-data =&gt; send to topic B</code></p>
  
  <p>As i understand properly: there is no guarantee for fully atomic processing in that case and i need to deal with messages deduplication on the client side, but the main issue is that <strong>ChainedKafkaTransactionManager doesn't synchronize with JpaTransactionManager</strong> (see <code>@KafkaListener</code> below)</p>
  
  <p>Kafka config:</p>
  
  <pre><code>@Production
  @EnableKafka
  @Configuration
  @EnableTransactionManagement
  public class KafkaConfig {
  
      private static final Logger log = LoggerFactory.getLogger(KafkaConfig.class);
  
      @Bean
      public ConsumerFactory&lt;String, byte[]&gt; commonConsumerFactory(@Value("${kafka.broker}") String bootstrapServer) {
  
          Map&lt;String, Object&gt; props = new HashMap&lt;&gt;();
          props.put(BOOTSTRAP_SERVERS_CONFIG, bootstrapServer);
  
          props.put(AUTO_OFFSET_RESET_CONFIG, 'earliest');
          props.put(SESSION_TIMEOUT_MS_CONFIG, 10000);
          props.put(ENABLE_AUTO_COMMIT_CONFIG, false);
          props.put(MAX_POLL_RECORDS_CONFIG, 10);
          props.put(MAX_POLL_INTERVAL_MS_CONFIG, 17000);
          props.put(FETCH_MIN_BYTES_CONFIG, 1048576);
          props.put(FETCH_MAX_WAIT_MS_CONFIG, 1000);
          props.put(ISOLATION_LEVEL_CONFIG, 'read_committed');
  
          props.put(KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
          props.put(VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
  
          return new DefaultKafkaConsumerFactory&lt;&gt;(props);
      }
  
      @Bean
      public ConcurrentKafkaListenerContainerFactory&lt;String, byte[]&gt; kafkaListenerContainerFactory(
              @Qualifier("commonConsumerFactory") ConsumerFactory&lt;String, byte[]&gt; consumerFactory,
              @Qualifier("chainedKafkaTM") ChainedKafkaTransactionManager chainedKafkaTM,
              @Qualifier("kafkaTemplate") KafkaTemplate&lt;String, byte[]&gt; kafkaTemplate,
              @Value("${kafka.concurrency:#{T(java.lang.Runtime).getRuntime().availableProcessors()}}") Integer concurrency
      ) {
  
          ConcurrentKafkaListenerContainerFactory&lt;String, byte[]&gt; factory = new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
          factory.getContainerProperties().setMissingTopicsFatal(false);
          factory.getContainerProperties().setTransactionManager(chainedKafkaTM);
  
          factory.setConsumerFactory(consumerFactory);
          factory.setBatchListener(true);
          var arbp = new DefaultAfterRollbackProcessor&lt;String, byte[]&gt;(new FixedBackOff(1000L, 3));
          arbp.setCommitRecovered(true);
          arbp.setKafkaTemplate(kafkaTemplate);
  
          factory.setAfterRollbackProcessor(arbp);
          factory.setConcurrency(concurrency);
  
          factory.afterPropertiesSet();
  
          return factory;
      }
  
      @Bean
      public ProducerFactory&lt;String, byte[]&gt; producerFactory(@Value("${kafka.broker}") String bootstrapServer) {
  
          Map&lt;String, Object&gt; configProps = new HashMap&lt;&gt;();
  
          configProps.put(BOOTSTRAP_SERVERS_CONFIG, bootstrapServer);
  
          configProps.put(BATCH_SIZE_CONFIG, 16384);
          configProps.put(ENABLE_IDEMPOTENCE_CONFIG, true);
  
          configProps.put(KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
          configProps.put(VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
  
          var kafkaProducerFactory = new DefaultKafkaProducerFactory&lt;String, byte[]&gt;(configProps);
          kafkaProducerFactory.setTransactionIdPrefix('kafka-tx-');
  
          return kafkaProducerFactory;
      }
  
      @Bean
      public KafkaTemplate&lt;String, byte[]&gt; kafkaTemplate(@Qualifier("producerFactory") ProducerFactory&lt;String, byte[]&gt; producerFactory) {
          return new KafkaTemplate&lt;&gt;(producerFactory);
      }
  
      @Bean
      public KafkaTransactionManager kafkaTransactionManager(@Qualifier("producerFactory") ProducerFactory&lt;String, byte[]&gt; producerFactory) {
          KafkaTransactionManager ktm = new KafkaTransactionManager&lt;&gt;(producerFactory);
          ktm.setTransactionSynchronization(SYNCHRONIZATION_ON_ACTUAL_TRANSACTION);
          return ktm;
      }
  
      @Bean
      public ChainedKafkaTransactionManager chainedKafkaTM(JpaTransactionManager jpaTransactionManager,
                                                           KafkaTransactionManager kafkaTransactionManager) {
          return new ChainedKafkaTransactionManager(kafkaTransactionManager, jpaTransactionManager);
      }
  
      @Bean(name = "transactionManager")
      public JpaTransactionManager transactionManager(EntityManagerFactory em) {
          return new JpaTransactionManager(em);
      }
  }
  </code></pre>
  
  <p>Kafka listener:</p>
  
  <pre><code>@KafkaListener(groupId = "${group.id}", idIsGroup = false, topics = "${topic.name.import}")
  public void consume(List&lt;byte[]&gt; records, @Header(KafkaHeaders.OFFSET) Long offset) {
      for (byte[] record : records) {
          // cause infinity rollback (perhaps due to batch listener)
          if (true)
              throw new RuntimeExcetion("foo");
  
          // spring-data storage with @Transactional("chainedKafkaTM"), since Spring-data can't determine TM among transactionManager, chainedKafkaTM, kafkaTransactionManager
          var result = storageService.persist(record);
  
          kafkaTemplate.send(result);
      }
  }
  </code></pre>
  
  <p>Spring-kafka version: 2.3.3
  Spring-boot version: 2.2.1</p>
  
  <p>What is a proper way to implement such use-case ?
  Spring-kafka documentation limited only to small/specific examples.</p>
  
  <p><strong>P.s.</strong> when i'm using <code>@Transactional(transactionManager = "chainedKafkaTM", rollbackFor = Exception.class)</code> on <code>@KafkaListener</code> method i am facing endless cyclic rollback, however <code>FixedBackOff(1000L, 3L)</code> is set.</p>
  
  <p><strong>EDIT:</strong> i'm planning to achieve max affordable synchronization between listener, producer and database with configurable retries num.</p>
  
  <p><strong>EDIT:</strong> Code snippets above edited with respect to advised configuration. Using ARBP doesn't solve infinity rollback cycle for me, since the first statement's predicate is always false (<code>SeekUtils.doSeeks</code>):</p>
  
  <pre><code>DefaultAfterRollbackProcessor
  ...
  @Override
      public void process(List&lt;ConsumerRecord&lt;K, V&gt;&gt; records, Consumer&lt;K, V&gt; consumer, Exception exception,
              boolean recoverable) {
  
          if (SeekUtils.doSeeks(((List) records), consumer, exception, recoverable,
                  getSkipPredicate((List) records, exception), LOGGER)
                      &amp;&amp; isCommitRecovered() &amp;&amp; this.kafkaTemplate != null &amp;&amp; this.kafkaTemplate.isTransactional()) {
              ConsumerRecord&lt;K, V&gt; skipped = records.get(0);
              this.kafkaTemplate.sendOffsetsToTransaction(
                      Collections.singletonMap(new TopicPartition(skipped.topic(), skipped.partition()),
                              new OffsetAndMetadata(skipped.offset() + 1)));
          }
      }
  </code></pre>
  
  <p>It is worth saying that there is no active transaction in Kafka Consumer method (<code>TransactionSynchronizationManager.isActualTransactionActive()</code>).</p>