[jira] [Updated] (CAMEL-25781) camel-azure-eventhubs - an event that completes when the checkpoint timer of its window has already run is not checkpointed until the next event of the partition (delivered again after a restart if the partition stays quiet)

["Claus Ibsen (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/CAMEL-25781?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Claus Ibsen updated CAMEL-25781:
--------------------------------
    Fix Version/s: 4.24.0

> camel-azure-eventhubs - an event that completes when the checkpoint timer of its window has already run is not checkpointed until the next event of the partition (delivered again after a restart if the partition stays quiet)
> --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
>
>                 Key: CAMEL-25781
>                 URL: https://issues.apache.org/jira/browse/CAMEL-25781
>             Project: Camel
>          Issue Type: Bug
>          Components: camel-azure
>            Reporter: shashank
>            Assignee: shashank
>            Priority: Minor
>             Fix For: 4.24.0
>
>
> {{EventHubsConsumer.processCommit}} (main bbd6f750aa00, lines 225-283) counts each completed event in the checkpoint window of its partition. The window is an {{EventHubsCheckpointUpdaterTask}} scheduled for {{checkpointBatchTimeout}}; when its timer runs, it takes the count and checkpoints the last event of the window. A new window is started when the wall clock has passed the scheduled time:
> {code:java}
> // EventHubsConsumer, line 234
> if (checkpointTask == null || checkpointTask.isExpired()) {
> // EventHubsCheckpointUpdaterTask, line 74
> return System.currentTimeMillis() > scheduledTime;
> {code}
> and the timer takes the count ({{processedEvents.getAndSet(0)}}, line 45 of the task) without the consumer's lock. So an event can be counted in a window whose timer has already run:
> # the event completes in the millisecond of the scheduled time, after the timer ran ({{>}}, not {{>=}}; the scheduler fires by {{nanoTime}} and the scheduled time is a truncated {{currentTimeMillis}}, so the timer often runs in that very millisecond);
> # the timer runs between the {{isExpired()}} check (line 234) and the {{setEventContext}} / count of the event (lines 244 and 248), because it does not take the lock;
> # a timer that has already started when a checkpoint by batch size cancels it ({{cancel(false)}}, line 255) takes the count of the next window and checkpoints an older event.
> There is also the other side of the same check: when the timer runs late (after the scheduled time by the wall clock: a GC pause, the single checkpoint thread busy, a wall clock that runs ahead of {{nanoTime}}), the next event starts a new window, the old timer takes its count and checkpoints the previous event, and the new timer finds nothing.
> In each case the event is not checkpointed by a timer. It is checkpointed with the next event of the partition. If the partition stays quiet, its offset is never checkpointed, and after a restart of the route or a rebalance of the partition to another consumer, it is delivered again from the last checkpoint: a duplicate, not a loss (the checkpoint is never moved past an event that was not completed; only the checkpoint of a completed event is missing). With {{>=}} alone the first window closes but the late-timer case gets wider, so the fix does not use the wall clock.
> The component docs only describe the two options ("Sets the batch timeout between each checkpoint update. Works jointly with checkpointBatchSize."): a completed event is expected to be checkpointed at the latest {{checkpointBatchTimeout}} after it completed.
> *Fix*: a new window starts when the timer of the current window has run ({{EventHubsCheckpointUpdaterTask.isDone()}}), not when the wall clock passed its time. The timer takes the count and the event context under the consumer's lock and marks itself done, so an event is either counted before (and checkpointed by that timer) or after (and starts a new window). A checkpoint by batch size marks the task done under the lock as well, so a timer that has already started takes nothing. {{isExpired()}} is kept for the {{CHECKPOINT_UPDATED_BY}} header.
> *Tests*: new {{EventHubsConsumerCheckpointWindowTest}} (3 tests, the style of the CAMEL-25683 tests: {{processCommit}} called directly, a mocked scheduler whose captured timer tasks the test runs). All three fail on main with {{Wanted but not invoked: eventContext.updateCheckpointAsync()}} for the event that completed last:
> * {{eventCompletedAfterTheTimerRanBeforeItsScheduledTimeIsCheckpointed}}: the timer runs before the wall clock passes its scheduled time (case 1), then an event completes; on main it is counted in the done window and no timer is scheduled;
> * {{eventCompletedAfterTheScheduledTimeBeforeTheTimerRanIsCheckpointed}}: the late timer; on main the old timer checkpoints the previous event and the new one finds 0;
> * {{timerStartedBeforeABatchSizeCheckpointDoesNotTakeTheNextWindow}}: case 3.
> Case 2 cannot be shown without a hook between the check and the count; it is closed by the same lock. With the fix: camel-azure-eventhubs unit tests 38 run, 0 failures.
> *How it was found*: the camel-test-dst world for Event Hubs (CAMEL-25738), seed 3042 of 5000, then the JUnit test above. On the camel-test-dst-stage1 branch seed 3042 fails without the fix (trace hash 1137160c..., "1 completed after the last checkpoint request ... counted for a window whose timer already ran (not yet expired)"); with the fix it passes, and 5000 of 5000 seeds pass.
> *Affected*: main, camel-4.23.x, camel-4.22.x and camel-4.18.x (the same {{isExpired()}} check and lock-free {{EventHubsCheckpointUpdaterTask}}), camel-4.14.x (the same {{>}} check against {{TimerTask.scheduledExecutionTime()}} of a {{java.util.Timer}}, without any lock). Read, not run.
> _Filed with Claude Code on behalf of allthingssecurity._



--
This message was sent by Atlassian Jira
(v8.20.10#820010)