Repository navigation
CAMEL-25781: camel-azure-eventhubs - an event that completes when the checkpoint timer of its window has run is checkpointed by the next timer - #28135
Conversation
|
🌟 Thank you for your contribution to the Apache Camel project! 🌟 🐫 Apache Camel Committers, please review the following items:
|
davsclaus
left a comment
davsclaus
left a comment
There was a problem hiding this comment.
Thanks, the fix looks correct. A new window starts on isDone() instead of the wall clock. The timer takes the count and the context under the consumer's lock and checkpoints after releasing it. cancel() on the batch-size path stops a timer that is already waiting for the lock from checkpointing the next window. I traced the three new tests, and each one fails without the fix.
1. Stale DST Javadoc (medium, please include in this PR). The Event Hubs DST world is on main now (CAMEL-25738, #28067), so "not on main yet" in the PR description no longer holds. components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/dst/EventHubsCheckpointDstTest.java (lines 35-39) still describes this exact bug as a "Known residual on main" (seed 3042, isExpired() with >, the timer not taking the lock). Could you drop or rewrite that paragraph here, so it does not go stale on merge? It would also help to rebase and run -Ddst.seeds=5000 on main to confirm 5000/5000. If seed 3042 can be pinned, or the default seed count raised, the world covers this as a regression test too.
2. Header behaviour change (low, worth stating in the description and the upgrade guide). Before this change, CamelAzureEventHubsCheckpointUpdatedBy=timeout was practically unreachable from the else if (checkpointTask.isExpired()) branch, because an expired task was always replaced by a fresh one first. Now an event that completes after the scheduled time, but before a late timer has run, joins the old window and gets timeout. That is more accurate, but it is visible to routes that read the header.
Nits (optional):
doneinEventHubsCheckpointUpdaterTaskis only read and written under the lock, sovolatileis redundant (harmless).- The test calls the private
processCommitthrough reflection; making it package-private would be less brittle.
Claude Code on behalf of davsclaus. This review was generated by an AI agent and may contain inaccuracies. Please verify all suggestions before applying.
|
🧪 CI tested the following changed modules:
🔁 1 test passed only after a retry on JDK 21 (1 retried attempt) Recovered flaky tests on JDK 21 (1)
🔬 Scalpel shadow comparison — Scalpel: 8 of 707 tested, 26 compile-only — current: 8 all testedMaveniverse Scalpel detected 8 affected modules (current approach: 8). Skip-tests mode would test 8 modules (2 direct + 7 downstream), skip tests for 26 (generated code, meta-modules) Modules Scalpel would test (8)
Modules with tests skipped (26)
All tested modules (35 modules, 6m 1s total)Total reactor time: 6m 1s
Top 20 slowest modules:
|
95e2244 to
ac0af0f
Compare
|
Thanks for the review. The branch is rebased on main (ac0af0f, which has the Event Hubs DST world), and the follow-up is 0dc4a77. 1. DST Javadoc. I rewrote the paragraph in
2. Header. Added to the 4.24 upgrade guide: Nits. Both done: camel-azure-eventhubs tests: 1040 run, 0 failures. That includes the 1000 default DST seeds, the pinned seed and the replay check. Claude Code on behalf of allthingssecurity |
gnodet-bot
left a comment
gnodet-bot
left a comment
There was a problem hiding this comment.
The fix correctly closes three checkpoint-window boundary races by moving window transition from wall-clock time (isExpired()) to the isDone() state flag set under lock. Lock sharing between consumer and updater task is safe (reentrant lock; cancel() is called from within processCommit() which already holds the lock — no deadlock). Test coverage is solid: three targeted unit tests and DST seed 3042 pinned.
One minor nit on the visibility change:
This review was generated by an AI agent, Hermès on behalf of @gnodet.
gnodet-bot
left a comment
gnodet-bot
left a comment
There was a problem hiding this comment.
Fix is correct. The lock protocol — isDone()/cancel() under the consumer's ReentrantLock, with run() acquiring the same lock on the executor thread — closes all three races described in the JIRA. All four of @davsclaus's points are addressed. Two low-severity observations below.
port/camel-4.22.x at minimum, and is a strong candidate for port/camel-4.18.x (LTS) since Event Hubs consumers on the LTS branch are exposed to the same race.
This review was generated by an AI agent, Hermès on behalf of @gnodet.
davsclaus
left a comment
davsclaus
left a comment
There was a problem hiding this comment.
LGTM, thanks. Both earlier points are addressed: the DST Javadoc no longer describes this bug as a residual, and seed 3042 is pinned with @DstSeed, so the world covers it as a regression test. The upgrade guide now says that CamelAzureEventHubsCheckpointUpdatedBy can be timeout. Both nits are done too (done is no longer volatile, and processCommit is package-private). The three gnodet-bot points (the @param eventContext, the Javadoc on the 2-arg constructor, and the never()/times(1) assertions on the cancelled timer) are fixed, and I resolved those threads.
Claude Code on behalf of davsclaus. This review was generated by an AI agent and may contain inaccuracies. Please verify all suggestions before applying.
|
@allthingssecurity this is approved and CI is green, but it now conflicts with Claude Code on behalf of davsclaus. |
… checkpoint timer of its window has run is checkpointed by the next timer The consumer started a new checkpoint window when the wall clock had passed the scheduled time of the current one (EventHubsCheckpointUpdaterTask.isExpired(), which compares with >), and the timer task took the count of the window without the consumer's lock. So an event was counted in a window whose timer had already run when it completed in the millisecond of the scheduled time, or when the timer ran between the isExpired() check and the count of the event, or when a timer that had already started took the count after a checkpoint by batch size. The event was then checkpointed only with the next event of the partition; if the partition stayed quiet, it was delivered again after a restart or a rebalance. A timer that ran late (after the scheduled time) left a new window whose timer found nothing to checkpoint. A new window now starts when the timer of the current one has run (isDone()), not by the wall clock, and the timer takes the count and the event context under the consumer's lock. A checkpoint by batch size cancels the timer under the lock too, so a timer that has already started takes nothing. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
… note for the timeout header, review nits EventHubsCheckpointDstTest no longer describes this bug as a known residual on main: seed 3042, which failed before this fix, is pinned as its own run with @DstSeed. The 4.24 upgrade guide notes the fix and that CamelAzureEventHubsCheckpointUpdatedBy can now be timeout. done in EventHubsCheckpointUpdaterTask is read and written under the lock only, so it is no longer volatile; processCommit is package-private and the test calls it directly. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…e 2-arg updater task constructor, assert the cancelled timer does not checkpoint again Review suggestions: - processCommit: document the eventContext parameter. - EventHubsCheckpointUpdaterTask(EventContext, AtomicInteger): document that it uses a private lock of its own (standalone or test use); EventHubsConsumer uses the constructor that takes its lock. - EventHubsConsumerCheckpointWindowTest: after the timer of the first window runs, assert that first is never checkpointed and that second (the event context of that timer, checkpointed by batch size) is checkpointed once, so a cancelled timer that checkpoints again fails the test. With cancel() not marking the task done, the test fails with "Wanted 1 time, but was 2 times". Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
dbd7616 to
df2c63c
Compare
|
Thanks, rebased on the current Claude Code on behalf of allthingssecurity |
gnodet-bot
left a comment
gnodet-bot
left a comment
There was a problem hiding this comment.
Review: Fix concurrent checkpoint task execution race conditions
Verdict: LGTM ✅
The concurrency fix is well-designed. The ReentrantLock-based protocol correctly coordinates the three competing paths (timer-triggered run(), batch-size cancel(), and event counting), and all concurrency questions are non-issues:
cancel()called fromprocessCommit()under the same lock:ReentrantLockis reentrant by design — no deadlock possible.returninsidetry { ... } finally { lock.unlock(); }:finallyalways executes in Java regardless of how thetryblock exits.donefield withoutvolatile: All reads and writes are under the sameReentrantLock, which provides full visibility and happens-before guarantees —volatileis not needed.isDone()called without its own lock guard:isDone()is called exclusively fromprocessCommit(), which acquireslock.lock()at entry.eventContext volatileread inside the lock: Redundant but harmless.AtomicInteger.getAndSet(0)under lock: Pre-existing pattern, redundant but correct.
The 3 new tests cover the targeted race conditions, and the upgrade guide correctly documents the CamelAzureEventHubsCheckpointUpdatedBy: timeout behavior change.
port/camel-4.22.x and port/camel-4.18.x.
gnodet
added
port/camel-4.22.x
🔄 Backport BotConflicts (🤖 agent dispatched to resolve):
ℹ️ If you push additional commits after |
|
✅ Backport to |
|
✅ Backport to |
Description
CAMEL-25781
EventHubsConsumer.processCommitstarted a new checkpoint window when the wall clock had passed the scheduled time of the current one (EventHubsCheckpointUpdaterTask.isExpired(), which compares with>), and the timer task took the count of the window without the consumer's lock. So an event could be counted in a window whose timer had already run:isExpired()check and the count of the event;And when the timer ran late (after the scheduled time by the wall clock), the next event started a new window, the old timer took its count and checkpointed the previous event, and the new timer found nothing.
The event was then checkpointed only with the next event of the partition. If the partition stayed quiet, it was delivered again after a restart or a rebalance (a duplicate, not a loss). Using
>=alone would close case 1 but make the late-timer case wider.Change:
isDone()), not by the wall clock;isExpired()stays for theCHECKPOINT_UPDATED_BYheader. Behaviour change: the header can now betimeout. An event that completes after the scheduled time of its window, but before the late timer has run, joins that window and getstimeout. Before, an expired task was always replaced first, sotimeoutwas practically unreachable. Noted in the 4.24 upgrade guide (new sectioncamel-azure-eventhubs - an event that completes when the checkpoint timer of its window has run is checkpointed);doneis no longervolatile(read and written only under the lock).Tests:
EventHubsConsumerCheckpointWindowTest(3 tests, in the style of the CAMEL-25683 tests:processCommit, now package-private, called directly, a mocked scheduler whose captured timer tasks the test runs): the timer runs before its scheduled time by the wall clock and then an event completes; the timer runs late; a timer that had started before a checkpoint by batch size. Without the change all three fail: the first two withWanted but not invoked: eventContext.updateCheckpointAsync()for the last completed event, the third because the timer of the first window, cancelled by the checkpoint by batch size, checkpointsseconda second time (Wanted 1 time, but was 2 times) and the last completed event is not checkpointed. Case 2 cannot be shown without a hook between the check and the count; the same lock closes it.main), seed 3042 of 5000. The branch is rebased on main.EventHubsCheckpointDstTestno longer lists this bug as a known residual on main, and pins seed 3042 (@DstSeed(3042),eventThatCompletesWhenTheTimerOfItsWindowRanIsCheckpointed). With the production classes from main the pinned seed fails; with this PR it passes, and-Ddst.seeds=5000passes 5000 of 5000.What was not run: the full root build, and the integration tests (they need an Azure Event Hubs namespace).
Target
mainbranch)Tracking
Apache Camel coding standards and style
mvn clean install -DskipTestslocally from root folder and I have committed all auto-generated changes.(I built and tested camel-azure-eventhubs, including the formatter and import-sort plugins. I did not run the full root build.)
AI-assisted contributions
Co-authored-bytrailers) and the PR description identifies the AI tool used.This PR was prepared with Claude Code (Claude Opus 5.5). The commits carry a
Co-Authored-Bytrailer.Claude Code on behalf of allthingssecurity
🤖 Generated with Claude Code