From 6b4df9acee764fab7af27b3414560b64835fae12 Mon Sep 17 00:00:00 2001 From: smjain <49463903+allthingssecurity@users.noreply.github.com> Date: Mon, 19 Oct 2026 13:08:04 +0530 Subject: [PATCH 1/3] CAMEL-25781: camel-azure-eventhubs - an event that completes when the 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 --- .../EventHubsCheckpointUpdaterTask.java | 51 +++++- .../azure/eventhubs/EventHubsConsumer.java | 9 +- ...EventHubsConsumerCheckpointWindowTest.java | 166 ++++++++++++++++++ 3 files changed, 221 insertions(+), 5 deletions(-) create mode 100644 components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumerCheckpointWindowTest.java diff --git a/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsCheckpointUpdaterTask.java b/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsCheckpointUpdaterTask.java index 90ee2924f8e3f..0b5c62797ced3 100644 --- a/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsCheckpointUpdaterTask.java +++ b/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsCheckpointUpdaterTask.java @@ -17,6 +17,8 @@ package org.apache.camel.component.azure.eventhubs; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; import com.azure.messaging.eventhubs.models.EventContext; import org.slf4j.Logger; @@ -32,20 +34,44 @@ public class EventHubsCheckpointUpdaterTask implements Runnable { private volatile EventContext eventContext; private final AtomicInteger processedEvents; private volatile long scheduledTime; + // the lock of the consumer, which counts the events under it + private final Lock lock; + // set under the lock when the task has taken the count of its window, or was cancelled + private volatile boolean done; public EventHubsCheckpointUpdaterTask(EventContext eventContext, AtomicInteger processedEvents) { + this(eventContext, processedEvents, new ReentrantLock()); + } + + public EventHubsCheckpointUpdaterTask(EventContext eventContext, AtomicInteger processedEvents, Lock lock) { this.eventContext = eventContext; this.processedEvents = processedEvents; + this.lock = lock; } @Override public void run() { // take the count when the checkpoint is started: an event that completes while it is being stored is counted - // for the next window from 0, and a checkpoint by batch size in the meantime cannot be confused with this one - final int processed = processedEvents.getAndSet(0); + // for the next window from 0, and a checkpoint by batch size in the meantime cannot be confused with this one. + // This is done under the lock of the consumer, so an event is either counted (and its context set) before, and + // checkpointed here, or after, and then it sees that this task is done and starts a new window. + final int processed; + final EventContext context; + lock.lock(); + try { + if (done) { + LOG.debug("skip checkpointing offset: the window was already checkpointed by batch size"); + return; + } + done = true; + processed = processedEvents.getAndSet(0); + context = eventContext; + } finally { + lock.unlock(); + } if (processed > 0) { LOG.debug("checkpointing offset after reaching timeout, with a batch of {}", processed); - eventContext.updateCheckpointAsync() + context.updateCheckpointAsync() .subscribe(unused -> LOG.debug("Processed one event..."), error -> { LOG.warn("Error when updating Checkpoint: {}", error.getMessage(), error); @@ -70,6 +96,25 @@ public long getScheduledTime() { return scheduledTime; } + /** + * Whether this task has taken the count of its window (or was cancelled), so a new event must start a new window. + */ + public boolean isDone() { + return done; + } + + /** + * Cancels the task under the lock of the consumer: if it is already running, it takes nothing. + */ + public void cancel() { + lock.lock(); + try { + done = true; + } finally { + lock.unlock(); + } + } + public boolean isExpired() { return System.currentTimeMillis() > scheduledTime; } diff --git a/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumer.java b/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumer.java index a1f5fce2385a1..04bc0ebdd8896 100644 --- a/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumer.java +++ b/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumer.java @@ -231,8 +231,11 @@ private void processCommit(final Exchange exchange, final EventContext eventCont EventHubsCheckpointUpdaterTask checkpointTask = checkpointTasksByPartition.get(partitionId); ScheduledFuture scheduledTask = scheduledTasksByPartition.get(partitionId); - if (checkpointTask == null || checkpointTask.isExpired()) { - checkpointTask = new EventHubsCheckpointUpdaterTask(eventContext, processedEvents); + // a new window starts when the timer of the current one has run, not when the wall clock passed its time: the + // timer can run in the millisecond of its scheduled time, or later, and the event must be counted in the + // window whose timer has not taken the count yet + if (checkpointTask == null || checkpointTask.isDone()) { + checkpointTask = new EventHubsCheckpointUpdaterTask(eventContext, processedEvents, lock); // delegate the checkpoint update to a dedicated Thread long timeout = getConfiguration().getCheckpointBatchTimeout(); checkpointTask.setScheduledTime(System.currentTimeMillis() + timeout); @@ -251,6 +254,8 @@ private void processCommit(final Exchange exchange, final EventContext eventCont processedEvents.set(0); exchange.getIn().setHeader(EventHubsConstants.CHECKPOINT_UPDATED_BY, COMPLETED_BY_SIZE); LOG.debug("eventhub consumer batch size of reached for partition {}", partitionId); + // a timer that has already started takes nothing (it waits for the lock) + checkpointTask.cancel(); if (scheduledTask != null) { scheduledTask.cancel(false); } diff --git a/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumerCheckpointWindowTest.java b/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumerCheckpointWindowTest.java new file mode 100644 index 0000000000000..052b1fc4c3c25 --- /dev/null +++ b/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumerCheckpointWindowTest.java @@ -0,0 +1,166 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.azure.eventhubs; + +import java.lang.reflect.Method; +import java.util.List; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; + +import com.azure.messaging.eventhubs.models.EventContext; +import com.azure.messaging.eventhubs.models.PartitionContext; +import org.apache.camel.AsyncProcessor; +import org.apache.camel.CamelContext; +import org.apache.camel.Endpoint; +import org.apache.camel.Exchange; +import org.apache.camel.ExtendedCamelContext; +import org.apache.camel.spi.ExchangeFactory; +import org.apache.camel.support.DefaultExchange; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import reactor.core.publisher.Mono; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.Mockito.atLeastOnce; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * The timer of a checkpoint window and the events of the partition: every completed event must be checkpointed by a + * timer that runs after it was counted, whenever the timer runs relative to the wall clock and to the batch size. + */ +class EventHubsConsumerCheckpointWindowTest { + + private final EventHubsEndpoint endpoint = mock(); + private final EventHubsConfiguration configuration = new EventHubsConfiguration(); + private final AsyncProcessor processor = mock(); + private final CamelContext context = mock(); + private final ExtendedCamelContext ecc = mock(); + private final ExchangeFactory exchangeFactory = mock(); + private final ScheduledExecutorService scheduledExecutorService = mock(); + @SuppressWarnings("rawtypes") + private final ScheduledFuture scheduledFuture = mock(ScheduledFuture.class); + + private EventHubsConsumer consumer; + private Method processCommit; + + @BeforeEach + void setUp() throws Exception { + configuration.setCheckpointBatchSize(500); + configuration.setCheckpointBatchTimeout(60_000); + + when(endpoint.getCamelContext()).thenReturn(context); + when(context.getCamelContextExtension()).thenReturn(ecc); + when(ecc.getExchangeFactory()).thenReturn(exchangeFactory); + when(exchangeFactory.newExchangeFactory(any())).thenReturn(exchangeFactory); + when(exchangeFactory.create(any(Endpoint.class), anyBoolean())) + .thenAnswer(invocation -> DefaultExchange.newFromEndpoint(invocation.getArgument(0))); + when(endpoint.getConfiguration()).thenReturn(configuration); + consumer = new EventHubsConsumer(endpoint, processor); + + var field = EventHubsConsumer.class.getDeclaredField("scheduledExecutorService"); + field.setAccessible(true); + field.set(consumer, scheduledExecutorService); + when(scheduledExecutorService.schedule(any(Runnable.class), anyLong(), any(TimeUnit.class))) + .thenReturn(scheduledFuture); + + processCommit = EventHubsConsumer.class.getDeclaredMethod("processCommit", Exchange.class, EventContext.class); + processCommit.setAccessible(true); + } + + @Test + void eventCompletedAfterTheTimerRanBeforeItsScheduledTimeIsCheckpointed() throws Exception { + EventContext first = eventContext(); + EventContext second = eventContext(); + + complete(first); + // the timer runs while the wall clock has not passed its scheduled time yet (the same millisecond) + scheduledTasks().get(0).run(); + verify(first).updateCheckpointAsync(); + + complete(second); + runTasksFrom(1); + + verify(second).updateCheckpointAsync(); + } + + @Test + void eventCompletedAfterTheScheduledTimeBeforeTheTimerRanIsCheckpointed() throws Exception { + EventContext first = eventContext(); + EventContext second = eventContext(); + + complete(first); + // the wall clock has passed the scheduled time, the timer has not run yet (late by a GC pause, a busy thread) + EventHubsCheckpointUpdaterTask firstTimer = (EventHubsCheckpointUpdaterTask) scheduledTasks().get(0); + firstTimer.setScheduledTime(System.currentTimeMillis() - 1); + + complete(second); + runTasksFrom(0); + + verify(second).updateCheckpointAsync(); + } + + @Test + void timerStartedBeforeABatchSizeCheckpointDoesNotTakeTheNextWindow() throws Exception { + configuration.setCheckpointBatchSize(2); + EventContext first = eventContext(); + EventContext second = eventContext(); + EventContext third = eventContext(); + + complete(first); + complete(second); // batch size reached: checkpoint, the timer of the window is cancelled + verify(second).updateCheckpointAsync(); + complete(third); // starts a new window + + // the timer of the first window had already started, so the cancel did not stop it + runTasksFrom(0); + + verify(third).updateCheckpointAsync(); + } + + private void complete(EventContext eventContext) throws Exception { + processCommit.invoke(consumer, DefaultExchange.newFromEndpoint(endpoint), eventContext); + } + + private List scheduledTasks() { + ArgumentCaptor tasks = ArgumentCaptor.forClass(Runnable.class); + verify(scheduledExecutorService, atLeastOnce()) + .schedule(tasks.capture(), anyLong(), any(TimeUnit.class)); + return tasks.getAllValues(); + } + + private void runTasksFrom(int index) { + List tasks = scheduledTasks(); + for (int i = index; i < tasks.size(); i++) { + tasks.get(i).run(); + } + } + + private static EventContext eventContext() { + EventContext eventContext = mock(EventContext.class); + PartitionContext partitionContext = mock(PartitionContext.class); + when(partitionContext.getPartitionId()).thenReturn("0"); + when(eventContext.getPartitionContext()).thenReturn(partitionContext); + when(eventContext.updateCheckpointAsync()).thenReturn(Mono.empty()); + return eventContext; + } +} From 459bacf9ecd92f3ce5230b00f77c81039d8f419d Mon Sep 17 00:00:00 2001 From: smjain <49463903+allthingssecurity@users.noreply.github.com> Date: Tue, 20 Oct 2026 12:55:07 +0530 Subject: [PATCH 2/3] CAMEL-25781: camel-azure-eventhubs - pin DST seed 3042, upgrade guide 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 --- .../EventHubsCheckpointUpdaterTask.java | 5 +++-- .../azure/eventhubs/EventHubsConsumer.java | 2 +- ...EventHubsConsumerCheckpointWindowTest.java | 8 +------- .../dst/EventHubsCheckpointDstTest.java | 20 +++++++++++++------ .../pages/camel-4x-upgrade-guide-4_24.adoc | 13 ++++++++++++ 5 files changed, 32 insertions(+), 16 deletions(-) diff --git a/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsCheckpointUpdaterTask.java b/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsCheckpointUpdaterTask.java index 0b5c62797ced3..94ae04d9238d5 100644 --- a/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsCheckpointUpdaterTask.java +++ b/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsCheckpointUpdaterTask.java @@ -36,8 +36,8 @@ public class EventHubsCheckpointUpdaterTask implements Runnable { private volatile long scheduledTime; // the lock of the consumer, which counts the events under it private final Lock lock; - // set under the lock when the task has taken the count of its window, or was cancelled - private volatile boolean done; + // set when the task has taken the count of its window, or was cancelled; read and written under the lock only + private boolean done; public EventHubsCheckpointUpdaterTask(EventContext eventContext, AtomicInteger processedEvents) { this(eventContext, processedEvents, new ReentrantLock()); @@ -98,6 +98,7 @@ public long getScheduledTime() { /** * Whether this task has taken the count of its window (or was cancelled), so a new event must start a new window. + * Called under the lock of the consumer. */ public boolean isDone() { return done; diff --git a/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumer.java b/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumer.java index 04bc0ebdd8896..d643d879dd8cc 100644 --- a/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumer.java +++ b/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumer.java @@ -222,7 +222,7 @@ private void onErrorListener(final ErrorContext errorContext) { * * @param exchange the exchange */ - private void processCommit(final Exchange exchange, final EventContext eventContext) { + void processCommit(final Exchange exchange, final EventContext eventContext) { lock.lock(); try { final String partitionId = eventContext.getPartitionContext().getPartitionId(); diff --git a/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumerCheckpointWindowTest.java b/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumerCheckpointWindowTest.java index 052b1fc4c3c25..04cf6a2438b80 100644 --- a/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumerCheckpointWindowTest.java +++ b/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumerCheckpointWindowTest.java @@ -16,7 +16,6 @@ */ package org.apache.camel.component.azure.eventhubs; -import java.lang.reflect.Method; import java.util.List; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; @@ -27,7 +26,6 @@ import org.apache.camel.AsyncProcessor; import org.apache.camel.CamelContext; import org.apache.camel.Endpoint; -import org.apache.camel.Exchange; import org.apache.camel.ExtendedCamelContext; import org.apache.camel.spi.ExchangeFactory; import org.apache.camel.support.DefaultExchange; @@ -61,7 +59,6 @@ class EventHubsConsumerCheckpointWindowTest { private final ScheduledFuture scheduledFuture = mock(ScheduledFuture.class); private EventHubsConsumer consumer; - private Method processCommit; @BeforeEach void setUp() throws Exception { @@ -82,9 +79,6 @@ void setUp() throws Exception { field.set(consumer, scheduledExecutorService); when(scheduledExecutorService.schedule(any(Runnable.class), anyLong(), any(TimeUnit.class))) .thenReturn(scheduledFuture); - - processCommit = EventHubsConsumer.class.getDeclaredMethod("processCommit", Exchange.class, EventContext.class); - processCommit.setAccessible(true); } @Test @@ -138,7 +132,7 @@ void timerStartedBeforeABatchSizeCheckpointDoesNotTakeTheNextWindow() throws Exc } private void complete(EventContext eventContext) throws Exception { - processCommit.invoke(consumer, DefaultExchange.newFromEndpoint(endpoint), eventContext); + consumer.processCommit(DefaultExchange.newFromEndpoint(endpoint), eventContext); } private List scheduledTasks() { diff --git a/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/dst/EventHubsCheckpointDstTest.java b/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/dst/EventHubsCheckpointDstTest.java index f27db39e752c4..3073d242c7faa 100644 --- a/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/dst/EventHubsCheckpointDstTest.java +++ b/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/dst/EventHubsCheckpointDstTest.java @@ -20,6 +20,7 @@ import org.apache.camel.test.dst.Dst; import org.apache.camel.test.dst.DstInstrument; +import org.apache.camel.test.dst.DstSeed; import org.apache.camel.test.dst.DstTest; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.Test; @@ -32,12 +33,9 @@ * partition goes quiet. Regression test for CAMEL-25683 (apache/camel#27922): before that fix 2064 of 5000 seeds fail. * A failing seed is replayed with {@code -Ddst.seed=}; {@code -Ddst.seeds=5000} runs more seeds. *

- * Known residual on main, outside the default 1000 seeds: seed 3042 of 5000 fails when an event completes in the same - * millisecond in which the timer of its window ran ({@code EventHubsCheckpointUpdaterTask.isExpired()} compares with - * {@code >}, so the window is reused and that event waits for the next one; if the partition stays quiet it is - * delivered again after a restart). In real time the window is narrow but not only that millisecond: the timer task - * does not take the consumer's lock, so it can also run between the {@code isExpired()} check and the count of the - * event. Virtual time makes the same-millisecond case far more likely than it is in real time. + * Seed 3042 (of 5000) failed before CAMEL-25781 and is pinned below: an event that completed in the millisecond in + * which the timer of its window ran joined that window after the timer had taken its count, so it was not checkpointed + * until the next event (or delivered again after a restart if the partition stayed quiet). */ @DstInstrument("org.apache.camel.component.azure.eventhubs") class EventHubsCheckpointDstTest { @@ -47,6 +45,16 @@ void everyCompletedEventIsCheckpointedWhenThePartitionGoesQuiet(Dst dst) throws new EventHubsWorld(dst).run(); } + /** + * CAMEL-25781: an event completes in the millisecond in which the timer of its window ran, and must be checkpointed + * by the next timer. + */ + @DstTest + @DstSeed(3042) + void eventThatCompletesWhenTheTimerOfItsWindowRanIsCheckpointed(Dst dst) throws Exception { + new EventHubsWorld(dst).run(); + } + /** The same seed twice in this JVM: identical step traces (the cross-JVM check compares the trace files). */ @Test void replayOfASeedGivesTheSameTrace() throws Exception { diff --git a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc index a1673f89bf16b..6c3c273f41cc6 100644 --- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc +++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc @@ -1681,6 +1681,19 @@ exception handler of the consumer, with the blob name in the message (by default poll that lists the blob). Before, the poll failed, and as every poll lists the container again, one such blob stopped the consumer from consuming any blob. The poll still fails when none of the listed blobs could be got. +=== camel-azure-eventhubs - an event that completes when the checkpoint timer of its window has run is checkpointed + +With `checkpointBatchTimeout`, the consumer now starts a new checkpoint window when the timer of the current window +has run, and no longer when the wall clock passes the time the timer was scheduled for. Before, an event that +completed in the millisecond in which the timer ran (or while the timer was taking the count) was counted in the +window whose count had already been taken: it was checkpointed only with the next event, and when the partition stayed +quiet it was delivered again after a restart. + +The `CamelAzureEventHubsCheckpointUpdatedBy` header can now be `timeout`. An event that completes after the +scheduled time of its window, but before the (late) timer has run, joins that window and gets `timeout`; its offset is +checkpointed by that timer. Before, such an event always started a new window, so `timeout` was practically never +set. Routes that read this header can now see `timeout` as well as `size`. + === camel-azure-servicebus - the dead-letter reason and description are bounded When the consumer dead-letters a message after a failed route with `enableDeadLettering=true`, the reason and the From df2c63c36368a49d92b9efd613685bcd958b9f3c Mon Sep 17 00:00:00 2001 From: smjain <49463903+allthingssecurity@users.noreply.github.com> Date: Tue, 20 Oct 2026 13:31:18 +0530 Subject: [PATCH 3/3] CAMEL-25781: camel-azure-eventhubs - Javadoc for processCommit and the 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 --- .../azure/eventhubs/EventHubsCheckpointUpdaterTask.java | 9 +++++++++ .../component/azure/eventhubs/EventHubsConsumer.java | 3 ++- .../eventhubs/EventHubsConsumerCheckpointWindowTest.java | 7 +++++++ 3 files changed, 18 insertions(+), 1 deletion(-) diff --git a/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsCheckpointUpdaterTask.java b/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsCheckpointUpdaterTask.java index 94ae04d9238d5..1de96d0c19ab7 100644 --- a/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsCheckpointUpdaterTask.java +++ b/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsCheckpointUpdaterTask.java @@ -39,6 +39,15 @@ public class EventHubsCheckpointUpdaterTask implements Runnable { // set when the task has taken the count of its window, or was cancelled; read and written under the lock only private boolean done; + /** + * Creates a task that synchronizes on a private lock of its own, unrelated to any consumer: for standalone or test + * use. {@link EventHubsConsumer} uses {@link #EventHubsCheckpointUpdaterTask(EventContext, AtomicInteger, Lock)} + * with its own lock, which it holds while it counts the events; with a private lock an event could be counted in a + * window whose timer has already taken the count. + * + * @param eventContext the event context to checkpoint + * @param processedEvents the count of the events processed since the last checkpoint + */ public EventHubsCheckpointUpdaterTask(EventContext eventContext, AtomicInteger processedEvents) { this(eventContext, processedEvents, new ReentrantLock()); } diff --git a/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumer.java b/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumer.java index d643d879dd8cc..eddf97f3a7b22 100644 --- a/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumer.java +++ b/components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumer.java @@ -220,7 +220,8 @@ private void onErrorListener(final ErrorContext errorContext) { /** * Strategy to commit the offset after message being processed successfully. * - * @param exchange the exchange + * @param exchange the exchange + * @param eventContext the event context of the event, used to update the checkpoint */ void processCommit(final Exchange exchange, final EventContext eventContext) { lock.lock(); diff --git a/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumerCheckpointWindowTest.java b/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumerCheckpointWindowTest.java index 04cf6a2438b80..3fc1b5525fb3b 100644 --- a/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumerCheckpointWindowTest.java +++ b/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumerCheckpointWindowTest.java @@ -39,6 +39,8 @@ import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.Mockito.atLeastOnce; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -128,6 +130,11 @@ void timerStartedBeforeABatchSizeCheckpointDoesNotTakeTheNextWindow() throws Exc // the timer of the first window had already started, so the cancel did not stop it runTasksFrom(0); + // the cancelled timer of the first window takes nothing: its event context is second (set when second was + // counted), which was checkpointed once by batch size and not again; first was never checkpointed by itself + verify(first, never()).updateCheckpointAsync(); + verify(second, times(1)).updateCheckpointAsync(); + verify(third).updateCheckpointAsync(); }