[
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)
