[gnodet]

Backport of #28135 to camel-4.18.x

This is not a straight cherry-pick of #28135 (CAMEL-25781). Because camel-4.18.x did not have
CAMEL-24198, CAMEL-25683, CAMEL-24159, or CAMEL-20199, conflict resolution brought forward the
relevant Event Hubs parts of all four. Without them, CAMEL-25781 has no sound base to build on.

What this PR brings

• CAMEL-25781 (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): A new checkpoint window starts when the timer of the current one has
  run (EventHubsCheckpointUpdaterTask.isDone()), not when the wall clock passes the scheduled
  time. Fixes events being re-delivered after a restart when a partition stays quiet.

• CAMEL-24198 (03cb654): Per-partition event counters (processedEventsByPartition) and
  per-partition timers (checkpointTasksByPartition, scheduledTasksByPartition). The consumer
  lock (ReentrantLock) is introduced so that processCommit runs under a lock. Previously the
  shared processedEvents counter mixed all partitions and a checkpoint on one partition reset the
  count for others.

• CAMEL-25683 (CAMEL-25683: camel-azure-eventhubs - after a checkpoint by batch size, the next events are checkpointed by the batch timeout again #27922): The timer takes its count when it starts and adds it back when it
  fails. The batch-size check uses >= (handles the race where addAndGet overshoots before the
  reset). A cancelled task is removed from the per-partition map.

• CAMEL-24159 (cd82682, CAMEL-24159: camel-azure-eventhubs - fix medium-severity bugs from code review #24867): volatile event context in the task. Failed checkpoint
  logging promoted from DEBUG to WARN with the stack trace.

• CAMEL-20199 (a52096e, CAMEL-20199: Replace synchronized with ReentrantLock in misc components #25247): The ReentrantLock that replaced the synchronized
  keyword introduced by CAMEL-24159.

After this PR, EventHubsConsumer and EventHubsCheckpointUpdaterTask are byte-identical to main.

Tests

• EventHubsConsumerCheckpointWindowTest — covers CAMEL-25781 (already present)
• EventHubsConsumerPerPartitionCheckpointTest — covers CAMEL-24198 per-partition isolation
• EventHubsConsumerCheckpointAfterBatchSizeTest — covers end-of-burst checkpointing

[gnodet-bot]

Good fix — this correctly replaces the wall-clock-based window boundary (isExpired) with an execution-based boundary (isDone) using per-partition ScheduledFuture tracking. The concurrency model is sound and the three test scenarios cover the key behavioral paths.

A few observations on the concurrency design worth noting for future readers:

Why >= checkpointBatchSize instead of ==: The processedEvents.addAndGet(processed) call at line 88 of EventHubsCheckpointUpdaterTask runs after lock.unlock() in the finally block, so it races with the processedEvents.set(0) reset in processCommit. The >= comparison ensures the batch flush fires even if addAndGet increments past the batch size before the counter resets — intentional and correct.

Single lock across partitions: The ReentrantLock in EventHubsConsumer is shared across all partition callbacks. processCommit holds it for the duration of the checkpoint round-trip, so high-frequency commits on one partition will briefly serialize others. This is acceptable at current scale but worth documenting if the partition count grows significantly.

doStop() lifecycle: The three clear() calls in doStop() remove the ScheduledFuture references without explicitly calling cancel() on them. This is fine as long as the executor itself is shut down in doShutdown() (via ShutdownAware), which cancels all pending tasks — but it assumes the shutdown sequence is doStop() → then doShutdown().

This review was generated by an AI agent, Hermès on behalf of @gnodet.

[davsclaus]

Thanks Guillaume. This backports the CAMEL-25781 fix to camel-4.18.x. A new checkpoint window now starts when the timer of the current one has run (EventHubsCheckpointUpdaterTask.isDone()), not when the wall clock has passed its scheduled time. The timer also takes the count under the consumer's lock. After this PR, EventHubsConsumer and EventHubsCheckpointUpdaterTask are identical to main, which is good. But it is not a straight cherry-pick of #28135. camel-4.18.x had neither CAMEL-24198 nor CAMEL-25683, and the conflict resolution brought in both.

1. Not blocking, but please update the description. Besides CAMEL-25781, this PR also backports:

   • CAMEL-24198 (03cb654): per-partition counters and timers, and the consumer lock. This is an Improvement, released in 4.22.0.
   • CAMEL-25683 (#27922): the timer takes its count when it starts and adds it back when it fails, the batch size check uses >=, and the cancelled task is removed.

   Taking them along is OK. On 4.18.x the shared processedEvents and lastTask mix the partitions, and processCommit runs without a lock from the partition threads. Without these two fixes, the CAMEL-25781 fix has no sound base to build on. The PR title and body should say this, and CAMEL-24198 and CAMEL-25683 should get 4.18.6 in their fix versions.

2. Not blocking. The tests for those two fixes were not brought over: EventHubsConsumerPerPartitionCheckpointTest, which needs the assertj-core test dependency from 03cb654, and EventHubsConsumerCheckpointAfterBatchSizeTest. Only EventHubsConsumerCheckpointWindowTest was added. The per-partition code is new on this branch, so these tests are worth adding. Leaving out the DST test is correct, because camel-test-dst is not on 4.18.x.

3. Not blocking. The upgrade note sits under "Upgrading from 4.18.3 to 4.18.4", but this ships in 4.18.6. It needs a new "Upgrading from 4.18.5 to 4.18.6" section. The note should also tell 4.18 users that checkpointBatchSize and checkpointBatchTimeout now count per partition (CAMEL-24198). Per the backport policy, the same entry also has to go into camel-4x-upgrade-guide-4_18.adoc on main, in its "Upgrading from 4.18.5 to 4.18.6" section, which does not have it yet.

The three gnodet-bot notes can be resolved. None of them is specific to this backport, because the code is the same on main:

• addAndGet(processed) runs outside the lock. That is correct, and the >= check covers it. If processCommit resets the count right after, the batch size checkpoint stores a later offset, so no events are left uncheckpointed.
• One ReentrantLock for all partitions. processCommit holds the lock only while it counts and starts updateCheckpointAsync(), not for the round trip of the checkpoint, so the cost is small.
• doStop() clears the maps without cancelling the futures. This was already the case before: the old lastScheduledTask was not cancelled on stop either. If anything, it is a follow-up for main, not for this backport.

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.

[oscerd]

TL;DR

On its merits this port looks sound: EventHubsConsumer and EventHubsCheckpointUpdaterTask end byte-identical to main, CI is green, and EventHubsConsumerCheckpointWindowTest covers CAMEL-25781 here as it does on main. The PR carries two more tickets than the earlier review lists.

Blockers

• None

Other findings: 1 (non-blocking)

A small addition to point 1 from @davsclaus. Besides CAMEL-24198 and CAMEL-25683, this PR also brings the Event Hubs part of:

• CAMEL-24159 (cd82682, #24867): the volatile event context in the task, processCommit made mutually exclusive (it added synchronized), and a failed checkpoint, from the timer or by batch size, logged at WARN with the stack trace instead of DEBUG with the message only.
• CAMEL-20199 (a52096e, #25247): the ReentrantLock that replaced that synchronized. This is the lock CAMEL-25781 now hands to the task; CAMEL-24198 relied on the existing synchronized, so the lock does not come from it.

Both are 4.22.0-only umbrella tickets across many components, so naming them in the PR description, with the part that comes along, may be clearer than adding 4.18.6 to their fix versions. The WARN logging is the change 4.18 users will notice in their logs, so one line about it in the description would help.

The earlier points (description, the two missing tests, the 4.18.5 → 4.18.6 section and the entry on main's 4_18 guide) are still open at this head.

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.

[gnodet-bot]

Following up on my earlier review — the concurrency logic was approved; this commit adds test coverage and fixes the upgrade guide section placement.

Upgrade guide section placement — ✅ corrected: 4.18.5 → 4.18.6 now appears above 4.18.4 → 4.18.5.

Test coverage — three new test classes cover per-partition isolation, post-batch-size timeout checkpointing, and failed-checkpoint recovery. The assertions are sound. The Sinks.Empty approach for controlling checkpoint completion timing is correct: Reactor's asMono() subscriber fires synchronously on tryEmitError/tryEmitEmpty (no scheduler hop), so the processCommit.invoke calls immediately after are guaranteed to see the updated counter state.

Two issues in the additions:

---

1. Upgrade guide — inaccurate statement about batch-size checkpoint logging (medium)

Line 35–36 states: "Checkpoint failures triggered by batch size are still logged at WARN." This is wrong. In the base branch (camel-4.18.x), the subscribe error handler in processCommit logged at DEBUG — identical to the timer path. The PR changes both to WARN. The "still at WARN" wording implies they were already at WARN before this change, which they were not. See the diff: - error -> LOG.debug("Error when updating Checkpoint: ...", error.getMessage()) → + error -> LOG.warn(...).

2. awaitility redundantly declared with explicit version in pom.xml (nit)

org.awaitility:awaitility is already a direct <dependency> in components/pom.xml (with ${awaitility-version}), inherited by all component submodules including camel-azure-eventhubs. No other camel-azure submodule that uses awaitility in tests re-declares it. The block in the module pom is redundant and the explicit <version> is inconsistent with how assertj-core is handled (version managed by parent BOM). The awaitility block can be removed; assertj-core (absent from components/pom.xml) must stay.

This review was generated by an AI agent, Hermès on behalf of @gnodet.

[gnodet-bot]

Following up on my earlier review — both issues have been addressed.

Upgrade guide logging wording — ✅ addressed: the sentence now correctly reads "from the timer and from the batch-size trigger are now logged at WARN level with the full stack trace instead of at DEBUG level with the message only."

Redundant awaitility dependency — ✅ removed. The assertj-core entry (not in components/pom.xml) correctly remains.

No new findings from the third commit. LGTM.

This review was generated by an AI agent, Hermès on behalf of @gnodet.

[davsclaus]

Thanks Guillaume, my earlier points are addressed:

• The description now lists everything the port brings: CAMEL-24198, CAMEL-25683, CAMEL-24159 and CAMEL-20199, including the WARN logging change.
• EventHubsConsumerPerPartitionCheckpointTest and EventHubsConsumerCheckpointAfterBatchSizeTest are added and identical to main. Only assertj-core is added to the pom, which is enough.
• The note sits in a new "Upgrading from 4.18.5 to 4.18.6" section and covers the per-partition counting.

One follow-up on main: the copy of this entry in camel-4x-upgrade-guide-4_18.adoc on main (added in 34af3c4) still has the sentence that 9c1f25a corrected here ("Checkpoint failures triggered by batch size are still logged at WARN"). On 4.18.x both paths logged at DEBUG before, so the two copies should be synced with the wording from this PR.

Non-blocking: CAMEL-24198, CAMEL-25683 and CAMEL-25781 should get 4.18.6 in their JIRA fix versions.

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.