title: Spring Kafka: Transactional producer error handling
body: <p>I have a Kafka spring application with a read-process-write pattern. I want to make sure that the Kafka transaction rolls back if there are any producer errors so that the record is re-consumed using a SeekToCurrentErrorHandler. The default behavior seems to be to log the producer error and continue processing/commit. To override this default behaviour, i have implemented a ProducerListener that throws an exception in the onError method. Is this the recommended approach for ensuring a rollback and the intent behind spring providing us with the listener hook?</p>
  
  <p>Logs showing an exception followed by a commit (The exception didnt result in a rollback)</p>
  
  <pre><code>2020-04-02 18:20:18.314|[ org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1]|                                o.s.kafka.core.KafkaTemplate|TRACE|                                 456|              d3410ae8-c964-41e7-98be-6706a6f2b3b2| Sending: ProducerRecord
  2020-04-02 18:20:18.345|[                      kafka-producer-network-thread | producer-13]|                           org.apache.kafka.clients.Metadata|ERROR|                                    |                                                  | [Producer clientId=producer-13, transactionalId=tx-0] Topic authorization failed for topics 
  2020-04-02 18:20:18.354|[ org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1]|                             o.s.k.s.LoggingProducerListener|ERROR|                                 456|              d3410ae8-c964-41e7-98be-6706a6f2b3b2| Exception thrown when sending a message with key='170854907' org.apache.kafka.common.errors.TopicAuthorizationException: Not authorized to access topics: [loyalty-retail-outlet-trans-resp-dev1]
  2020-04-02 18:20:18.367|[ org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1]|      o.s.k.l.KafkaMessageListenerContainer$ListenerConsumer|INFO |                                    |                                                  | Sending offsets to transaction: {loyalty-retail-outlet-earn-req-dev-5=OffsetAndMetadata{offset=2220, leaderEpoch=null, metadata=''}}
  2020-04-02 18:20:18.368|[ org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1]|                         o.s.k.c.DefaultKafkaProducerFactory|TRACE|                                    |                                                  | CloseSafeProducer [delegate=org.apache.kafka.clients.producer.KafkaProducer@5abb103c, txId=tx-0] sendOffsetsToTransaction({loyalty-retail-outlet-earn-req-dev-5=OffsetAndMetadata{offset=2220, leaderEpoch=null, metadata=''}}, earn-unit-test)
  2020-04-02 18:20:18.769|[ org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1]|                             o.s.k.t.KafkaTransactionManager|DEBUG|                                    |                                                  | Initiating transaction commit
  2020-04-02 18:20:18.769|[ org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1]|                         o.s.k.c.DefaultKafkaProducerFactory|DEBUG|                                    |                                                  | CloseSafeProducer [delegate=org.apache.kafka.clients.producer.KafkaProducer@5abb103c, txId=tx-0] commitTransaction()
  2020-04-02 18:20:18.816|[ org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1]|                         o.s.k.c.DefaultKafkaProducerFactory|TRACE|                                    |                                                  | CloseSafeProducer [delegate=org.apache.kafka.clients.producer.KafkaProducer@5abb103c, txId=tx-0] close(PT5S)
  </code></pre>
  
  <p>The records are produced within a Kafka listener using Kafka Template (read-process-write pattern).</p>
  
  <p>Kafka Template config</p>
  
  <pre><code>    @Bean
      public KafkaTemplate&lt;Integer, TransactionResponse&gt; kafkaTemplate(
              ProducerFactory&lt;Integer, TransactionResponse&gt; producerFactory
              , ProducerListener&lt;Integer, TransactionResponse&gt; producerListener) {
          KafkaTemplate&lt;Integer, TransactionResponse&gt; kafkaTemplate = new KafkaTemplate&lt;&gt;(producerFactory);
  //        kafkaTemplate.setProducerListener(producerListener);
          return kafkaTemplate;
      }
  </code></pre>
  
  <p>application.properties</p>
  
  <pre><code>spring:
    kafka:
      producer:
        transaction-id-prefix: tx-
        acks: all
        key-serializer: org.apache.kafka.common.serialization.IntegerSerializer
        value-serializer: io.confluent.kafka.serializers.KafkaAvroSerializer
        properties:
          enable.idempotence: true
          delivery.timeout.ms: 180000
  </code></pre>
  
  <p>listener</p>
  
  <pre><code>   @KafkaListener(topics = "${earn.request.topic}", clientIdPrefix = "EarnConsumer", containerFactory = "earnListenerContainerFactory")
      public void listen(List&lt;TransactionRequest&gt; requestList,
                         @Header(KafkaHeaders.GROUP_ID) String groupId,
                         @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partitions,
                         @Header(KafkaHeaders.OFFSET) String offsets,
                         @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
  
  </code></pre>
  
  <p>send response method (executes within the listener code)</p>
  
  <pre><code>    public void sendResponse(TransactionResponse transactionResponse) {
          kafkaTemplate.send(earnResponseTopic, transactionResponse.getEventSummary().getMemberId(), transactionResponse);
      }
  </code></pre>
  
  <p>container config</p>
  
  <pre><code>   @Bean
      public ConcurrentKafkaListenerContainerFactory&lt;Integer, EarnTransactionRequest&gt; earnListenerContainerFactory(
              ConsumerFactory&lt;Integer, EarnTransactionRequest&gt; consumerFactory
              , SeekToCurrentBatchErrorHandler seekToCurrentBatchErrorHandler
              , KafkaTransactionManager ktm
      ) {
          ConcurrentKafkaListenerContainerFactory&lt;Integer, EarnTransactionRequest&gt; containerFactory = new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
          containerFactory.setConsumerFactory(consumerFactory);
          containerFactory.setBatchListener(true);
          containerFactory.setBatchErrorHandler(seekToCurrentBatchErrorHandler);
          containerFactory.setConcurrency(numConcurrentConsumers);
  
          containerFactory.getContainerProperties().setTransactionManager(ktm);
          containerFactory.getContainerProperties().setAckOnError(false);
          containerFactory.getContainerProperties().setAckMode(ContainerProperties.AckMode.BATCH);
          containerFactory.getContainerProperties().setCommitLogLevel(LogIfLevelEnabled.Level.INFO);
          containerFactory.getContainerProperties().setLogContainerConfig(true);
          containerFactory.getContainerProperties().setMissingTopicsFatal(true);
  
          return containerFactory;
      }
  </code></pre>
  
  <p><strong>EDIT: Simplified application</strong></p>
  
  <p>@Component
  public class QuickTest {</p>
  
  <pre><code>private final String responseTopic;
  private final KafkaTemplate&lt;Integer, TransactionResponse&gt; kafkaTemplate;
  
  public QuickTest(@Value("${response.topic}") String responseTopic
          , KafkaTemplate&lt;Integer, TransactionResponse&gt; kafkaTemplate) {
      this.responseTopic = responseTopic;
      this.kafkaTemplate = kafkaTemplate;
  }
  
  @KafkaListener(topics = "${request.topic}", clientIdPrefix = "Consumer")
  public void listen(TransactionRequest requestList) {
      kafkaTemplate.send(responseTopic, 123456789, null);
  }
  </code></pre>
  
  <p>}</p>
  
  <p><strong>Logs from start of one transaction to other</strong></p>
  
  <pre><code>
  2020-04-03 19:04:54.901|[ org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1]|      o.s.k.l.KafkaMessageListenerContainer$ListenerConsumer|TRACE|Processing ConsumerRecord(topic = req-dev, partition = 1, leaderEpoch = 4, offset = 2185, CreateTime = 1585642853682, serialized key size = -1, serialized value size = 184, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = {})
  
  2020-04-03 19:04:54.901|[ org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1]|                             o.s.k.t.KafkaTransactionManager|DEBUG|Creating new transaction with name [null]: PROPAGATION_REQUIRED,ISOLATION_DEFAULT
  
  2020-04-03 19:04:54.901|[ org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1]|                         o.s.k.c.DefaultKafkaProducerFactory|DEBUG|CloseSafeProducer [delegate=org.apache.kafka.clients.producer.KafkaProducer@43926a5b, txId=transactionx-g21.req-dev.1] beginTransaction()
  
  2020-04-03 19:04:54.901|[ org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1]|                             o.s.k.t.KafkaTransactionManager|DEBUG|Created Kafka transaction on producer [CloseSafeProducer [delegate=org.apache.kafka.clients.producer.KafkaProducer@43926a5b, txId=transactionx-g21.req-dev.1]]
  
  2020-04-03 19:04:54.902|[ org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1]|             o.s.k.l.a.RecordMessagingMessageListenerAdapter|DEBUG|Processing [GenericMessage [payload={"eventSummary": {"eventId": "102"}}]]
  
  2020-04-03 19:04:54.902|[ org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1]|                                o.s.kafka.core.KafkaTemplate|TRACE|Sending: ProducerRecord(topic= resp-test, partition=null, headers=RecordHeaders(headers = [], isReadOnly = false), key=123456789, value=null, timestamp=null)
  
  2020-04-03 19:04:54.902|[ org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1]|                         o.s.k.c.DefaultKafkaProducerFactory|TRACE|CloseSafeProducer [delegate=org.apache.kafka.clients.producer.KafkaProducer@43926a5b, txId=transactionx-g21.req-dev.1] send(ProducerRecord(topic= resp-test, partition=null, headers=RecordHeaders(headers = [], isReadOnly = false), key=123456789, value=null, timestamp=null))
  
  2020-04-03 19:04:54.928|[                       kafka-producer-network-thread | producer-8]|                        o.apache.kafka.clients.NetworkClient|WARN |[Producer clientId=producer-8, transactionalId=transactionx-g21.req-dev.1] Error while fetching metadata with correlation id 22 : { resp-test=TOPIC_AUTHORIZATION_FAILED}
  
  2020-04-03 19:04:54.928|[                       kafka-producer-network-thread | producer-8]|                           org.apache.kafka.clients.Metadata|ERROR|[Producer clientId=producer-8, transactionalId=transactionx-g21.req-dev.1] Topic authorization failed for topics [ resp-test]
  
  2020-04-03 19:04:54.928|[ org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1]|                             o.s.k.s.LoggingProducerListener|ERROR|Exception thrown when sending a message with key='123456789' and payload='null' to topic  resp-test:
  
  org.apache.kafka.common.errors.TopicAuthorizationException: Not authorized to access topics: [ resp-test]
  2020-04-03 19:04:54.928|[ org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1]|                                o.s.kafka.core.KafkaTemplate|DEBUG|Failed to send: ProducerRecord(topic= resp-test, partition=null, headers=RecordHeaders(headers = [], isReadOnly = false), key=123456789, value=null, timestamp=null)
  
  org.apache.kafka.common.errors.TopicAuthorizationException: Not authorized to access topics: [ resp-test]
  2020-04-03 19:04:54.928|[ org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1]|                                o.s.kafka.core.KafkaTemplate|TRACE|Sent: ProducerRecord(topic= resp-test, partition=null, headers=RecordHeaders(headers = [], isReadOnly = false), key=123456789, value=null, timestamp=null)
  
  2020-04-03 19:04:54.928|[ org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1]|      o.s.k.l.KafkaMessageListenerContainer$ListenerConsumer|TRACE|Ack: ConsumerRecord(topic = req-dev, partition = 1, leaderEpoch = 4, offset = 2185, CreateTime = 1585642853682, serialized key size = -1, serialized value size = 184, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = {"eventSummary": {"eventId": "102"}})
  
  2020-04-03 19:04:54.928|[ org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1]|      o.s.k.l.KafkaMessageListenerContainer$ListenerConsumer|DEBUG|Sending offsets to transaction: {req-dev-1=OffsetAndMetadata{offset=2186, leaderEpoch=null, metadata=''}}
  
  2020-04-03 19:04:54.928|[ org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1]|                         o.s.k.c.DefaultKafkaProducerFactory|TRACE|CloseSafeProducer [delegate=org.apache.kafka.clients.producer.KafkaProducer@43926a5b, txId=transactionx-g21.req-dev.1] sendOffsetsToTransaction({req-dev-1=OffsetAndMetadata{offset=2186, leaderEpoch=null, metadata=''}}, g21)
  
  2020-04-03 19:04:55.043|[ org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1]|                             o.s.k.t.KafkaTransactionManager|DEBUG|Initiating transaction commit
  
  2020-04-03 19:04:55.043|[ org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1]|                         o.s.k.c.DefaultKafkaProducerFactory|DEBUG|CloseSafeProducer [delegate=org.apache.kafka.clients.producer.KafkaProducer@43926a5b, txId=transactionx-g21.req-dev.1] commitTransaction()
  
  2020-04-03 19:04:55.090|[ org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1]|                         o.s.k.c.DefaultKafkaProducerFactory|TRACE|CloseSafeProducer [delegate=org.apache.kafka.clients.producer.KafkaProducer@43926a5b, txId=transactionx-g21.req-dev.1] close(PT5S)
  
  2020-04-03 19:04:55.091|[ org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1]|      o.s.k.l.KafkaMessageListenerContainer$ListenerConsumer|TRACE|Processing ConsumerRecord(topic = req-dev, partition = 1, leaderEpoch = 4, offset = 2186, CreateTime = 1585644055280, serialized key size = -1, serialized value size = 184, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = {"eventSummary": {"eventId": "104"})
  
  2020-04-03 19:04:55.091|[ org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1]|                             o.s.k.t.KafkaTransactionManager|DEBUG|Creating new transaction with name [null]: PROPAGATION_REQUIRED,ISOLATION_DEFAULT
  
  2020-04-03 19:04:55.091|[ org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1]|                         o.s.k.c.DefaultKafkaProducerFactory|DEBUG|CloseSafeProducer [delegate=org.apache.kafka.clients.producer.KafkaProducer@43926a5b, txId=transactionx-g21.req-dev.1] beginTransaction()
  </code></pre>