This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 727cab81a8b8 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)
727cab81a8b8 is described below
commit 727cab81a8b88033c811ac57b88d7e4a407ef3dd
Author: allthingssecurity <[email protected]>
AuthorDate: Wed Oct 21 10:14:48 2026 +0530
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)
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 could be counted in a window whose timer had already run, and 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 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. The
CamelAzureEventHubsCheckpointUpdatedBy header can now be timeout (noted in the
4.24 upgrade guide).
EventHubsConsumerCheckpointWindowTest covers the three cases, and
EventHubsCheckpointDstTest pins DST seed 3042, which found the bug.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../eventhubs/EventHubsCheckpointUpdaterTask.java | 61 +++++++-
.../azure/eventhubs/EventHubsConsumer.java | 14 +-
.../EventHubsConsumerCheckpointWindowTest.java | 167 +++++++++++++++++++++
.../eventhubs/dst/EventHubsCheckpointDstTest.java | 20 ++-
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 13 ++
5 files changed, 262 insertions(+), 13 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 90ee2924f8e3..1de96d0c19ab 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,53 @@ 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 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());
+ }
+
+ 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 +105,26 @@ public class EventHubsCheckpointUpdaterTask implements
Runnable {
return scheduledTime;
}
+ /**
+ * 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;
+ }
+
+ /**
+ * 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 a1f5fce2385a..eddf97f3a7b2 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,9 +220,10 @@ public class EventHubsConsumer extends DefaultConsumer
implements ShutdownAware
/**
* 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
*/
- 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();
@@ -231,8 +232,11 @@ public class EventHubsConsumer extends DefaultConsumer
implements ShutdownAware
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 +255,8 @@ public class EventHubsConsumer extends DefaultConsumer
implements ShutdownAware
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 000000000000..3fc1b5525fb3
--- /dev/null
+++
b/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumerCheckpointWindowTest.java
@@ -0,0 +1,167 @@
+/*
+ * 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.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.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.never;
+import static org.mockito.Mockito.times;
+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;
+
+ @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);
+ }
+
+ @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);
+
+ // 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();
+ }
+
+ private void complete(EventContext eventContext) throws Exception {
+ consumer.processCommit(DefaultExchange.newFromEndpoint(endpoint),
eventContext);
+ }
+
+ private List<Runnable> scheduledTasks() {
+ ArgumentCaptor<Runnable> tasks =
ArgumentCaptor.forClass(Runnable.class);
+ verify(scheduledExecutorService, atLeastOnce())
+ .schedule(tasks.capture(), anyLong(), any(TimeUnit.class));
+ return tasks.getAllValues();
+ }
+
+ private void runTasksFrom(int index) {
+ List<Runnable> 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;
+ }
+}
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 f27db39e752c..3073d242c7fa 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 java.util.Map;
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 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
* 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=<seed>}; {@code
-Ddst.seeds=5000} runs more seeds.
* <p>
- * 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 @@ class EventHubsCheckpointDstTest {
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 d3b28ccc0f28..3fda251c39ed 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
@@ -1768,6 +1768,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
