[
https://issues.apache.org/jira/browse/CAMEL-25683?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Work on CAMEL-25683 started by shashank.
----------------------------------------
> 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
> Priority: Major
>
> {{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)
