I have a spring boot application where we are using spring Kafka in consumer section i have made enable.auto.commit to false and set my listener ack-mode to manual_immediate
I have concurrent consumers so after consuming the record I call acknowledgment.acknowledge() but here i still face the duplicate issue problem whenever rebalance happens other consumer start consuming the same message which is already consumed by one consumer. Any idea what magic is happening behind the scene.
Anyone know when using manual_immediate does it commit message by commitSync or commitAsync ? is there way we can change the behaviour to avoid duplicates record message reading. Is there a way we can use hybrid model in Spring Kafka
In spring boot Kafka is there a way we can see whenever a rebalance happen i can log it.
How to create rebalance if we want to do it for some testing purpose?
Whenever you call acknowledge on the listener thread, it will use commitSync() by default; use the syncCommits wrapper property to use asynchronous commits.
If you call it on a different thread, the commit is queued for processing by the consumer thread as soon as possible.
Duplicates cannot be avoided if a forced rebalance occurs because your listener took too long to process the records received by poll() .
You can increase max.poll.interval.ms and/or decrease max.poll.records to make sure you can process records in time.
You can add a ConsumerRebalanceListener to the container properties to register rebalances.
Reduce max.poll.interval.ms to a small value to reproduce it in a test.
First, regardless of the acknowledgment mode, a message is never guaranteed to be consumed only once. For example, a rebalancing can occur between the time a message is consumed and the time offset is committed, causing Kafka to deliver the message again to the newly allocated consumer. It is the responsibility of the applications to be idempotent against duplicate messages.
To listen for rebalancing events, an implementation of ConsumerRebalanceListener is needed. You can connect this implementation to Spring's automatically configured ConcurrentKafkaListenerContainerFactory instance. A more detailed description of how this can be done has already been answered here .
If you want to create a forced rebalancing for the test, you can do so by killing one of the 1+ existing consumers. If you use spring-kafka, you can do this using an @Autowired instance of KafkaListenerEndpointRegistry and kill/rest/(re)start any consumers. Something about this should do:
@Autowired KafkaListenerEndpointRegistry registry; public void myTest() { Collection<MessageListenerContainer> containers = registry.getAllListenerContainers() containers.get(0).stop() }