allthingssecurity opened a new pull request, #27922: URL: https://github.com/apache/camel/pull/27922
# Description [CAMEL-25683](https://issues.apache.org/jira/browse/CAMEL-25683) `EventHubsConsumer.processCommit` keeps, per partition, a checkpoint task that the scheduler runs `checkpointBatchTimeout` after the first event of a window, and checkpoints at once when `checkpointBatchSize` events have completed. When the batch size was reached it cancelled the scheduled future but left the task registered. Until the time that task had been scheduled for, the following events of the partition found it "not expired" and only set their event context on it, so nothing checkpointed them by time. If no event of the partition completed after that time (the end of a burst), those events stayed uncheckpointed until the next event, and a restart or the partition moving to another consumer delivered them again (up to `checkpointBatchSize - 1` per partition; with the defaults, a burst of 700 events within 5 seconds leaves events 501-700 uncheckpointed). This change removes the task and its future from the two maps when the batch size is reached, after cancelling the future, so the next event of the partition starts a new timeout window, as the first event after start does. The size checkpoint, the timer task and the `CamelAzureEventHubsCheckpointUpdatedBy` header work as before. A partition can now get at most one more timer checkpoint per size checkpoint (one Blob write) than before. A second, smaller cause of the same symptom is fixed with it: `EventHubsCheckpointUpdaterTask.run()` reset the per-partition counter (`processedEvents.set(0)`) in the completion callback of its checkpoint, on a reactor thread and without the consumer's lock. The counter is shared with the next task of the partition, so an event that completed while that checkpoint was being stored (and started the next window) was wiped from the count, and the next task then found 0 and skipped its checkpoint. The callback now resets the counter only if it still holds the value the task checkpointed (`compareAndSet`); otherwise the count stays a little high, which only makes the next size checkpoint come earlier. No upgrade guide entry: no option or default changes, and the timer checkpoints at most once per window as the option documentation says ("Works jointly with checkpointBatchSize"); users only see the missing checkpoints made. Tests: - New `EventHubsConsumerCheckpointAfterBatchSizeTest` (3 tests, in the style of `EventHubsConsumerPerPartitionCheckpointTest`: mocked `EventContext`, `processCommit` called directly, but with a real single-thread scheduler; batch size 2, timeout 1000 ms): an event after a size checkpoint, and an event after two size checkpoints, must be checkpointed by the timeout when no other event follows; and an event that completes while the timeout checkpoint of the previous window is still being stored (its `Mono` completes when the test says so) must be checkpointed by the next timeout. Without the change each fails after 10 s with `Wanted but not invoked: eventContext.updateCheckpointAsync()` (the first two without the change in `processCommit`, the third without the change in `EventHubsCheckpointUpdaterTask`). - camel-azure-eventhubs unit tests with the change: 33 run, 0 failures. The integration tests (Azure) were not run. # Target - [x] I checked that the commit is targeting the correct branch (Camel 4 uses the `main` branch) # Tracking - [ ] If this is a large change, bug fix, or code improvement, I checked there is a [JIRA issue](https://issues.apache.org/jira/browse/CAMEL) filed for the change (usually before you start working on it). (The ticket is drafted and not filed yet; CAMEL-25683 is a placeholder until it is.) # Apache Camel coding standards and style - [x] I checked that each commit in the pull request has a meaningful subject line and body. - [ ] I have run `mvn clean install -DskipTests` locally from root folder and I have committed all auto-generated changes. (I built and tested the affected module, including the formatter and import-sort plugins. I did not run the full root build.) # AI-assisted contributions - [x] If this PR includes AI-generated code, commits have proper co-authorship attribution (e.g., `Co-authored-by` trailers) and the PR description identifies the AI tool used. This PR was prepared with Claude Code (Claude Opus 5.5). The commit carries a `Co-Authored-By` trailer. _Claude Code on behalf of allthingssecurity_ 🤖 Generated with [Claude Code](https://claude.com/claude-code) -- 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]
