allthingssecurity commented on code in PR #28135:
URL: https://github.com/apache/camel/pull/28135#discussion_r4322748740
##########
components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumer.java:
##########
@@ -222,7 +222,7 @@ private void onErrorListener(final ErrorContext
errorContext) {
*
* @param exchange the exchange
*/
- private void processCommit(final Exchange exchange, final EventContext
eventContext) {
+ void processCommit(final Exchange exchange, final EventContext
eventContext) {
Review Comment:
Added `@param eventContext` (and aligned `@param exchange`) in dbd7616e5f9d.
_Claude Code on behalf of allthingssecurity_
##########
components/camel-azure/camel-azure-eventhubs/src/main/java/org/apache/camel/component/azure/eventhubs/EventHubsCheckpointUpdaterTask.java:
##########
@@ -32,20 +34,44 @@ 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;
public EventHubsCheckpointUpdaterTask(EventContext eventContext,
AtomicInteger processedEvents) {
+ this(eventContext, processedEvents, new ReentrantLock());
Review Comment:
Accurate: only the tests use the 2-arg constructor, `EventHubsConsumer` uses
the one that takes its lock. Added a Javadoc note on the 2-arg constructor in
dbd7616e5f9d.
_Claude Code on behalf of allthingssecurity_
##########
components/camel-azure/camel-azure-eventhubs/src/test/java/org/apache/camel/component/azure/eventhubs/EventHubsConsumerCheckpointWindowTest.java:
##########
@@ -0,0 +1,160 @@
+/*
+ * 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.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);
+
+ verify(third).updateCheckpointAsync();
Review Comment:
Added in dbd7616e5f9d, with one adjustment: when `second` is counted it
becomes the event context of the first window's timer, so a cancelled timer
that checkpoints again would checkpoint `second`, not `first`. The test now
asserts `verify(first, never())` and `verify(second, times(1))` after the
timers run. With `cancel()` not marking the task done it fails with "Wanted 1
time, but was 2 times" on `second`; with the fix it passes (module: 1040 tests,
0 failures).
_Claude Code on behalf of allthingssecurity_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
