title: Spring Kafka transaction enabled but consumer still received rollback message
body: <p>I am using Spring kafka transaction for my producer and consumer applications.</p>
  
  <p>The requirement is on producer side there are multiple steps: send message to kafka and then save to db. If save to db failed want to rollback the message send to kafka as well.</p>
  
  <p>So on the consumer side, i set the <code>isolation.leve</code> to <code>read_committed</code>, then if the message is rollback from kafka, the consumer shouldn't read it.</p>
  
  <p>Code for Producer application is:</p>
  
  <pre><code>@Configuration
  @EnableKafka
  public class KafkaConfiguration {
  
    @Bean
    public ProducerFactory&lt;String, Customer&gt; producerFactory() {
      DefaultKafkaProducerFactory&lt;String, Customer&gt; pf = new DefaultKafkaProducerFactory&lt;&gt;(producerConfigs());
      pf.setTransactionIdPrefix("customer.txn.tx-");
      return pf;
    }
  
    @Bean
    public Map&lt;String, Object&gt; producerConfigs() {
      Map&lt;String, Object&gt; props = new HashMap&lt;&gt;();
      // create a minimum Producer configs
      props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "http://127.0.0.1:9092");
      props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
      props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);
      props.put("schema.registry.url", "http://127.0.0.1:8081");
  
      // create safe Producer
      props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
      props.put(ProducerConfig.ACKS_CONFIG, "all");
      props.put(ProducerConfig.RETRIES_CONFIG, Integer.toString(Integer.MAX_VALUE));
      props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "5"); // kafka 2.0 &gt;= 1.1 so we can keep this as 5. Use 1 otherwise.
  
      // high throughput producer (at the expense of a bit of latency and CPU usage)
      props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");
      props.put(ProducerConfig.LINGER_MS_CONFIG, "20");
      props.put(ProducerConfig.BATCH_SIZE_CONFIG, Integer.toString(32 * 1024)); // 32 KB batch size
      return props;
    }
  
    @Bean
    public KafkaTemplate&lt;String, Customer&gt; kafkaTemplate() {
      return new KafkaTemplate&lt;&gt;(producerFactory());
    }
  
    @Bean
    public KafkaTransactionManager kafkaTransactionManager(ProducerFactory&lt;String, Customer&gt; producerFactory) {
      KafkaTransactionManager&lt;String, Customer&gt; ktm = new KafkaTransactionManager&lt;&gt;(producerFactory);
      ktm.setTransactionSynchronization(AbstractPlatformTransactionManager.SYNCHRONIZATION_ON_ACTUAL_TRANSACTION);
      return ktm;
    }
  
    @Bean
    @Primary
    public JpaTransactionManager jpaTransactionManager(EntityManagerFactory entityManagerFactory) {
      return new JpaTransactionManager(entityManagerFactory);
    }
  
    @Bean(name = "chainedTransactionManager")
    public ChainedTransactionManager chainedTransactionManager(JpaTransactionManager jpaTransactionManager,
                                                               KafkaTransactionManager kafkaTransactionManager) {
      return new ChainedTransactionManager(kafkaTransactionManager, jpaTransactionManager);
    }
  }
  
  
  @Component
  @Slf4j
  public class KafkaProducerService {
  
    private KafkaTemplate&lt;String, Customer&gt; kafkaTemplate;
    private CustomerConverter customerConverter;
    private CustomerRepository customerRepository;
  
    public KafkaProducerService(KafkaTemplate&lt;String, Customer&gt; kafkaTemplate, CustomerConverter customerConverter, CustomerRepository customerRepository) {
      this.kafkaTemplate = kafkaTemplate;
      this.customerConverter = customerConverter;
      this.customerRepository = customerRepository;
    }
  
    @Transactional(transactionManager = "chainedTransactionManager", rollbackFor = Exception.class)
    public void sendEvents(String topic, CustomerModel customer) {
      LOGGER.info("Sending to Kafka: topic: {}, key: {}, customer: {}", topic, customer.getKey(), customer);
  //    kafkaTemplate.send(topic, customer.getKey(), customerConverter.convertToAvro(customer));
      kafkaTemplate.executeInTransaction(kt -&gt; kt.send(topic, customer.getKey(), customerConverter.convertToAvro(customer)));
      customerRepository.saveToDb();
    }
  }
  
  
  </code></pre>
  
  <p>So i explicitly throw an exception in the saveToDb method and I can see exception throw out. But the consumer application can still see the message.</p>
  
  <p>Code for consumer:</p>
  
  <pre><code>@Slf4j
  @Configuration
  @EnableKafka
  public class KafkaConfiguration {
  
    @Bean
    ConcurrentKafkaListenerContainerFactory&lt;String, Customer&gt; kafkaListenerContainerFactory() {
      ConcurrentKafkaListenerContainerFactory&lt;String, Customer&gt; factory = new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
      factory.setConsumerFactory(consumerFactory());
      factory.setAfterRollbackProcessor(new DefaultAfterRollbackProcessor&lt;String, Customer&gt;(-1));
  
  
  //    SeekToCurrentErrorHandler errorHandler =
  //        new SeekToCurrentErrorHandler((record, exception) -&gt; {
  //          // recover after 3 failures - e.g. send to a dead-letter topic
  ////          LOGGER.info("***in error handler data, {}", record);
  ////          LOGGER.info("***in error handler headers, {}", record.headers());
  ////          LOGGER.info("value: {}", new String(record.headers().headers("springDeserializerExceptionValue").iterator().next().value()));
  //        }, 3);
  //
  //    factory.setErrorHandler(errorHandler);
  
      return factory;
    }
  
    @Bean
    public ConsumerFactory&lt;String, Customer&gt; consumerFactory() {
      return new DefaultKafkaConsumerFactory&lt;&gt;(consumerConfigs());
    }
  
    @Bean
    public Map&lt;String, Object&gt; consumerConfigs() {
      Map&lt;String, Object&gt; props = new HashMap&lt;&gt;();
      props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
  //    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
  //    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class);
  
      props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
      props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer2.class);
      props.put(ErrorHandlingDeserializer2.VALUE_DESERIALIZER_CLASS, KafkaAvroDeserializer.class);
  
      props.put("schema.registry.url", "http://127.0.0.1:8081");
      props.put("specific.avro.reader", "true");
      props.put("isolation.level", "read_committed");
  
  //    props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
      props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
      props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"); // disable auto commit of offsets
      props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "100"); // disable auto commit of offsets
      return props;
    }
  }
  
  @Component
  @Slf4j
  public class KafkaConsumerService {
  
    @KafkaListener(id = "demo-consumer-stream-group", topics = "customer.txn")
    @Transactional
    public void process(ConsumerRecord&lt;String, Customer&gt; record) {
      LOGGER.info("Customer key: {} and value: {}", record.key(), record.value());
      LOGGER.info("topic: {}, partition: {}, offset: {}", record.topic(), record.partition(), record.offset());
    }
  }
  
  </code></pre>
  
  <p>Did I miss something here?</p>