This is an automated email from the ASF dual-hosted git repository.
gnodet pushed a commit to branch camel-4.18.x
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/camel-4.18.x by this push:
new 5c2c941ab4de [backport camel-4.18.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 (#28206)
5c2c941ab4de is described below
commit 5c2c941ab4de4974bfb4d6bc2c2900902351ba13
Author: Guillaume Nodet <[email protected]>
AuthorDate: Wed Oct 21 19:07:15 2026 +0200
[backport camel-4.18.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 (#28206)
* [backport camel-4.18.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
* Address review: add missing tests, fix upgrade guide section placement
* Address review: fix upgrade guide logging wording; remove redundant
awaitility dep
---
.../camel-azure/camel-azure-eventhubs/pom.xml | 5 +
.../eventhubs/EventHubsCheckpointUpdaterTask.java | 77 ++++++-
.../azure/eventhubs/EventHubsConsumer.java | 113 ++++++----
...ntHubsConsumerCheckpointAfterBatchSizeTest.java | 229 +++++++++++++++++++++
.../EventHubsConsumerCheckpointWindowTest.java | 167 +++++++++++++++
...ventHubsConsumerPerPartitionCheckpointTest.java | 199 ++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_18.adoc | 23 +++
7 files changed, 763 insertions(+), 50 deletions(-)
diff --git a/components/camel-azure/camel-azure-eventhubs/pom.xml
b/components/camel-azure/camel-azure-eventhubs/pom.xml
index b0b9b5121773..81b496d5b879 100644
--- a/components/camel-azure/camel-azure-eventhubs/pom.xml
+++ b/components/camel-azure/camel-azure-eventhubs/pom.xml
@@ -72,5 +72,10 @@
<artifactId>commons-lang3</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.assertj</groupId>
+ <artifactId>assertj-core</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</project>
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 49723ac7c4f7..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;
@@ -29,26 +31,63 @@ public class EventHubsCheckpointUpdaterTask implements
Runnable {
private static final Logger LOG =
LoggerFactory.getLogger(EventHubsCheckpointUpdaterTask.class);
- private EventContext eventContext;
+ 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() {
- if (processedEvents.get() > 0) {
- LOG.debug("checkpointing offset after reaching timeout, with a
batch of {}", processedEvents.get());
- eventContext.updateCheckpointAsync()
+ // 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.
+ // 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);
+ context.updateCheckpointAsync()
.subscribe(unused -> LOG.debug("Processed one event..."),
- error -> LOG.debug("Error when updating
Checkpoint: {}", error.getMessage()),
- () -> {
- LOG.debug("Checkpoint updated.");
- processedEvents.set(0);
- });
+ error -> {
+ LOG.warn("Error when updating Checkpoint: {}",
error.getMessage(), error);
+ // these events are not checkpointed: count
them again for the next checkpoint
+ processedEvents.addAndGet(processed);
+ },
+ () -> LOG.debug("Checkpoint updated."));
} else {
LOG.debug("skip checkpointing offset even if timeout is reached.
No events processed");
}
@@ -66,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 ce1e6e79da56..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
@@ -16,10 +16,13 @@
*/
package org.apache.camel.component.azure.eventhubs;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.ReentrantLock;
import com.azure.messaging.eventhubs.EventProcessorClient;
import com.azure.messaging.eventhubs.models.ErrorContext;
@@ -46,16 +49,15 @@ public class EventHubsConsumer extends DefaultConsumer
implements ShutdownAware
// we use the EventProcessorClient as recommended by Azure docs to consume
from all partitions
private EventProcessorClient processorClient;
- private final AtomicInteger processedEvents;
+ private final ReentrantLock lock = new ReentrantLock();
private final AtomicInteger pendingExchanges = new AtomicInteger();
+ private final Map<String, AtomicInteger> processedEventsByPartition = new
ConcurrentHashMap<>();
+ private final Map<String, ScheduledFuture<?>> scheduledTasksByPartition =
new ConcurrentHashMap<>();
+ private final Map<String, EventHubsCheckpointUpdaterTask>
checkpointTasksByPartition = new ConcurrentHashMap<>();
private ScheduledExecutorService scheduledExecutorService;
- private ScheduledFuture<?> lastScheduledTask;
- private EventHubsCheckpointUpdaterTask lastTask;
public EventHubsConsumer(final EventHubsEndpoint endpoint, final Processor
processor) {
super(endpoint, processor);
-
- this.processedEvents = new AtomicInteger();
}
@Override
@@ -82,6 +84,10 @@ public class EventHubsConsumer extends DefaultConsumer
implements ShutdownAware
processorClient.stop();
}
+ processedEventsByPartition.clear();
+ scheduledTasksByPartition.clear();
+ checkpointTasksByPartition.clear();
+
// shutdown camel consumer
super.doStop();
}
@@ -214,46 +220,71 @@ 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) {
- if (lastTask == null || lastTask.isExpired()) {
- lastTask = new EventHubsCheckpointUpdaterTask(eventContext,
processedEvents);
- // delegate the checkpoint update to a dedicated Thread
- long timeout = getConfiguration().getCheckpointBatchTimeout();
- lastTask.setScheduledTime(System.currentTimeMillis() + timeout);
- lastScheduledTask = scheduledExecutorService.schedule(lastTask,
timeout, TimeUnit.MILLISECONDS);
- } else {
- // updates the eventContext to use for the offset to be the most
accurate
- lastTask.setEventContext(eventContext);
- }
-
+ void processCommit(final Exchange exchange, final EventContext
eventContext) {
+ lock.lock();
try {
- var cnt = processedEvents.incrementAndGet();
- if (cnt == getConfiguration().getCheckpointBatchSize()) {
- processedEvents.set(0);
-
exchange.getIn().setHeader(EventHubsConstants.CHECKPOINT_UPDATED_BY,
COMPLETED_BY_SIZE);
- LOG.debug("eventhub consumer batch size of reached");
- if (lastScheduledTask != null) {
- lastScheduledTask.cancel(false);
- }
- eventContext.updateCheckpointAsync()
- .subscribe(unused -> LOG.debug("Processed one
event..."),
- error -> LOG.debug("Error when updating
Checkpoint: {}", error.getMessage()),
- () -> {
- LOG.debug("Checkpoint updated.");
- });
- } else if (lastTask != null && lastTask.isExpired()) {
-
exchange.getIn().setHeader(EventHubsConstants.CHECKPOINT_UPDATED_BY,
COMPLETED_BY_TIMEOUT);
- LOG.debug("eventhub consumer batch timeout reached");
+ final String partitionId =
eventContext.getPartitionContext().getPartitionId();
+ final AtomicInteger processedEvents =
processedEventsByPartition.computeIfAbsent(partitionId,
+ ignored -> new AtomicInteger());
+ EventHubsCheckpointUpdaterTask checkpointTask =
checkpointTasksByPartition.get(partitionId);
+ ScheduledFuture<?> scheduledTask =
scheduledTasksByPartition.get(partitionId);
+
+ // 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);
+ scheduledTask =
scheduledExecutorService.schedule(checkpointTask, timeout,
TimeUnit.MILLISECONDS);
+ checkpointTasksByPartition.put(partitionId, checkpointTask);
+ scheduledTasksByPartition.put(partitionId, scheduledTask);
} else {
- LOG.debug("neither eventhub consumer batch size of {}/{} nor
batch timeout reached yet", cnt,
- getConfiguration().getCheckpointBatchSize());
+ // updates the eventContext to use for the offset to be the
most accurate
+ checkpointTask.setEventContext(eventContext);
+ }
+
+ try {
+ var cnt = processedEvents.incrementAndGet();
+ // >= and not ==: a failed timeout checkpoint adds its count
back, so the count can pass the batch size
+ if (cnt >= getConfiguration().getCheckpointBatchSize()) {
+ 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);
+ }
+ // the next event starts a new timeout window: the
cancelled task must not be reused until its
+ // scheduled time, else the events of the rest of that
window are not checkpointed by the timeout
+ checkpointTasksByPartition.remove(partitionId);
+ scheduledTasksByPartition.remove(partitionId);
+ eventContext.updateCheckpointAsync()
+ .subscribe(unused -> LOG.debug("Processed one
event..."),
+ error -> LOG.warn("Error when updating
Checkpoint: {}", error.getMessage(), error),
+ () -> {
+ LOG.debug("Checkpoint updated.");
+ });
+ } else if (checkpointTask.isExpired()) {
+
exchange.getIn().setHeader(EventHubsConstants.CHECKPOINT_UPDATED_BY,
COMPLETED_BY_TIMEOUT);
+ LOG.debug("eventhub consumer batch timeout reached for
partition {}", partitionId);
+ } else {
+ LOG.debug("neither eventhub consumer batch size of {}/{}
nor batch timeout reached yet for partition {}",
+ cnt, getConfiguration().getCheckpointBatchSize(),
partitionId);
+ }
+ // we assume that the scheduled task has done the update by
its side
+ } catch (Exception ex) {
+ getExceptionHandler().handleException(
+ "Error occurred during updating the checkpoint. This
exception is ignored.",
+ exchange, ex);
}
- // we assume that the scheduled task has done the update by its
side
- } catch (Exception ex) {
- getExceptionHandler().handleException("Error occurred during
updating the checkpoint. This exception is ignored.",
- exchange, ex);
+ } finally {
+ lock.unlock();
}
}
diff --git
a/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumerCheckpointAfterBatchSizeTest.java
b/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumerCheckpointAfterBatchSizeTest.java
new file mode 100644
index 000000000000..19a22d70484b
--- /dev/null
+++
b/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumerCheckpointAfterBatchSizeTest.java
@@ -0,0 +1,229 @@
+/*
+ * 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.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+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.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import reactor.core.publisher.Mono;
+import reactor.core.publisher.Sinks;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * The events completed after a checkpoint by batch size must be checkpointed
by the batch timeout, also when no further
+ * event arrives (end of a burst). Uses a real scheduler with a short batch
timeout.
+ */
+class EventHubsConsumerCheckpointAfterBatchSizeTest {
+
+ 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 scheduler =
Executors.newSingleThreadScheduledExecutor();
+
+ private EventHubsConsumer consumer;
+ private Method processCommit;
+
+ @BeforeEach
+ void setUp() throws Exception {
+ configuration.setCheckpointBatchSize(2);
+ configuration.setCheckpointBatchTimeout(1000);
+
+ 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, scheduler);
+
+ processCommit =
EventHubsConsumer.class.getDeclaredMethod("processCommit", Exchange.class,
EventContext.class);
+ processCommit.setAccessible(true);
+ }
+
+ @AfterEach
+ void tearDown() {
+ scheduler.shutdownNow();
+ }
+
+ @Test
+ void eventAfterBatchSizeCheckpointIsCheckpointedByTimeout() throws
Exception {
+ EventContext first = eventContext("0");
+ EventContext second = eventContext("0");
+ EventContext third = eventContext("0");
+
+ complete(first);
+ complete(second);
+ // the batch size is reached: checkpoint at the second event
+ verify(second, times(1)).updateCheckpointAsync();
+
+ // the last event of the burst: no other event follows
+ complete(third);
+
+ await().atMost(10, TimeUnit.SECONDS)
+ .untilAsserted(() -> verify(third,
times(1)).updateCheckpointAsync());
+ }
+
+ @Test
+ void eventsAfterSecondBatchSizeCheckpointAreCheckpointedByTimeout() throws
Exception {
+ EventContext[] events = new EventContext[5];
+ for (int i = 0; i < events.length; i++) {
+ events[i] = eventContext("0");
+ complete(events[i]);
+ }
+ // checkpoints by batch size at the second and the fourth event
+ verify(events[1], times(1)).updateCheckpointAsync();
+ verify(events[3], times(1)).updateCheckpointAsync();
+
+ await().atMost(10, TimeUnit.SECONDS)
+ .untilAsserted(() -> verify(events[4],
times(1)).updateCheckpointAsync());
+ }
+
+ @Test
+ void
eventCompletedWhileTheTimeoutCheckpointIsStoredIsCheckpointedByTheNextTimeout()
throws Exception {
+ configuration.setCheckpointBatchSize(10);
+ // the checkpoint of the first window is stored only when the test
says so
+ Sinks.Empty<Void> stored = Sinks.empty();
+ AtomicLong checkpointedAt = new AtomicLong();
+ EventContext first = eventContext("0");
+ doAnswer(invocation -> {
+ checkpointedAt.set(System.currentTimeMillis());
+ return stored.asMono();
+ }).when(first).updateCheckpointAsync();
+ EventContext second = eventContext("0");
+
+ complete(first);
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> verify(first,
times(1)).updateCheckpointAsync());
+ // the window of the first task is over, so the next event starts a
new task
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
System.currentTimeMillis() > checkpointedAt.get());
+
+ // an event completes while the checkpoint of the first window is
being stored
+ complete(second);
+ stored.tryEmitEmpty();
+
+ await().atMost(10, TimeUnit.SECONDS)
+ .untilAsserted(() -> verify(second,
times(1)).updateCheckpointAsync());
+ }
+
+ @Test
+ void eventAfterBatchSizeCheckpointDuringTimeoutCheckpointIsCheckpointed()
throws Exception {
+ configuration.setCheckpointBatchSize(3);
+ // the checkpoint of the first window is stored only when the test
says so
+ Sinks.Empty<Void> stored = Sinks.empty();
+ AtomicLong checkpointedAt = new AtomicLong();
+ EventContext first = eventContext("0");
+ doAnswer(invocation -> {
+ checkpointedAt.set(System.currentTimeMillis());
+ return stored.asMono();
+ }).when(first).updateCheckpointAsync();
+ EventContext second = eventContext("0");
+ EventContext third = eventContext("0");
+ EventContext fourth = eventContext("0");
+
+ complete(first);
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> verify(first,
times(1)).updateCheckpointAsync());
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
System.currentTimeMillis() > checkpointedAt.get());
+
+ // while the timeout checkpoint of the first event is stored, more
events complete, and the count climbs back
+ // to the value that checkpoint started with
+ complete(second);
+ complete(third);
+ complete(fourth);
+ stored.tryEmitEmpty();
+
+ // the last event of the burst must still be checkpointed, by batch
size or by the next timeout
+ await().atMost(10, TimeUnit.SECONDS)
+ .untilAsserted(() -> verify(fourth,
times(1)).updateCheckpointAsync());
+ }
+
+ @Test
+ void batchSizeCheckpointStillWorksAfterFailedTimeoutCheckpoint() throws
Exception {
+ configuration.setCheckpointBatchSize(3);
+ // the checkpoint of the first window fails only when the test says so
+ Sinks.Empty<Void> stored = Sinks.empty();
+ AtomicLong checkpointedAt = new AtomicLong();
+ EventContext first = eventContext("0");
+ EventContext second = eventContext("0");
+ doAnswer(invocation -> {
+ checkpointedAt.set(System.currentTimeMillis());
+ return stored.asMono();
+ }).when(second).updateCheckpointAsync();
+ EventContext third = eventContext("0");
+ EventContext fourth = eventContext("0");
+
+ complete(first);
+ complete(second);
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(() ->
verify(second, times(1)).updateCheckpointAsync());
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
System.currentTimeMillis() > checkpointedAt.get());
+
+ // one event completes while the timeout checkpoint of two events is
stored, then storing fails: the two
+ // events are counted again, so the count is 3, and the next event
takes it past the batch size
+ complete(third);
+ stored.tryEmitError(new IllegalStateException("store failed"));
+
+ Exchange exchange = DefaultExchange.newFromEndpoint(endpoint);
+ processCommit.invoke(consumer, exchange, fourth);
+
+ verify(fourth, times(1)).updateCheckpointAsync();
+ assertEquals(EventHubsConstants.COMPLETED_BY_SIZE,
+
exchange.getIn().getHeader(EventHubsConstants.CHECKPOINT_UPDATED_BY));
+ }
+
+ private void complete(EventContext eventContext) throws Exception {
+ processCommit.invoke(consumer,
DefaultExchange.newFromEndpoint(endpoint), eventContext);
+ }
+
+ private static EventContext eventContext(String partitionId) {
+ EventContext eventContext = mock(EventContext.class);
+ PartitionContext partitionContext = mock(PartitionContext.class);
+ when(partitionContext.getPartitionId()).thenReturn(partitionId);
+ 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/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/EventHubsConsumerPerPartitionCheckpointTest.java
b/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumerPerPartitionCheckpointTest.java
new file mode 100644
index 000000000000..570833d609bc
--- /dev/null
+++
b/components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumerPerPartitionCheckpointTest.java
@@ -0,0 +1,199 @@
+/*
+ * 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.Map;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import com.azure.messaging.eventhubs.EventProcessorClient;
+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.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class EventHubsConsumerPerPartitionCheckpointTest {
+
+ 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 EventProcessorClient processorClient = 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(3);
+ 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);
+
+ setField(consumer, "scheduledExecutorService",
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 processedEventsAreTrackedPerPartition() throws Exception {
+ invokeProcessCommit(exchange(), eventContext("0"));
+ invokeProcessCommit(exchange(), eventContext("0"));
+ invokeProcessCommit(exchange(), eventContext("1"));
+
+ assertThat(processedEventsForPartition("0")).isEqualTo(2);
+ assertThat(processedEventsForPartition("1")).isEqualTo(1);
+ }
+
+ @Test
+ void batchSizeCheckpointOnOnePartitionDoesNotResetOtherPartition() throws
Exception {
+ configuration.setCheckpointBatchSize(2);
+ EventContext partitionZeroContext = eventContext("0");
+
when(partitionZeroContext.updateCheckpointAsync()).thenReturn(Mono.empty());
+
+ invokeProcessCommit(exchange(), eventContext("1"));
+ invokeProcessCommit(exchange(), partitionZeroContext);
+ invokeProcessCommit(exchange(), partitionZeroContext);
+
+ assertThat(processedEventsForPartition("0")).isZero();
+ assertThat(processedEventsForPartition("1")).isEqualTo(1);
+ verify(partitionZeroContext, times(1)).updateCheckpointAsync();
+ }
+
+ @Test
+ void eachPartitionSchedulesOwnCheckpointTask() throws Exception {
+ ArgumentCaptor<Runnable> tasks =
ArgumentCaptor.forClass(Runnable.class);
+
+ invokeProcessCommit(exchange(), eventContext("0"));
+ invokeProcessCommit(exchange(), eventContext("1"));
+
+ verify(scheduledExecutorService, times(2)).schedule(tasks.capture(),
anyLong(), any(TimeUnit.class));
+
assertThat(tasks.getAllValues().get(0)).isNotSameAs(tasks.getAllValues().get(1));
+
+ Map<String, EventHubsCheckpointUpdaterTask> checkpointTasks =
getCheckpointTasksByPartition();
+ assertThat(checkpointTasks).hasSize(2);
+
assertThat(checkpointTasks.get("0")).isNotSameAs(checkpointTasks.get("1"));
+ }
+
+ @Test
+ void reusesCheckpointTaskWithinSamePartition() throws Exception {
+ invokeProcessCommit(exchange(), eventContext("3"));
+ invokeProcessCommit(exchange(), eventContext("3"));
+
+ verify(scheduledExecutorService,
times(1)).schedule(any(Runnable.class), anyLong(), any(TimeUnit.class));
+ assertThat(getCheckpointTasksByPartition()).hasSize(1);
+ assertThat(processedEventsForPartition("3")).isEqualTo(2);
+ }
+
+ @Test
+ void doStopClearsPerPartitionCheckpointState() throws Exception {
+ setField(consumer, "processorClient", processorClient);
+ invokeProcessCommit(exchange(), eventContext("0"));
+ invokeProcessCommit(exchange(), eventContext("1"));
+
+ consumer.doStop();
+
+ assertThat(getProcessedEventsByPartition()).isEmpty();
+ assertThat(getCheckpointTasksByPartition()).isEmpty();
+ assertThat(getScheduledTasksByPartition()).isEmpty();
+ verify(processorClient).stop();
+ }
+
+ private void invokeProcessCommit(Exchange exchange, EventContext
eventContext) throws Exception {
+ processCommit.invoke(consumer, exchange, eventContext);
+ }
+
+ private Exchange exchange() {
+ return DefaultExchange.newFromEndpoint(endpoint);
+ }
+
+ private EventContext eventContext(String partitionId) {
+ EventContext eventContext = mock(EventContext.class);
+ PartitionContext partitionContext = mock(PartitionContext.class);
+ when(partitionContext.getPartitionId()).thenReturn(partitionId);
+ when(eventContext.getPartitionContext()).thenReturn(partitionContext);
+ return eventContext;
+ }
+
+ @SuppressWarnings("unchecked")
+ private Map<String, AtomicInteger> getProcessedEventsByPartition() throws
Exception {
+ return (Map<String, AtomicInteger>) getField(consumer,
"processedEventsByPartition");
+ }
+
+ @SuppressWarnings("unchecked")
+ private Map<String, EventHubsCheckpointUpdaterTask>
getCheckpointTasksByPartition() throws Exception {
+ return (Map<String, EventHubsCheckpointUpdaterTask>)
getField(consumer, "checkpointTasksByPartition");
+ }
+
+ @SuppressWarnings("unchecked")
+ private Map<String, ScheduledFuture<?>> getScheduledTasksByPartition()
throws Exception {
+ return (Map<String, ScheduledFuture<?>>) getField(consumer,
"scheduledTasksByPartition");
+ }
+
+ private int processedEventsForPartition(String partitionId) throws
Exception {
+ AtomicInteger counter =
getProcessedEventsByPartition().get(partitionId);
+ return counter == null ? 0 : counter.get();
+ }
+
+ private static void setField(Object target, String name, Object value)
throws Exception {
+ var field = target.getClass().getDeclaredField(name);
+ field.setAccessible(true);
+ field.set(target, value);
+ }
+
+ private static Object getField(Object target, String name) throws
Exception {
+ var field = target.getClass().getDeclaredField(name);
+ field.setAccessible(true);
+ return field.get(target);
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_18.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_18.adoc
index 94b2681660c9..c9cb359db72a 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_18.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_18.adoc
@@ -11,6 +11,29 @@ Note that manual migration is still required.
See the xref:camel-upgrade-recipes-tool.adoc[documentation] page for details.
====
+== Upgrading from 4.18.5 to 4.18.6
+
+=== camel-azure-eventhubs - per-partition checkpoint counters and timer;
checkpoint window boundary fix
+
+Several fixes from 4.22.x have been brought forward to 4.18.6 to give this
change a sound base.
+
+`checkpointBatchSize` and `checkpointBatchTimeout` now count events per
partition, not across all
+partitions. Each partition has its own `AtomicInteger` counter and its own
`ScheduledFuture` timer.
+A checkpoint on one partition no longer resets the count of another. Before
this change, the shared
+counter mixed events across partitions, so the batch size was reached across
all partitions combined
+and a checkpoint on a busy partition could reset the count for a quiet one.
+
+The `CamelAzureEventHubsCheckpointUpdatedBy` header can now be `timeout`. 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. Routes that
read this header may now see
+`timeout` more often.
+
+Checkpoint failures 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.
+
== Upgrading from 4.18.4 to 4.18.5
=== camel-docling - output path handling
