Losing messages in RetryingBatchErrorHandler/ErrorHandlingUtils during rebalancing

[vooft]

In what version(s) of Spring for Apache Kafka are you seeing this issue?

For example:

2.3.7+ (since introduction of RetryingBatchErrorHandler)

Describe the bug

If a listener doing a retry and rebalancing kicks in, then the consumer will be unpaused and every keep-alive consumer.poll() invocation in the error handler will start returning records and the error handler will throw them away.

spring-kafka/spring-kafka/src/main/java/org/springframework/kafka/listener/ErrorHandlingUtils.java

Line 69 in e4d9641

consumer.poll(Duration.ZERO);

consumer.pause(consumer.assignment());
try {
    while (nextBackOff != BackOffExecution.STOP) {
        consumer.poll(Duration.ZERO); // after rebalancing this poll will start returning records
        try {
...

It seems that the rebalancing listener KafkaMessageListenerContainer can re-pause the consumer again, but it is not aware that the error handler paused it.

To Reproduce

Create a listener with infinite retry RetryingBatchErrorHandler that will be constantly failing and force rebalancing. After rebalancing finished RetryingBatchErrorHandler will drain all the messages from assigned partitions without any processing.

Expected behavior

RetryingBatchErrorHandler should re-pause consumer after rebalancing.

Sample

Test is using RetryingBatchErrorHandler, but logic in ErrorHandlingUtils is essentially the same
https://github.com/vooft/kafka-retry-issue/blob/master/src/test/java/com/example/kafkaissue/KafkaIssueTest.java

[garyrussell]

Thanks for reporting; this is indeed a bug.

Please note that the RecoveringBatchErrorHandler is the preferred mechanism (pre 2.8) and its functionality is the default in the DefaultErrorHander since 2.8. With that mechanism, the listener throws a BatchListenerFailedException to indicate which record in the batch failed.

[vooft]

Thank you for your answer @garyrussell
I'm looking at DefaultErrorHander and it seems that if you just throw an arbitrary exception from the listener it will still delegate the handling to RetryingBatchErrorHandler, not RecoveringBatchErrorHandler, or am I missing something?

In spring-kafka 2.8.x:

Default fallback is FallbackBatchErrorHandler
https://github.com/spring-projects/spring-kafka/blob/2.8.x/spring-kafka/src/main/java/org/springframework/kafka/listener/DefaultErrorHandler.java#L93

And FallbackBatchErrorHandler extends RetryingBatchErrorHandler
https://github.com/spring-projects/spring-kafka/blob/2.8.x/spring-kafka/src/main/java/org/springframework/kafka/listener/FallbackBatchErrorHandler.java#L32

[garyrussell]

Yes; I agree; I was just pointing out that the recovering mechanism is preferred to the (older) retrying mechanism that suffers from this bug. I will have a fix soon.

[vooft]

Gotcha, thank you!