CAMEL-25781: camel-azure-eventhubs - an event that completes when the checkpoint timer of its window has run is checkpointed by the next timer

components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsCheckpointUpdaterTask.java

@@ -17,6 +17,8 @@
 package org.apache.camel.component.azure.eventhubs;
 
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 
 import com.azure.messaging.eventhubs.models.EventContext;
 import org.slf4j.Logger;
@@ -32,20 +34,53 @@ public class EventHubsCheckpointUpdaterTask implements Runnable {
     private volatile EventContext eventContext;
     private final AtomicInteger processedEvents;
     private volatile long scheduledTime;
+    // the lock of the consumer, which counts the events under it
+    private final Lock lock;
+    // set when the task has taken the count of its window, or was cancelled; read and written under the lock only
+    private boolean done;
 
+    /**
+     * Creates a task that synchronizes on a private lock of its own, unrelated to any consumer: for standalone or test
+     * use. {@link EventHubsConsumer} uses {@link #EventHubsCheckpointUpdaterTask(EventContext, AtomicInteger, Lock)}
+     * with its own lock, which it holds while it counts the events; with a private lock an event could be counted in a
+     * window whose timer has already taken the count.
+     *
+     * @param eventContext    the event context to checkpoint
+     * @param processedEvents the count of the events processed since the last checkpoint
+     */
     public EventHubsCheckpointUpdaterTask(EventContext eventContext, AtomicInteger processedEvents) {
+        this(eventContext, processedEvents, new ReentrantLock());
+    }
+
+    public EventHubsCheckpointUpdaterTask(EventContext eventContext, AtomicInteger processedEvents, Lock lock) {
         this.eventContext = eventContext;
         this.processedEvents = processedEvents;
+        this.lock = lock;
     }
 
     @Override
     public void run() {
         // take the count when the checkpoint is started: an event that completes while it is being stored is counted
-        // for the next window from 0, and a checkpoint by batch size in the meantime cannot be confused with this one
-        final int processed = processedEvents.getAndSet(0);
+        // for the next window from 0, and a checkpoint by batch size in the meantime cannot be confused with this one.
+        // This is done under the lock of the consumer, so an event is either counted (and its context set) before, and
+        // checkpointed here, or after, and then it sees that this task is done and starts a new window.
+        final int processed;
+        final EventContext context;
+        lock.lock();
+        try {
+            if (done) {
+                LOG.debug("skip checkpointing offset: the window was already checkpointed by batch size");
+                return;
+            }
+            done = true;
+            processed = processedEvents.getAndSet(0);
+            context = eventContext;
+        } finally {
+            lock.unlock();
+        }
         if (processed > 0) {
             LOG.debug("checkpointing offset after reaching timeout, with a batch of {}", processed);
-            eventContext.updateCheckpointAsync()
+            context.updateCheckpointAsync()
                     .subscribe(unused -> LOG.debug("Processed one event..."),
                             error -> {
                                 LOG.warn("Error when updating Checkpoint: {}", error.getMessage(), error);
@@ -70,6 +105,26 @@ public long getScheduledTime() {
         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;
     }

components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumer.java

@@ -220,9 +220,10 @@ private void onErrorListener(final ErrorContext errorContext) {
     /**
      * Strategy to commit the offset after message being processed successfully.
      *
-     * @param exchange the exchange
+     * @param exchange     the exchange
+     * @param eventContext the event context of the event, used to update the checkpoint
      */
-    private void processCommit(final Exchange exchange, final EventContext eventContext) {
+    void processCommit(final Exchange exchange, final EventContext eventContext) {
         lock.lock();
         try {
             final String partitionId = eventContext.getPartitionContext().getPartitionId();
@@ -231,8 +232,11 @@ private void processCommit(final Exchange exchange, final EventContext eventCont
             EventHubsCheckpointUpdaterTask checkpointTask = checkpointTasksByPartition.get(partitionId);
             ScheduledFuture<?> scheduledTask = scheduledTasksByPartition.get(partitionId);
 
-            if (checkpointTask == null || checkpointTask.isExpired()) {
-                checkpointTask = new EventHubsCheckpointUpdaterTask(eventContext, processedEvents);
+            // a new window starts when the timer of the current one has run, not when the wall clock passed its time: the
+            // timer can run in the millisecond of its scheduled time, or later, and the event must be counted in the
+            // window whose timer has not taken the count yet
+            if (checkpointTask == null || checkpointTask.isDone()) {
+                checkpointTask = new EventHubsCheckpointUpdaterTask(eventContext, processedEvents, lock);
                 // delegate the checkpoint update to a dedicated Thread
                 long timeout = getConfiguration().getCheckpointBatchTimeout();
                 checkpointTask.setScheduledTime(System.currentTimeMillis() + timeout);
@@ -251,6 +255,8 @@ private void processCommit(final Exchange exchange, final EventContext eventCont
                     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);
                     }

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;
+    }
+}

components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/dst/EventHubsCheckpointDstTest.java

@@ -20,6 +20,7 @@
 
 import org.apache.camel.test.dst.Dst;
 import org.apache.camel.test.dst.DstInstrument;
+import org.apache.camel.test.dst.DstSeed;
 import org.apache.camel.test.dst.DstTest;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.Test;
@@ -32,12 +33,9 @@
  * partition goes quiet. Regression test for CAMEL-25683 (apache/camel#27922): before that fix 2064 of 5000 seeds fail.
  * A failing seed is replayed with {@code -Ddst.seed=<seed>}; {@code -Ddst.seeds=5000} runs more seeds.
  * <p>
- * Known residual on main, outside the default 1000 seeds: seed 3042 of 5000 fails when an event completes in the same
- * millisecond in which the timer of its window ran ({@code EventHubsCheckpointUpdaterTask.isExpired()} compares with
- * {@code >}, so the window is reused and that event waits for the next one; if the partition stays quiet it is
- * delivered again after a restart). In real time the window is narrow but not only that millisecond: the timer task
- * does not take the consumer's lock, so it can also run between the {@code isExpired()} check and the count of the
- * event. Virtual time makes the same-millisecond case far more likely than it is in real time.
+ * Seed 3042 (of 5000) failed before CAMEL-25781 and is pinned below: an event that completed in the millisecond in
+ * which the timer of its window ran joined that window after the timer had taken its count, so it was not checkpointed
+ * until the next event (or delivered again after a restart if the partition stayed quiet).
  */
 @DstInstrument("org.apache.camel.component.azure.eventhubs")
 class EventHubsCheckpointDstTest {
@@ -47,6 +45,16 @@ void everyCompletedEventIsCheckpointedWhenThePartitionGoesQuiet(Dst dst) throws
         new EventHubsWorld(dst).run();
     }
 
+    /**
+     * CAMEL-25781: an event completes in the millisecond in which the timer of its window ran, and must be checkpointed
+     * by the next timer.
+     */
+    @DstTest
+    @DstSeed(3042)
+    void eventThatCompletesWhenTheTimerOfItsWindowRanIsCheckpointed(Dst dst) throws Exception {
+        new EventHubsWorld(dst).run();
+    }
+
     /** The same seed twice in this JVM: identical step traces (the cross-JVM check compares the trace files). */
     @Test
     void replayOfASeedGivesTheSameTrace() throws Exception {

docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc

@@ -1681,6 +1681,19 @@ exception handler of the consumer, with the blob name in the message (by default
 poll that lists the blob). Before, the poll failed, and as every poll lists the container again, one such blob stopped
 the consumer from consuming any blob. The poll still fails when none of the listed blobs could be got.
 
+=== camel-azure-eventhubs - an event that completes when the checkpoint timer of its window has run is checkpointed
+
+With `checkpointBatchTimeout`, the consumer now starts a new checkpoint window when the timer of the current window
+has run, and no longer when the wall clock passes the time the timer was scheduled for. Before, an event that
+completed in the millisecond in which the timer ran (or while the timer was taking the count) was counted in the
+window whose count had already been taken: it was checkpointed only with the next event, and when the partition stayed
+quiet it was delivered again after a restart.
+
+The `CamelAzureEventHubsCheckpointUpdatedBy` header can now be `timeout`. An event that completes after the
+scheduled time of its window, but before the (late) timer has run, joins that window and gets `timeout`; its offset is
+checkpointed by that timer. Before, such an event always started a new window, so `timeout` was practically never
+set. Routes that read this header can now see `timeout` as well as `size`.
+
 === camel-azure-servicebus - the dead-letter reason and description are bounded
 
 When the consumer dead-letters a message after a failed route with `enableDeadLettering=true`, the reason and the