Repository navigation
[backport camel-4.22.x] CAMEL-25781: camel-azure-eventhubs - an event that completes when the checkpoint timer of its window has run is checkpointed by the next timer - #28207
Conversation
… that completes when the checkpoint timer of its window has run is checkpointed by the next timer
gnodet
added
the
backport
davsclaus
left a comment
davsclaus
left a comment
There was a problem hiding this comment.
Thanks Guillaume. This backports the CAMEL-25781 fix to camel-4.22.x. The CAMEL-25781 part matches #28135, and EventHubsConsumerCheckpointWindowTest is identical. But camel-4.22.x does not have CAMEL-25683 (#27922, main only), and the conflict resolution brought in half of it. EventHubsCheckpointUpdaterTask.run() now takes the count with getAndSet(0) and adds it back when the checkpoint fails. processCommit still checks cnt == getConfiguration().getCheckpointBatchSize().
-
Blocking. Because of the add back, the count can jump past the batch size after a failed timer checkpoint. For example, with a batch size of 500, 250 counted plus 300 added back gives 550. After that,
==never matches, so no checkpoint by batch size happens until the next timer resets the count. Main uses>=for exactly this reason. Please bring the consumer part of CAMEL-25683 too:- change the check to
>= - remove the cancelled task from
checkpointTasksByPartitionandscheduledTasksByPartition. With theisDone()check this is redundant, but it keeps the file the same as main. - add
EventHubsConsumerCheckpointAfterBatchSizeTest
CAMEL-25683 is a bug fix for events that are delivered again after a restart, so backporting it fits the backport policy. Please also mention it in the PR description and add 4.22.3 to its fix versions.
- change the check to
-
Not blocking. The upgrade note sits under "Upgrading Camel 4.21 to 4.22". 4.22.2 is already tagged, so this fix ships in 4.22.3 and needs a new "Upgrading from 4.22.2 to 4.22.3" section. The same entry also has to go into
camel-4x-upgrade-guide-4_22.adocon main, which already has that section.
Leaving out the DST test change from #28135 is correct, because camel-test-dst is not on 4.22.x. CI is still running.
Claude Code on behalf of davsclaus. This review was generated by an AI agent and may contain inaccuracies. Please verify all suggestions before applying.
|
🌟 Thank you for your contribution to the Apache Camel project! 🌟 🐫 Apache Camel Committers, please review the following items:
|
|
🧪 CI tested the following changed modules:
🔬 Scalpel shadow comparison — Scalpel: 10 tested, 28 compile-only — current: 9 all testedMaveniverse Scalpel detected 38 affected modules (current approach: 9).
|
gnodet-bot
left a comment
gnodet-bot
left a comment
There was a problem hiding this comment.
This backport introduces two issues that need to be fixed before it can be merged.
1. Missing == → >= fix from CAMEL-25683
This cherry-pick adds the processedEvents.addAndGet(processed) error-recovery path in EventHubsCheckpointUpdaterTask.run(), which is correct. However, the batch-size check in EventHubsConsumer.processCommit() still uses cnt == (line 253 on the PR branch) rather than cnt >= as on main (line 254).
The == vs >= gap was already fixed on main by CAMEL-25683 (commit 83aaa80) with the comment "a failed timeout checkpoint adds its count back, so the count can pass the batch size". When the error-recovery path does addAndGet(processed), the counter can overshoot checkpointBatchSize, making the == condition permanently false and silently disabling batch-size checkpoints. This backport needs the same == → >= fix from CAMEL-25683.
2. Upgrade guide placed under wrong version section
The new upgrade note is added as a === subsection under == Upgrading Camel 4.21 to 4.22 (the general 4.22 section). Since 4.22.2 is already tagged, this fix ships in 4.22.3. The established convention in this file is per-patch-level sections at the top (e.g. == Upgrading from 4.22.0 to 4.22.1 at line 14). A new == Upgrading from 4.22.2 to 4.22.3 section should be added at the top of the file instead.
This review was generated by an AI agent, Hermès on behalf of @gnodet.
oscerd
left a comment
oscerd
left a comment
There was a problem hiding this comment.
TL;DR
The CAMEL-25781 part matches #28135 and CI is green; the open point from @davsclaus (the half-imported CAMEL-25683: cnt == at EventHubsConsumer.java:253 next to the new add-back at EventHubsCheckpointUpdaterTask.java:88) is still unchanged at 7d65de0, and I have one more reason for the map removal he asked for.
Blockers
- None new from my side; the earlier "changes requested" items (
>=, removal of the cancelled task from both maps,EventHubsConsumerCheckpointAfterBatchSizeTest, the 4.22.2 → 4.22.3 guide section, the same note on main's 4_22 guide) are all still open at this head.
Other findings: 1 (non-blocking)
One more reason for that map removal, beyond keeping the file equal to main. With isDone() replacing isExpired(), a timer that scheduledTask.cancel(false) stops before it starts never runs. So on this branch, checkpointTask.cancel() (EventHubsConsumer.java:258) is the only thing that marks the window done after a checkpoint by batch size. If that line were ever dropped or moved, the next events of the partition would keep joining a task that never runs, and the partition would not get a timeout checkpoint again. On main, removing the task from checkpointTasksByPartition is a second guard.
From a static trace, none of the three tests in EventHubsConsumerCheckpointWindowTest would fail here without that cancel(): the mocked ScheduledFuture ignores the cancel, so timerStartedBeforeABatchSizeCheckpointDoesNotTakeTheNextWindow passes by letting the first timer checkpoint third. EventHubsConsumerCheckpointAfterBatchSizeTest#eventAfterBatchSizeCheckpointIsCheckpointedByTimeout uses a real scheduler and covers this path; it ports without pom changes (Awaitility comes from components/pom.xml), and with the earlier point applied both production files become byte-identical to main.
Reviewed with Claude Code on behalf of oscerd. This review was generated by an AI agent and may contain inaccuracies; please verify all suggestions before applying.
- Change batch-size check from == to >= (a failed timeout checkpoint adds its count back, so the count can pass the batch size) - Remove cancelled task from checkpointTasksByPartition and scheduledTasksByPartition after a checkpoint by batch size - Add EventHubsConsumerCheckpointAfterBatchSizeTest - Move upgrade note to a new '4.22.2 to 4.22.3' section
davsclaus
left a comment
davsclaus
left a comment
There was a problem hiding this comment.
Thanks Guillaume, my change request is addressed. The batch-size check is now >=, the cancelled task is removed from both per-partition maps, and EventHubsConsumerCheckpointAfterBatchSizeTest is added. EventHubsConsumer, EventHubsCheckpointUpdaterTask and the two new tests are now identical to main. The upgrade note sits in a new "Upgrading from 4.22.2 to 4.22.3" section, and #28238 adds the same text to the 4_22 guide on main.
Non-blocking: CAMEL-25683 and CAMEL-25781 still need 4.22.3 in their JIRA fix versions. That can be done at merge.
LGTM, once CI is green.
This review was generated by an AI agent (Claude Code on behalf of davsclaus) and may contain inaccuracies. Please verify all suggestions before applying.
gnodet-bot
left a comment
gnodet-bot
left a comment
There was a problem hiding this comment.
Following up on my earlier review — both issues have been addressed.
==→>=fix — ✅ addressed.cnt >= getConfiguration().getCheckpointBatchSize()is now inEventHubsConsumer.processCommit()with the correct comment explaining why.- Upgrade guide section — ✅ addressed. A new
== Upgrading from 4.22.2 to 4.22.3top-level section is placed correctly before== Upgrading from 4.22.1 to 4.22.2. - Map cleanup — ✅ addressed. Both
checkpointTasksByPartition.remove(partitionId)andscheduledTasksByPartition.remove(partitionId)are called after a batch-size checkpoint. EventHubsConsumerCheckpointAfterBatchSizeTest— ✅ added. Five tests with a real scheduler, includingbatchSizeCheckpointStillWorksAfterFailedTimeoutCheckpointwhich directly validates theaddAndGet+>=interaction.EventHubsConsumerCheckpointWindowTest— ✅ added. Three tests covering theisDone()-based window boundary logic with a mocked scheduler.
The locking design in EventHubsCheckpointUpdaterTask (shared consumer lock, done flag set atomically before getAndSet(0)) is correct: an event that completes while the timer holds the lock will see isDone() == true and start a fresh window, so the add-back in the error path cannot race with a concurrent batch-size checkpoint.
No new findings. LGTM.
This review was generated by an AI agent, Hermès on behalf of @gnodet.
4 participants
Backport of #28135
Cherry-pick of 727cab8 to camel-4.22.x.
Conflicts resolved automatically by backport bot agent.
Also brings the consumer part of CAMEL-25683 (#27922) to camel-4.22.x:
==to>=(a failed timeout checkpoint adds its count back, so the count can pass the batch size)checkpointTasksByPartitionandscheduledTasksByPartitionafter a checkpoint by batch sizeEventHubsConsumerCheckpointAfterBatchSizeTestThe upgrade guide note is placed in a new "Upgrading from 4.22.2 to 4.22.3" section.
The same note has been added to main's upgrade guide in #28238.