davsclaus opened a new pull request, #24686: URL: https://github.com/apache/camel/pull/24686
## Summary _Claude Code on behalf of davsclaus_ Fixes the fifth bug reported in [CAMEL-23994](https://issues.apache.org/jira/browse/CAMEL-23994): in batching mode with manual async commit, calling `manual.commit()` only recorded the offset in the in-memory `OffsetCache` but the cache was never flushed to Kafka. **Root cause:** `DefaultKafkaManualAsyncCommit.commit()` calls `commitManager.recordOffset()` which stores in `OffsetCache`. The cache is flushed to Kafka only when `commitManager.commit(partition)` is called. The streaming facade (`KafkaRecordStreamingProcessorFacade`) calls this after processing each partition's records, but the batching facade (`KafkaRecordBatchingProcessorFacade`) never did. **Impact:** With `batching=true` + `allowManualCommit=true` + `DefaultKafkaManualAsyncCommitFactory` + `autoCommitEnable=false`, offsets reached Kafka only on graceful partition revocation. On crash, the group offset was still at the initial position, causing the entire history since startup to be redelivered. **Fix:** The batching facade now flushes recorded offsets for each partition after `processExchange()`, mirroring the streaming facade's behavior. Note: `DefaultKafkaManualSyncCommit` was not affected since it calls `forceCommit()` which bypasses the cache and directly calls `consumer.commitSync()`. ## Test plan - [x] All 141 unit tests pass - [ ] CI integration tests will validate the batching + manual commit path 🤖 Generated with [Claude Code](https://claude.com/claude-code) 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]
