shashank created CAMEL-25683:
--------------------------------
Summary: camel-azure-eventhubs - after a checkpoint by
checkpointBatchSize, the following events of the partition are not checkpointed
by checkpointBatchTimeout, so a restart or rebalance after a burst delivers
them again
Key: CAMEL-25683
URL: https://issues.apache.org/jira/browse/CAMEL-25683
Project: Camel
Issue Type: Bug
Components: camel-azure-eventhubs
Reporter: shashank
Assignee: shashank
{{EventHubsConsumer.processCommit}} (lines 225-278 at main cd2ce899764e) keeps,
per partition, a checkpoint task that a scheduler runs
{{checkpointBatchTimeout}} after the first event of a window, and checkpoints
at once when {{checkpointBatchSize}} events have completed. When the batch size
is reached it cancels the scheduled future (253-255) but leaves the task in
{{checkpointTasksByPartition}}. Until the time that task was scheduled for,
every following event of the partition finds it not expired (234) and only sets
its event context on it (244). The task is cancelled, so nothing checkpoints
these events by time. The first event that completes after that time starts a
new task, which then checkpoints all of them.
If no event of the partition completes after that time - the end of a burst -
the events completed after the size checkpoint are not checkpointed until the
next event arrives, which can be hours later. A restart of the application, or
the partition moving to another consumer of the group (load balancing when
instances are added or removed), starts from the last checkpoint and delivers
them again: up to {{checkpointBatchSize - 1}} events per partition.
Example with the defaults (batch size 500, timeout 5000 ms): a burst of 700
events within 5 seconds, then quiet. Events 1-500 are checkpointed by size;
events 501-700 stay uncheckpointed until the next event of that partition.
Before CAMEL-24198 (one task for the consumer) the same happened (4.18.x:
{{lastTask}} / {{lastScheduledTask}}).
A second, smaller cause of the same symptom:
{{EventHubsCheckpointUpdaterTask.run()}} (lines 43-50) resets the per-partition
counter with {{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 completes while
that checkpoint is being stored (the round trip to the checkpoint store) and
starts the next window is wiped from the count; the next task then finds 0 and
skips its checkpoint.
h3. Reproduction
New {{EventHubsConsumerCheckpointAfterBatchSizeTest}} (3 tests, the style of
{{EventHubsConsumerPerPartitionCheckpointTest}}: mocked {{EventContext}},
{{processCommit}} called directly, but a real single-thread scheduler; batch
size 2, timeout 1000 ms):
* events 1 and 2 (checkpoint by size at event 2), then event 3 and nothing
else: event 3 must be checkpointed within 10 seconds. On main: {{Wanted but not
invoked: eventContext.updateCheckpointAsync()}} after 10 s.
* events 1-5 (checkpoints by size at 2 and 4): event 5 must be checkpointed. On
main: the same.
* batch size 10: event 1 is checkpointed by the timeout, whose {{Mono}}
completes only after event 2 has completed (and started the next window): event
2 must be checkpointed by the next timeout. On main: the same.
h3. Proposed fix
When the batch size is reached, remove the task and its future from the two
maps after cancelling the future, so the next event of the partition starts a
new timeout window, as the first event after start does. In the timer task,
reset the counter only if it still holds the value that was checkpointed
({{compareAndSet}}); otherwise it stays a little high, which only makes the
next size checkpoint come earlier. A partition can get at most one more timer
checkpoint per size checkpoint than before. Otherwise the size checkpoint, the
timer task and the header {{CamelAzureEventHubsCheckpointUpdatedBy}} work as
before. No option or default changes; the timer checkpoints at most once per
window, as documented ("Works jointly with checkpointBatchSize"), so no upgrade
guide entry.
camel-azure-eventhubs unit tests with the fix: 33 run, 0 failures. The
integration tests (Azure) were not run.
How it was found: the backlog of an earlier review round (reading of
{{processCommit}}); the user impact was then worked out from the code and the
Event Hubs checkpoint semantics (a new owner of a partition starts at its
checkpoint), and shown with the test.
Affected: main (cd2ce899764e; tested), camel-4.23.x and camel-4.22.x (the same
per-partition code), camel-4.18.x ({{lastTask}} / {{lastScheduledTask}}, same
shape) and camel-4.14.x ({{java.util.Timer}} task, same shape) - read, not run.
Duplicate check (2026-10-15): JIRA text "eventhubs checkpoint batch",
"eventhubs checkpoint timeout", "checkpointBatchTimeout" (CAMEL-24198
per-partition batching, CAMEL-24158 umbrella, CAMEL-17841 add batching: none
about the cancelled task); GitHub pull requests "EventHubsConsumer checkpoint",
"checkpointBatchTimeout" (#24903 = CAMEL-24198, #18957, #25247: other changes).
_Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
