Spring Kafka Auto Commit Offset In Case of Failures

spring apache-kafka spring-kafka
rishi picture rishi · Mar 21, 2018 · Viewed 7.6k times · Source

I am using Spring Kafka 1.2.2.RELEASE. I have a Kafka Listener as consumer that listens to a topic and index the document in elastic. My Auto commit offset property is set to true //default.

I was under the impression that in case there is an exception in the listener(elastic is down) the offsets should not be committed and the same message should be processed for the next poll

However this is not happening and the consumer commits the offset on the next poll.After reading posts and documentation i learnt that this is the case that with auto commit set to true to next poll will commit all offset

My doubt is why is the consumer calling the next poll and also how can i prevent any offset from committing with auto commit to true or do i need to set this property to false and commit manually.

Answer

Gary Russell picture Gary Russell · Mar 21, 2018

I prefer to set it to false; it is more reliable for the container to manage the offsets for you.

Set the container's AckMode to RECORD (it defaults to BATCH) and the container will commit the offset for you after the listener returns.

Also consider upgrading to at least 1.3.3 (current version is 2.1.4); 1.3.x introduced a much simpler threading model, thanks to KIP-62

EDIT

With auto-commit, the offset will be committed regardless of success/failure. The container won't commit after a failure, unless ackOnError is true (another reason to not use auto commit).

However, that still won't help because the broker won't send the same record again. You have to perform a seek operation on the Consumer for that.

In 2.0.1 (current version is 2.1.4), we added the SeekToCurrentErrorHandler which will cause the failed and unprocessed records to be re-sent on the next poll. See the reference manual.

You can also use a ConsumerAwareListener to perform the seek yourself (also added in 2.0).

With older versions (>= 1.1) you have to use a ConsumerSeekAware listener which is quite a bit more complicated.

Another alternative is to add retry so the delivery will be re-attempted according to the retry settings.

Related questions

Spring Boot Kafka: Unable to start consumer due to NoSuchBeanDefinitionException

With in my spring boot service while trying to start a kafka consumer seeing NoSuchBeanDefinitionException and unable to start the service itself. Below is my bean class which has all the required beans created for Kafka configuration Spring Boot Version: 1.5.2.…

Read More
Spring Kafka - How to reset offset to latest with a group id?

I am currently using Spring Integration Kafka to make real-time statistics. Though, the group name makes Kafka search all the previous values the listener didn't read. @Value("${kafka.consumer.group.id}") private String consumerGroupId; @Bean public ConsumerFactory<String, String&…

Read More
Spring Kafka Producer not sending to Kafka 1.0.0 (Magic v1 does not support record headers)

I am using this docker-compose setup for setting up Kafka locally: https://github.com/wurstmeister/kafka-docker/ docker-compose up works fine, creating topics via shell works fine. Now I try to connect to Kafka via spring-kafka:2.1.0.RELEASE When starting up the …

Read More