This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new c2962f91b523 CAMEL-25790: camel-google-pubsub - stop pulling once the
consumer stops, also after a restart (#28196)
c2962f91b523 is described below
commit c2962f91b523fe30124fc911e95827d9e2eee882
Author: allthingssecurity <[email protected]>
AuthorDate: Thu Oct 22 00:39:08 2026 +0530
CAMEL-25790: camel-google-pubsub - stop pulling once the consumer stops,
also after a restart (#28196)
* CAMEL-25790: camel-google-pubsub - stop pulling once the consumer stops,
also after a restart
deferShutdown (CAMEL-23291) stops the subscribers and cancels the
synchronous pulls, but the consumer stays started while the shutdown
strategy waits for its pending exchanges. The subscriber loops only
checked isRunAllowed(), so they subscribed or pulled again at once and
kept routing new messages during the graceful shutdown, which then waited
for its full timeout and forced the shutdown.
The same check let a loop of the previous start continue after a route
restart: its subscriber terminated after the new start, the loop saw the
consumer running and subscribed again, next to the loops of the new
start. Every restart added concurrentConsumers subscriber loops, and the
executors of the previous start stayed until doShutdown.
Each loop now belongs to a generation, incremented on start, on
deferShutdown and on stop; a loop of an older generation ends instead of
subscribing or pulling again. A restart also shuts down (not gracefully)
the executors of the previous start, whose loops end on their own.
* CAMEL-25790: camel-google-pubsub - address review: note the aborted
stopRoute case in deferShutdown
When a stopRoute with abortAfterTimeout=true times out, the route is
started again while the consumer is still started, so doStart does not
run and no subscriber loop pulls again until the route is stopped and
started.
* CAMEL-25790: camel-google-pubsub - address review: keep the retry-delay
scheduler across restarts, and check the release of the old executor
From Andrea Cosentino's review of #28196:
- doStart creates the PubSubReconnectTask scheduler only when there is none
and no longer shuts it
down on a restart; doShutdown still does. A loop of the previous start
that waits in the 5 s retry
delay (BackgroundTask.run waits on a latch that the scheduled task
releases) is then still woken
up and ends on the generation check, and a throttled route no longer gets
a new scheduler on
every cycle.
- asyncPullLoopOfPreviousStartEndsWhenConsumerIsRestarted verifies that the
consumer executor of the
first start is shut down through the ExecutorServiceManager and that the
scheduler is created
once. With the previous change the second check fails
(TooManyActualInvocations).
- The test class is package-private.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../google/pubsub/GooglePubsubConsumer.java | 50 ++++-
.../GooglePubsubConsumerStopPullingTest.java | 209 +++++++++++++++++++++
2 files changed, 251 insertions(+), 8 deletions(-)
diff --git
a/components/camel-google/camel-google-pubsub/src/main/java/org/apache/camel/component/google/pubsub/GooglePubsubConsumer.java
b/components/camel-google/camel-google-pubsub/src/main/java/org/apache/camel/component/google/pubsub/GooglePubsubConsumer.java
index c6198be41605..cbb0a5c9597b 100644
---
a/components/camel-google/camel-google-pubsub/src/main/java/org/apache/camel/component/google/pubsub/GooglePubsubConsumer.java
+++
b/components/camel-google/camel-google-pubsub/src/main/java/org/apache/camel/component/google/pubsub/GooglePubsubConsumer.java
@@ -27,6 +27,7 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
import com.google.api.core.AbstractApiService;
import com.google.api.core.ApiFuture;
@@ -67,6 +68,12 @@ public class GooglePubsubConsumer extends DefaultConsumer
implements ShutdownAwa
private final GooglePubsubEndpoint endpoint;
private final Processor processor;
private final AtomicInteger pendingExchanges = new AtomicInteger();
+ /**
+ * The subscriber loops started by the current start of this consumer.
Incremented when the consumer starts, and
+ * when it stops pulling (deferShutdown, doStop): a loop of an earlier
generation ends instead of subscribing or
+ * pulling again, also when the consumer has been started again meanwhile.
+ */
+ private final AtomicLong generation = new AtomicLong();
private ExecutorService executor;
private ScheduledExecutorService taskExecutor;
private final List<Subscriber> subscribers;
@@ -102,11 +109,22 @@ public class GooglePubsubConsumer extends DefaultConsumer
implements ShutdownAwa
resolvedMaxDeliveryAttempts,
endpoint.getDestinationName());
}
+ // started again after a stop: the executor of the previous start is
otherwise only shut down in doShutdown.
+ // Its subscriber loops end on their own (generation), so a regular
shutdown lets them finish and then
+ // releases the threads
+ if (executor != null) {
+
endpoint.getCamelContext().getExecutorServiceManager().shutdown(executor);
+ }
executor = endpoint.createExecutor(this);
- taskExecutor = endpoint.getCamelContext().getExecutorServiceManager()
- .newSingleThreadScheduledExecutor(this, "PubSubReconnectTask");
+ // the scheduler of the retry delay is kept until doShutdown: a loop
of the previous start that waits in that
+ // delay must still be woken up, and then ends on the generation check
+ if (taskExecutor == null) {
+ taskExecutor =
endpoint.getCamelContext().getExecutorServiceManager()
+ .newSingleThreadScheduledExecutor(this,
"PubSubReconnectTask");
+ }
+ long current = generation.incrementAndGet();
for (int i = 0; i < endpoint.getConcurrentConsumers(); i++) {
- executor.submit(new SubscriberWrapper());
+ executor.submit(new SubscriberWrapper(current));
}
}
@@ -114,6 +132,7 @@ public class GooglePubsubConsumer extends DefaultConsumer
implements ShutdownAwa
protected void doStop() throws Exception {
localLog.info("Stopping Google PubSub consumer for {}/{}",
endpoint.getProjectId(), endpoint.getDestinationName());
+ generation.incrementAndGet();
if (!subscribers.isEmpty()) {
localLog.info("Stopping subscribers for {}/{}",
endpoint.getProjectId(), endpoint.getDestinationName());
subscribers.forEach(AbstractApiService::stopAsync);
@@ -144,6 +163,11 @@ public class GooglePubsubConsumer extends DefaultConsumer
implements ShutdownAwa
@Override
public boolean deferShutdown(ShutdownRunningTask shutdownRunningTask) {
+ // stop pulling: the consumer stays started until the pending
exchanges are done, so the subscriber loops must
+ // not subscribe or pull again when their subscriber terminates or
their pull is cancelled.
+ // Note: when a stopRoute with abortAfterTimeout=true times out, the
route is started again while this consumer
+ // is still started, so doStart does not run and no loop pulls again
until the route is stopped and started
+ generation.incrementAndGet();
if (!subscribers.isEmpty()) {
subscribers.forEach(AbstractApiService::stopAsync);
}
@@ -217,11 +241,21 @@ public class GooglePubsubConsumer extends DefaultConsumer
implements ShutdownAwa
private class SubscriberWrapper implements Runnable {
private final String subscriptionName;
+ private final long wrapperGeneration;
- SubscriberWrapper() {
+ SubscriberWrapper(long wrapperGeneration) {
+ this.wrapperGeneration = wrapperGeneration;
subscriptionName =
ProjectSubscriptionName.format(endpoint.getProjectId(),
endpoint.getDestinationName());
}
+ /**
+ * Whether this loop may subscribe or pull (again): the consumer runs
and has not stopped pulling or been
+ * started again since this loop was started.
+ */
+ private boolean isPullAllowed() {
+ return isRunAllowed() && !isSuspendingOrSuspended() &&
wrapperGeneration == generation.get();
+ }
+
@Override
public void run() {
try {
@@ -250,7 +284,7 @@ public class GooglePubsubConsumer extends DefaultConsumer
implements ShutdownAwa
}
private void asynchronousPull(String subscriptionName) throws
IOException {
- while (isRunAllowed() && !isSuspendingOrSuspended()) {
+ while (isPullAllowed()) {
MessageReceiver messageReceiver = new
CamelMessageReceiver(GooglePubsubConsumer.this, endpoint, processor);
Subscriber subscriber =
endpoint.getComponent().getSubscriber(subscriptionName, messageReceiver,
endpoint);
@@ -261,7 +295,7 @@ public class GooglePubsubConsumer extends DefaultConsumer
implements ShutdownAwa
subscribers.add(subscriber);
subscriberAdded = true;
// a stop while the subscriber was starting missed it;
stop it here or awaitTerminated never returns
- if (!isRunAllowed() || isSuspendingOrSuspended()) {
+ if (!isPullAllowed()) {
subscriber.stopAsync();
}
subscriber.awaitTerminated();
@@ -320,7 +354,7 @@ public class GooglePubsubConsumer extends DefaultConsumer
implements ShutdownAwa
}
private void synchronousPull(String subscriptionName) throws
ExecutionException, InterruptedException {
- while (isRunAllowed() && !isSuspendingOrSuspended()) {
+ while (isPullAllowed()) {
ApiFuture<PullResponse> synchronousPullResponseFuture = null;
try (SubscriberStub subscriber =
endpoint.getComponent().getSubscriberStub(endpoint)) {
@@ -333,7 +367,7 @@ public class GooglePubsubConsumer extends DefaultConsumer
implements ShutdownAwa
synchronousPullResponseFuture =
subscriber.pullCallable().futureCall(pullRequest);
pendingSynchronousPullResponses.add(synchronousPullResponseFuture);
// a stop while the pull was being issued missed this
future; cancel it here or get() blocks
- if (!isRunAllowed() || isSuspendingOrSuspended()) {
+ if (!isPullAllowed()) {
synchronousPullResponseFuture.cancel(true);
}
PullResponse pullResponse =
synchronousPullResponseFuture.get();
diff --git
a/components/camel-google/camel-google-pubsub/src/test/java/org/apache/camel/component/google/pubsub/GooglePubsubConsumerStopPullingTest.java
b/components/camel-google/camel-google-pubsub/src/test/java/org/apache/camel/component/google/pubsub/GooglePubsubConsumerStopPullingTest.java
new file mode 100644
index 000000000000..de56be730fd7
--- /dev/null
+++
b/components/camel-google/camel-google-pubsub/src/test/java/org/apache/camel/component/google/pubsub/GooglePubsubConsumerStopPullingTest.java
@@ -0,0 +1,209 @@
+/*
+ * 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.google.pubsub;
+
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import com.google.api.core.SettableApiFuture;
+import com.google.api.gax.rpc.UnaryCallable;
+import com.google.cloud.pubsub.v1.Subscriber;
+import com.google.cloud.pubsub.v1.stub.SubscriberStub;
+import com.google.pubsub.v1.PullRequest;
+import com.google.pubsub.v1.PullResponse;
+import org.apache.camel.CamelContext;
+import org.apache.camel.ExtendedCamelContext;
+import org.apache.camel.Processor;
+import org.apache.camel.ShutdownRunningTask;
+import org.apache.camel.spi.ExchangeFactory;
+import org.apache.camel.spi.ExecutorServiceManager;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+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.anyString;
+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;
+
+/**
+ * Once the consumer stops pulling (deferShutdown during a graceful shutdown,
or a stop), a subscriber loop must not
+ * subscribe or pull again when its subscriber terminates or its pull is
cancelled: not while the consumer waits for its
+ * pending exchanges, and not after the consumer was started again (route
restart), which starts its own loops.
+ */
+class GooglePubsubConsumerStopPullingTest {
+
+ private final GooglePubsubEndpoint endpoint = mock();
+ private final GooglePubsubComponent component = mock();
+ private final Processor processor = mock();
+ private final CamelContext context = mock();
+ private final ExtendedCamelContext ecc = mock();
+ private final ExchangeFactory ef = mock();
+ private final ExecutorServiceManager executorServiceManager = mock();
+
+ private final List<ExecutorService> executors = new
CopyOnWriteArrayList<>();
+ // subscribers created and subscribers between awaitRunning and the end of
awaitTerminated
+ private final AtomicInteger created = new AtomicInteger();
+ private final AtomicInteger running = new AtomicInteger();
+ // the first subscriber completes its termination (stream close, ack
flush) only when this is released
+ private final CountDownLatch firstTermination = new CountDownLatch(1);
+
+ @BeforeEach
+ void setUp() throws Exception {
+ when(endpoint.getCamelContext()).thenReturn(context);
+ when(context.getCamelContextExtension()).thenReturn(ecc);
+ when(ecc.getExchangeFactory()).thenReturn(ef);
+ when(ef.newExchangeFactory(any())).thenReturn(ef);
+
when(context.getExecutorServiceManager()).thenReturn(executorServiceManager);
+ when(executorServiceManager.newSingleThreadScheduledExecutor(any(),
anyString())).thenAnswer(invocation -> {
+ ScheduledExecutorService taskExecutor =
Executors.newSingleThreadScheduledExecutor();
+ executors.add(taskExecutor);
+ return taskExecutor;
+ });
+
+ when(endpoint.getComponent()).thenReturn(component);
+ when(endpoint.createExecutor(any())).thenAnswer(invocation -> {
+ ExecutorService consumerExecutor =
Executors.newSingleThreadExecutor();
+ executors.add(consumerExecutor);
+ return consumerExecutor;
+ });
+ when(endpoint.getConcurrentConsumers()).thenReturn(1);
+ when(endpoint.getMaxMessagesPerPoll()).thenReturn(1);
+ when(endpoint.getProjectId()).thenReturn("test-project");
+ when(endpoint.getDestinationName()).thenReturn("test-subscription");
+ when(endpoint.isMaxDeliveryAttemptsExplicitlySet()).thenReturn(true);
+ when(endpoint.getMaxDeliveryAttempts()).thenReturn(0);
+
+ when(component.getSubscriber(anyString(), any(), any()))
+ .thenAnswer(invocation ->
newSubscriber(created.incrementAndGet() == 1));
+ }
+
+ @AfterEach
+ void tearDown() {
+ firstTermination.countDown();
+ executors.forEach(ExecutorService::shutdownNow);
+ }
+
+ private Subscriber newSubscriber(boolean first) {
+ Subscriber subscriber = mock();
+ CountDownLatch stopRequested = new CountDownLatch(1);
+ when(subscriber.startAsync()).thenReturn(subscriber);
+ doAnswer(invocation -> {
+ running.incrementAndGet();
+ return null;
+ }).when(subscriber).awaitRunning();
+ doAnswer(invocation -> {
+ stopRequested.countDown();
+ return subscriber;
+ }).when(subscriber).stopAsync();
+ doAnswer(invocation -> {
+ stopRequested.await();
+ if (first) {
+ firstTermination.await();
+ }
+ running.decrementAndGet();
+ return null;
+ }).when(subscriber).awaitTerminated();
+ return subscriber;
+ }
+
+ @Test
+ void asyncPullDoesNotSubscribeAgainAfterDeferShutdown() throws Exception {
+ firstTermination.countDown();
+ GooglePubsubConsumer consumer = new GooglePubsubConsumer(endpoint,
processor);
+ consumer.start();
+ try {
+ await().atMost(5, TimeUnit.SECONDS).until(() -> running.get() ==
1);
+
+ // graceful shutdown: the consumer stays started while the
shutdown strategy waits for its pending exchanges
+
consumer.deferShutdown(ShutdownRunningTask.CompleteCurrentTaskOnly);
+ // the subscriber ends, and the loop does not start another one
while the shutdown waits
+ await().during(1, TimeUnit.SECONDS).atMost(5,
TimeUnit.SECONDS).untilAsserted(() -> {
+ assertEquals(1, created.get(), "subscribers created after the
consumer stopped pulling");
+ assertEquals(0, running.get(), "subscribers running while the
shutdown waits for the pending exchanges");
+ });
+ } finally {
+ consumer.stop();
+ }
+ }
+
+ @Test
+ void asyncPullLoopOfPreviousStartEndsWhenConsumerIsRestarted() throws
Exception {
+ GooglePubsubConsumer consumer = new GooglePubsubConsumer(endpoint,
processor);
+ consumer.start();
+ try {
+ await().atMost(5, TimeUnit.SECONDS).until(() -> running.get() ==
1);
+
+ // route restart: stop, then start again before the first
subscriber has terminated
+ ExecutorService firstConsumerExecutor = executors.get(0);
+ consumer.stop();
+ consumer.start();
+ await().atMost(5, TimeUnit.SECONDS).until(() -> created.get() ==
2);
+ // the executor of the first start is released, and the scheduler
of the retry delay is kept
+ verify(executorServiceManager).shutdown(firstConsumerExecutor);
+ verify(executorServiceManager,
times(1)).newSingleThreadScheduledExecutor(any(), anyString());
+
+ firstTermination.countDown();
+ // the loop of the first start ends once its subscriber
terminated, and does not subscribe again
+ await().during(1, TimeUnit.SECONDS).atMost(5,
TimeUnit.SECONDS).untilAsserted(() -> {
+ assertEquals(2, created.get(), "subscribers created by a start
and a restart with one concurrent consumer");
+ assertEquals(1, running.get(), "subscribers running after the
restart");
+ });
+ } finally {
+ consumer.stop();
+ }
+ }
+
+ @Test
+ void syncPullDoesNotPullAgainAfterDeferShutdown() throws Exception {
+ when(endpoint.isSynchronousPull()).thenReturn(true);
+ AtomicInteger pulls = new AtomicInteger();
+ SubscriberStub subscriberStub = mock();
+ @SuppressWarnings("unchecked")
+ UnaryCallable<PullRequest, PullResponse> pullCallable =
mock(UnaryCallable.class);
+ when(subscriberStub.pullCallable()).thenReturn(pullCallable);
+ doAnswer(invocation -> {
+ pulls.incrementAndGet();
+ // a pull that waits for messages until it is cancelled
+ return SettableApiFuture.<PullResponse> create();
+ }).when(pullCallable).futureCall(any(PullRequest.class));
+ when(component.getSubscriberStub(any())).thenReturn(subscriberStub);
+
+ GooglePubsubConsumer consumer = new GooglePubsubConsumer(endpoint,
processor);
+ consumer.start();
+ try {
+ await().atMost(5, TimeUnit.SECONDS).until(() -> pulls.get() == 1);
+
+
consumer.deferShutdown(ShutdownRunningTask.CompleteCurrentTaskOnly);
+ await().during(1, TimeUnit.SECONDS).atMost(5, TimeUnit.SECONDS)
+ .untilAsserted(() -> assertEquals(1, pulls.get(), "pulls
issued after the consumer stopped pulling"));
+ } finally {
+ consumer.stop();
+ }
+ }
+}
