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.