diff --git a/components/camel-azure/camel-azure-eventhubs/pom.xml b/components/camel-azure/camel-azure-eventhubs/pom.xml index b0b9b51217732..81b496d5b8791 100644 --- a/components/camel-azure/camel-azure-eventhubs/pom.xml +++ b/components/camel-azure/camel-azure-eventhubs/pom.xml @@ -72,5 +72,10 @@ commons-lang3 test + + org.assertj + assertj-core + test + 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 49723ac7c4f79..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; @@ -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 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 ce1e6e79da56f..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 @@ -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 processedEventsByPartition = new ConcurrentHashMap<>(); + private final Map> scheduledTasksByPartition = new ConcurrentHashMap<>(); + private final Map 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 @@ protected void doStop() throws Exception { processorClient.stop(); } + processedEventsByPartition.clear(); + scheduledTasksByPartition.clear(); + checkpointTasksByPartition.clear(); + // shutdown camel consumer super.doStop(); } @@ -214,46 +220,71 @@ 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) { - 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/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/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/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 0000000000000..570833d609bce --- /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 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 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 getProcessedEventsByPartition() throws Exception { + return (Map) getField(consumer, "processedEventsByPartition"); + } + + @SuppressWarnings("unchecked") + private Map getCheckpointTasksByPartition() throws Exception { + return (Map) getField(consumer, "checkpointTasksByPartition"); + } + + @SuppressWarnings("unchecked") + private Map> getScheduledTasksByPartition() throws Exception { + return (Map>) 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); + } +} 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 94b2681660c9d..c9cb359db72a0 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 @@ -11,6 +11,29 @@ Note that manual migration is still required. See the xref:camel-upgrade-recipes-tool.adoc[documentation] page for details. ==== +== Upgrading from 4.18.5 to 4.18.6 + +=== camel-azure-eventhubs - per-partition checkpoint counters and timer; checkpoint window boundary fix + +Several fixes from 4.22.x have been brought forward to 4.18.6 to give this change a sound base. + +`checkpointBatchSize` and `checkpointBatchTimeout` now count events per partition, not across all +partitions. Each partition has its own `AtomicInteger` counter and its own `ScheduledFuture` timer. +A checkpoint on one partition no longer resets the count of another. Before this change, the shared +counter mixed events across partitions, so the batch size was reached across all partitions combined +and a checkpoint on a busy partition could reset the count for a quiet one. + +The `CamelAzureEventHubsCheckpointUpdatedBy` header can now be `timeout`. 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. Routes that read this header may now see +`timeout` more often. + +Checkpoint failures from the timer and from the batch-size trigger are now logged at `WARN` level with the full stack trace instead of +at `DEBUG` level with the message only. + == Upgrading from 4.18.4 to 4.18.5 === camel-docling - output path handling