This is an automated email from the ASF dual-hosted git repository. gnodet pushed a commit to branch backport/28135-to-camel-4.18.x in repository https://gitbox.apache.org/repos/asf/camel.git
commit 6f05afe2af0cdfef35f506c11f2cc8ecfe8ab568 Author: Guillaume Nodet <[email protected]> AuthorDate: Wed Oct 21 04:49:35 2026 +0000 [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 --- .../eventhubs/EventHubsCheckpointUpdaterTask.java | 77 ++++++++-- .../azure/eventhubs/EventHubsConsumer.java | 113 +++++++++----- .../EventHubsConsumerCheckpointWindowTest.java | 167 +++++++++++++++++++++ .../ROOT/pages/camel-4x-upgrade-guide-4_18.adoc | 13 ++ 4 files changed, 320 insertions(+), 50 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 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/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/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..7fa38fe6eeb2 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 @@ -136,6 +136,19 @@ The new behavior activates automatically when all of the following are true: No configuration changes are required. The existing `maxAutoLockRenewDuration` option controls how long Camel will continue renewing a message's lock. +=== 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-http - gzip double decompression with HttpClient 5.6+ When using HttpClient 5.6+ (including transitively via a Spring Boot BOM on Camel 4.18.x), HttpClient
