CODEWITHSUNDEEP
apache-kafkatransactionskafka-consumer-apikafka-transactions-api
user5402945
user5402945

Reputation: 165

How @transactional work in Kafka Consumer

I am using @trasactional in my Kafka Consumer. In that same method (marked with @transactional), I am doing some DB transaction. In case of DB transaction failure, my consumer try to consume the message 10 times & after that throws "Backoff FixedBackOff {interval=0, currentAttempts=10, maxAttempts=9} exhausted for consumerDetails" exception.

 @Transactional
      @KafkaListener(topics = "xyz") 
      public void  consume(final ConsumerDetails consumerDetail) {
     System.out.println("received key: [" + consumerDetail.key() + "] and value: [" + consumerDetail.value() + "]"); System.out.println(
     "consumer processing done");
     throw new RuntimeException();
      }

Is there any way that after fixing db issue, I can make my consumer to consume same message from Kafka topic? or does this exception "Backoff FixedBackOff", means that I lost that message?

Upvotes: 0

Views: 839

Answers (1)

Hamide Ahadi
Hamide Ahadi

Reputation: 56

When you use the @Transactional annotation in your Kafka consumer and an exception occurs (like a database transaction failure), the message is not acknowledged, and the consumer will retry consuming the message according to the configured retry policy. However, if the maximum number of attempts is reached (in your case, 10 attempts), the consumer will stop processing that message and throw the Backoff FixedBackOff exception.

To handle this situation and ensure that you can retry consuming the same message after fixing the database issue, you can Use a Dead Letter Topic (DLT): Configure a Dead Letter Topic where failed messages can be sent after exhausting the retries. This way, you can later inspect the messages in the DLT and reprocess them if needed.

Here’s an example of how to configure a Dead Letter Topic:

@Bean
public ConcurrentKafkaListenerContainerFactory<ConsumerDetails> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<ConsumerDetails> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.setRetryTemplate(retryTemplate());
    factory.setErrorHandler(new SeekToCurrentErrorHandler(new DeadLetterPublishingRecoverer(kafkaTemplate), 
        new FixedBackOff(1000L, 3))); // Customize backoff and retries
    return factory;
}

Upvotes: 0

Related Questions