From 7d65de039a5497d8d71a892ea48dc01015e28b8b Mon Sep 17 00:00:00 2001 From: Guillaume Nodet Date: Wed, 21 Oct 2026 04:49:33 +0000 Subject: [PATCH 1/2] [backport camel-4.22.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 --- .../EventHubsCheckpointUpdaterTask.java | 75 +++++++- .../azure/eventhubs/EventHubsConsumer.java | 14 +- ...EventHubsConsumerCheckpointWindowTest.java | 167 ++++++++++++++++++ .../pages/camel-4x-upgrade-guide-4_22.adoc | 13 ++ 4 files changed, 257 insertions(+), 12 deletions(-) create mode 100644 components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumerCheckpointWindowTest.java 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 e8a5355675d3f..1de96d0c19ab7 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; @@ -32,23 +34,60 @@ 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() { - 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.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"); } @@ -66,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; } 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 7f0ca1bf32598..10191379f8170 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 @@ -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); @@ -250,6 +254,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); } 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 0000000000000..3fc1b5525fb3b --- /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 scheduledTasks() { + ArgumentCaptor tasks = ArgumentCaptor.forClass(Runnable.class); + verify(scheduledExecutorService, atLeastOnce()) + .schedule(tasks.capture(), anyLong(), any(TimeUnit.class)); + return tasks.getAllValues(); + } + + private void runTasksFrom(int index) { + List 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_22.adoc b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc index e4e4834f1758e..b69ccec0eb69e 100644 --- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc +++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc @@ -1409,6 +1409,19 @@ filtering performed by other components. Ordinary application headers are unaffected. +=== 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-storage-datalake - openInputStream no longer uses Blob Query API The `getFile` operation on the Data Lake consumer/producer now uses the standard From dbc6191a111e7b6f9c3d4f8dbcfcee3d0feb3cca Mon Sep 17 00:00:00 2001 From: gnodet Date: Wed, 21 Oct 2026 11:42:50 +0000 Subject: [PATCH 2/2] Address review: bring CAMEL-25683 consumer changes to backport - Change batch-size check from == to >= (a failed timeout checkpoint adds its count back, so the count can pass the batch size) - Remove cancelled task from checkpointTasksByPartition and scheduledTasksByPartition after a checkpoint by batch size - Add EventHubsConsumerCheckpointAfterBatchSizeTest - Move upgrade note to a new '4.22.2 to 4.22.3' section --- .../azure/eventhubs/EventHubsConsumer.java | 7 +- ...sConsumerCheckpointAfterBatchSizeTest.java | 229 ++++++++++++++++++ .../pages/camel-4x-upgrade-guide-4_22.adoc | 28 ++- 3 files changed, 250 insertions(+), 14 deletions(-) create mode 100644 components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumerCheckpointAfterBatchSizeTest.java 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 10191379f8170..eddf97f3a7b22 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 @@ -250,7 +250,8 @@ void processCommit(final Exchange exchange, final EventContext eventContext) { 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); @@ -259,6 +260,10 @@ void processCommit(final Exchange exchange, final EventContext eventContext) { 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 0000000000000..19a22d70484b6 --- /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 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 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 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/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc index b69ccec0eb69e..cd996a55a4fe2 100644 --- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc +++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc @@ -11,6 +11,21 @@ Note that manual migration is still required. See the xref:camel-upgrade-recipes-tool.adoc[documentation] page for details. ==== +== Upgrading from 4.22.2 to 4.22.3 + +=== 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`. + == Upgrading from 4.22.1 to 4.22.2 === camel-management - Route group exchange counters count once per entry into the group again @@ -1409,19 +1424,6 @@ filtering performed by other components. Ordinary application headers are unaffected. -=== 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-storage-datalake - openInputStream no longer uses Blob Query API The `getFile` operation on the Data Lake consumer/producer now uses the standard