DefaultErrorHandler is not able to seek in case of an exception during the commit

[frosiere]

The listener error handler (DefaultErrorHandler) is not able to seek in case of an exception during the commit.

The code is not protected against exceptions such as a RebalanceInProgressException. The full list of exceptions can be found on KafkaConsumer#commitSync and KafkaConsumer#commitAsync methods.

Due to this bug, some records are skipped during a reprocessing.

The issue can be fixed by wrapping the commit into a try block with a finally block for the seek part.

See FailedBatchProcessor#seekOrRecover for more details.

The following portion of code

if (offsets.size() > 0) {
    commit(consumer, container, offsets);
}
if (isSeekAfterError()) {
...

shoud/could be replaced by

try {
    if (offsets.size() > 0) {
        commit(consumer, container, offsets);
    }
} finally {
    if (isSeekAfterError()) {
...

The proposed fix has been tested and is working fine.

I can contribute by dropping a PR, I can also provide more details if needed.

Any other ideas, suggestions are more than welcome.

[sobychacko]

@frosiere A PR is greatly welcomed.

[frosiere]

Great, I will drop a PR this week, sorry for the delay.
The try finally blocks are more of a workaround.
The real solution would be to rework the code as follows:

• if seekAfterError is true, seek+commit instead of commit+seek+commit
• if seekAfterError is false, commit

This would optimize and solve the problem.

[frosiere]

In the end, I had to keep the workaround, as the proposal would require an important rework of this layer. The redesign would require further discussion on this topic.

From a general point of view, it could be summarized as follows

In case of exception:

• Seek or recover
• Commit the offsets

This is not possible with the current implementation, as the seekOrRecover method throws a KafkaException after the seek, preventing the commit method from being called. Commits are also handled in 3 different places, whereas they should only be handled in one place after the seek or recover.

It seems that the FailedBatchProcessor#seekOrRecover and SeekUtils codes are trying to handle too many cases, complicating all the code around the seek or recover.

Any comments or feedback would be appreciated.

[sobychacko]

I am okay with the workaround (for now), as we want to avoid a major refactor in this area. The PR with the workaround looks good.

[frosiere]

I'm sorry to ask about a closed issue, but my team is still using Spring Kafka 2.9.13 and suffering from this problem.
Would it be possible to drop another PR to fix the issue on Spring Kafka 2.9.x?
Thank you for your support.

[sobychacko]

@frosiere, the 2.9.x version is already out of OSS support, and there won't be another OSS release for 2.9.x. See the support timeline here: https://spring.io/projects/spring-kafka#support. Even 3.0.x will go out of OSS support in a few months time. Any chance you can upgrade to the latest GA -3.1.x?