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 83aaa80754f9 CAMEL-25683: camel-azure-eventhubs - after a batch size
checkpoint, the batch timeout checkpoints again (#27922)
83aaa80754f9 is described below
commit 83aaa80754f998c5b298b942c6b4ddb88f0f75ec
Author: allthingssecurity <[email protected]>
AuthorDate: Sat Oct 17 01:28:36 2026 +0530
CAMEL-25683: camel-azure-eventhubs - after a batch size checkpoint, the
batch timeout checkpoints again (#27922)
When checkpointBatchSize was reached, processCommit cancelled the
scheduled timeout checkpoint of the partition but left the task
registered. Until the time that task had been scheduled for, the next
events of the partition only updated that cancelled task, so nothing
checkpointed them by time. At the end of a burst those events stayed
uncheckpointed until the next event, and a restart or a partition moving
to another consumer delivered them again (up to checkpointBatchSize - 1
per partition; with the defaults, a burst of 700 events within 5 seconds
left events 501-700 uncheckpointed).
* The cancelled task is now removed with its future, so the next event
starts a new timeout window, as the first event after start does.
* The timeout task reset the per-partition counter when its checkpoint
was stored, without the lock, which could wipe the count of the next
window. It now takes the count when the checkpoint starts and adds it
back if storing the checkpoint fails, and the batch size check also
checkpoints when that count went past the batch size.
No option or default changes.
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../eventhubs/EventHubsCheckpointUpdaterTask.java | 18 +-
.../azure/eventhubs/EventHubsConsumer.java | 7 +-
...ntHubsConsumerCheckpointAfterBatchSizeTest.java | 229 +++++++++++++++++++++
3 files changed, 246 insertions(+), 8 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 e8a5355675d3..90ee2924f8e3 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
@@ -40,15 +40,19 @@ public class EventHubsCheckpointUpdaterTask implements
Runnable {
@Override
public void run() {
- if (processedEvents.get() > 0) {
- LOG.debug("checkpointing offset after reaching timeout, with a
batch of {}", processedEvents.get());
+ // 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);
+ if (processed > 0) {
+ LOG.debug("checkpointing offset after reaching timeout, with a
batch of {}", processed);
eventContext.updateCheckpointAsync()
.subscribe(unused -> LOG.debug("Processed one event..."),
- error -> LOG.warn("Error when updating Checkpoint:
{}", error.getMessage(), error),
- () -> {
- 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");
}
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 7f0ca1bf3259..a1f5fce2385a 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
@@ -246,13 +246,18 @@ public class EventHubsConsumer extends DefaultConsumer
implements ShutdownAware
try {
var cnt = processedEvents.incrementAndGet();
- if (cnt == getConfiguration().getCheckpointBatchSize()) {
+ // >= 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);
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),
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;
+ }
+}
