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 03cb654a4335 CAMEL-24198: Track Event Hubs checkpoint batching per
partition
03cb654a4335 is described below
commit 03cb654a4335fe35e1de81d237d32630184edbd0
Author: Omar Atie <[email protected]>
AuthorDate: Sun Jul 19 01:41:15 2026 -0700
CAMEL-24198: Track Event Hubs checkpoint batching per partition
The consumer used shared processedEvents, lastTask, and lastScheduledTask
fields across all partitions, so checkpoint commits from different
partitions could interfere with each other. Replaced these with
per-partition ConcurrentHashMap fields so each partition tracks its own
batch count and schedules its own timeout checkpoint independently.
Closes #24903
---
.../camel-azure/camel-azure-eventhubs/pom.xml | 5 +
.../azure/eventhubs/EventHubsConsumer.java | 46 +++--
...ventHubsConsumerPerPartitionCheckpointTest.java | 199 +++++++++++++++++++++
3 files changed, 233 insertions(+), 17 deletions(-)
diff --git a/components/camel-azure/camel-azure-eventhubs/pom.xml
b/components/camel-azure/camel-azure-eventhubs/pom.xml
index e5468bee1745..638eae7eafad 100644
--- a/components/camel-azure/camel-azure-eventhubs/pom.xml
+++ b/components/camel-azure/camel-azure-eventhubs/pom.xml
@@ -65,6 +65,11 @@
<artifactId>camel-test-junit6</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.assertj</groupId>
+ <artifactId>assertj-core</artifactId>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-core</artifactId>
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 2f3c6c3c5827..21a04f735b7e 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,6 +16,8 @@
*/
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;
@@ -46,16 +48,14 @@ 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 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 +82,10 @@ public class EventHubsConsumer extends DefaultConsumer
implements ShutdownAware
processorClient.stop();
}
+ processedEventsByPartition.clear();
+ scheduledTasksByPartition.clear();
+ checkpointTasksByPartition.clear();
+
// shutdown camel consumer
super.doStop();
}
@@ -210,15 +214,23 @@ public class EventHubsConsumer extends DefaultConsumer
implements ShutdownAware
* @param exchange the exchange
*/
private synchronized void processCommit(final Exchange exchange, final
EventContext eventContext) {
- if (lastTask == null || lastTask.isExpired()) {
- lastTask = new EventHubsCheckpointUpdaterTask(eventContext,
processedEvents);
+ 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);
+
+ if (checkpointTask == null || checkpointTask.isExpired()) {
+ checkpointTask = 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);
+ checkpointTask.setScheduledTime(System.currentTimeMillis() +
timeout);
+ scheduledTask = scheduledExecutorService.schedule(checkpointTask,
timeout, TimeUnit.MILLISECONDS);
+ checkpointTasksByPartition.put(partitionId, checkpointTask);
+ scheduledTasksByPartition.put(partitionId, scheduledTask);
} else {
// updates the eventContext to use for the offset to be the most
accurate
- lastTask.setEventContext(eventContext);
+ checkpointTask.setEventContext(eventContext);
}
try {
@@ -226,9 +238,9 @@ public class EventHubsConsumer extends DefaultConsumer
implements ShutdownAware
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);
+ LOG.debug("eventhub consumer batch size of reached for
partition {}", partitionId);
+ if (scheduledTask != null) {
+ scheduledTask.cancel(false);
}
eventContext.updateCheckpointAsync()
.subscribe(unused -> LOG.debug("Processed one
event..."),
@@ -236,12 +248,12 @@ public class EventHubsConsumer extends DefaultConsumer
implements ShutdownAware
() -> {
LOG.debug("Checkpoint updated.");
});
- } else if (lastTask != null && lastTask.isExpired()) {
+ } else if (checkpointTask.isExpired()) {
exchange.getIn().setHeader(EventHubsConstants.CHECKPOINT_UPDATED_BY,
COMPLETED_BY_TIMEOUT);
- LOG.debug("eventhub consumer batch timeout reached");
+ LOG.debug("eventhub consumer batch timeout reached for
partition {}", partitionId);
} else {
- LOG.debug("neither eventhub consumer batch size of {}/{} nor
batch timeout reached yet", cnt,
- getConfiguration().getCheckpointBatchSize());
+ 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) {
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);
+ }
+}
