CAMEL-23994: Fix batching auto-commit committing unprocessed record offsets#24688
Conversation
…ffsets In batching mode, CommitSynchronization.onComplete() called the no-arg commitManager.commit() which commits the consumer's current position. Because Kafka pre-fetches, this position is ahead of what was actually processed, silently skipping records on restart. Fix: compute the max offset per TopicPartition from the batch exchanges and explicitly recordOffset + commit for each partition, so only the offsets of actually-processed records are committed. Signed-off-by: Claus Ibsen <davsclaus@apache.org> Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> Signed-off-by: Claus Ibsen <claus.ibsen@gmail.com>
|
🌟 Thank you for your contribution to the Apache Camel project! 🌟 🐫 Apache Camel Committers, please review the following items:
|
gnodet
left a comment
gnodet
left a comment
There was a problem hiding this comment.
Correct fix. The old commitManager.commit() (no-arg) committed the Kafka consumer's current position, which can be ahead of what was actually delivered to the processor due to Kafka's internal pre-fetching. On restart, unprocessed records between the last batch and the committed position would be silently lost.
The fix correctly:
computeBatchOffsets()extracts max offset perTopicPartitionfrom the actual exchange headers (TOPIC,PARTITION,OFFSET) usingoffsets.merge(tp, offset, Math::max)— only what was actually in the batchcommitBatchOffsets()does explicitrecordOffset()+commit(partition)per partition — mirroring how the streaming facade commits
Both the onComplete() and breakOnFirstError paths in CommitSynchronization are updated. The CommitSynchronization constructor now captures the batch offsets at creation time, ensuring the offsets are fixed before the processor runs (correct snapshot semantics).
Reviewed with Claude Code on behalf of gnodet. This review was generated by an AI agent and may contain inaccuracies; please verify all suggestions before applying.
|
🧪 CI tested the following changed modules:
🔬 Scalpel shadow comparison — Scalpel: 9 tested, 29 compile-only — current: 9 all testedMaveniverse Scalpel detected 38 affected modules (current approach: 9).
|
2 participants
Claude Code on behalf of davsclaus
Summary
Fixes Bug 3 from CAMEL-23994: batching auto-commit over-commits unprocessed record offsets.
Root cause: In batching mode,
CommitSynchronization.onComplete()called the no-argcommitManager.commit()which commits the Kafka consumer's current position. Because Kafka pre-fetches records ahead of what was actually delivered to the processor, the committed offset jumps past records that haven't been processed yet. On restart, those skipped records are silently lost.Fix: Compute the maximum offset per
TopicPartitionfrom the actual batch exchanges (via theirKafkaConstants.TOPIC,PARTITION, andOFFSETheaders), then explicitly callcommitManager.recordOffset()+commitManager.commit(partition)for each partition. This ensures only offsets of actually-processed records are committed — matching the behavior of the streaming facade.Changes
computeBatchOffsets(List<Exchange>)method that extracts max offset per TopicPartition from batch exchange headersCommitSynchronizationinner class to accept and use explicit batch offsets instead of committing the consumer's current positionautoCommitResultProcessing()to compute and pass batch offsets toCommitSynchronizationTest plan
mvn testin camel-kafka module)Co-Authored-By: Claude Opus 4.6 noreply@anthropic.com