davsclaus opened a new pull request, #24688: URL: https://github.com/apache/camel/pull/24688
_Claude Code on behalf of davsclaus_ ## Summary Fixes **Bug 3** from [CAMEL-23994](https://issues.apache.org/jira/browse/CAMEL-23994): batching auto-commit over-commits unprocessed record offsets. **Root cause:** In batching mode, `CommitSynchronization.onComplete()` called the no-arg `commitManager.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 `TopicPartition` from the actual batch exchanges (via their `KafkaConstants.TOPIC`, `PARTITION`, and `OFFSET` headers), then explicitly call `commitManager.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 - Added `computeBatchOffsets(List<Exchange>)` method that extracts max offset per TopicPartition from batch exchange headers - Modified `CommitSynchronization` inner class to accept and use explicit batch offsets instead of committing the consumer's current position - Updated `autoCommitResultProcessing()` to compute and pass batch offsets to `CommitSynchronization` ## Test plan - [x] All 141 existing unit tests pass (`mvn test` in camel-kafka module) - [ ] CI integration tests (Kafka testcontainers) Co-Authored-By: Claude Opus 4.6 <[email protected]> -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
