shashank created CAMEL-25781:
--------------------------------
Summary: 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
{{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)
