Custom CommonErrorHandler, ackMode = MANUAL_IMMEDIATE, offset not committed after recovery. Bug / by design?

[mkocsar]

I have created a custom CommonErrorHandler whose seeksAfterHandling returns false. I have ackMode = MANUAL_IMMEDIATE and enable.auto.commit = false. After handleOne returns true indicating that the record was recovered and the subsequent isAckAfterHandle call also returns true, I would expect the offset of the record to be committed, but it isn't. Is this a bug or by design?

Though I don't have a comprehensive understanding of KafkaMessageListenerContainer, I wonder if ackCurrent in commitOffsetsIfNeededAfterHandlingError should be called with this.isAnyManualAck instead of this.isManualAck.

If it is intentional that the offset is not committed, what would be the idiomatic way of getting it committed after successful error recovery in this setup?

[nujaadeen]

The root cause is in commitOffsetsIfNeededAfterHandlingError, it guards the ackCurrent() call with this.isManualAck, which is only true for AckMode.MANUAL. Since MANUAL_IMMEDIATE sets this.isManualImmediateAck (not isManualAck), the commit is silently skipped even when recovery succeeds.

The fix is straightforward: replace this.isManualAck with this.isAnyManualAck in that guard, which already correctly combines both modes everywhere else in the codebase.

Workaround until fixed: commit the offset directly inside handleOne using the Consumer parameter:

javaconsumer.commitSync(Collections.singletonMap(
    new TopicPartition(record.topic(), record.partition()),
    new OffsetAndMetadata(record.offset() + 1)
));
return true;

If you're using DefaultErrorHandler instead of a custom one, setCommitRecovered(true) achieves the same thing.