This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch camel-4.18.x
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/camel-4.18.x by this push:
new e319fd1ada11 CAMEL-24315: camel-google-pubsub - stop subscribers that
finish starting after the consumer stopped
e319fd1ada11 is described below
commit e319fd1ada11a60f048a8212fb6416d89132ccf3
Author: Claus Ibsen <[email protected]>
AuthorDate: Mon Aug 3 09:43:14 2026 +0200
CAMEL-24315: camel-google-pubsub - stop subscribers that finish starting
after the consumer stopped
Backport of #25248. Fixes a race condition where a subscriber or
synchronous pull still starting when the consumer stops gets missed by
doStop(). The wrapper thread then parks forever in awaitTerminated() or
get(), leaking threads and causing up to 20s shutdown delays per stuck
consumer.
After registering the subscriber or pull future, re-check the consumer
state and stop/cancel it if the consumer is no longer allowed to run.
Closes #25287
Co-authored-by: Claude Opus 4.6 <[email protected]>
---
.../google/pubsub/GooglePubsubConsumer.java | 8 +
.../pubsub/GooglePubsubConsumerStopRaceTest.java | 166 +++++++++++++++++++++
2 files changed, 174 insertions(+)
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 e6fed0419d68..39bfb41338ce 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
@@ -255,6 +255,10 @@ public class GooglePubsubConsumer extends DefaultConsumer
implements ShutdownAwa
// Only add to list after successful startup
subscribers.add(subscriber);
subscriberAdded = true;
+ // a stop while the subscriber was starting missed it;
stop it here or awaitTerminated never returns
+ if (!isRunAllowed() || isSuspendingOrSuspended()) {
+ subscriber.stopAsync();
+ }
subscriber.awaitTerminated();
} catch (Exception e) {
// Remove from list if it was added
@@ -321,6 +325,10 @@ 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()) {
+ synchronousPullResponseFuture.cancel(true);
+ }
PullResponse pullResponse =
synchronousPullResponseFuture.get();
for (ReceivedMessage message :
pullResponse.getReceivedMessagesList()) {
PubsubMessage pubsubMessage = message.getMessage();
diff --git
a/components/camel-google/camel-google-pubsub/src/test/java/org/apache/camel/component/google/pubsub/GooglePubsubConsumerStopRaceTest.java
b/components/camel-google/camel-google-pubsub/src/test/java/org/apache/camel/component/google/pubsub/GooglePubsubConsumerStopRaceTest.java
new file mode 100644
index 000000000000..de9d8b6f50ab
--- /dev/null
+++
b/components/camel-google/camel-google-pubsub/src/test/java/org/apache/camel/component/google/pubsub/GooglePubsubConsumerStopRaceTest.java
@@ -0,0 +1,166 @@
+/*
+ * 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.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+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.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.junit.jupiter.api.Assertions.assertTrue;
+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.when;
+
+/**
+ * A subscriber or pull that is still starting when the consumer stops must
still be stopped or cancelled, otherwise the
+ * consumer thread parks forever.
+ */
+public class GooglePubsubConsumerStopRaceTest {
+
+ 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 ExecutorService consumerExecutor =
Executors.newSingleThreadExecutor();
+ private final ScheduledExecutorService taskExecutor =
Executors.newSingleThreadScheduledExecutor();
+
+ @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())).thenReturn(taskExecutor);
+
+ when(endpoint.getComponent()).thenReturn(component);
+ when(endpoint.createExecutor(any())).thenReturn(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);
+ }
+
+ @AfterEach
+ void tearDown() {
+ consumerExecutor.shutdownNow();
+ taskExecutor.shutdownNow();
+ }
+
+ @Test
+ void subscriberStillStartingWhenConsumerStopsIsStopped() throws Exception {
+ CountDownLatch enteredAwaitRunning = new CountDownLatch(1);
+ CountDownLatch startupGate = new CountDownLatch(1);
+ CountDownLatch stopAsyncCalled = new CountDownLatch(1);
+
+ Subscriber subscriber = mock();
+ when(subscriber.startAsync()).thenReturn(subscriber);
+ doAnswer(invocation -> {
+ enteredAwaitRunning.countDown();
+ startupGate.await();
+ return null;
+ }).when(subscriber).awaitRunning();
+ doAnswer(invocation -> {
+ stopAsyncCalled.countDown();
+ return subscriber;
+ }).when(subscriber).stopAsync();
+ doAnswer(invocation -> {
+ stopAsyncCalled.await(10, TimeUnit.SECONDS);
+ return null;
+ }).when(subscriber).awaitTerminated();
+ when(component.getSubscriber(anyString(), any(),
any())).thenReturn(subscriber);
+
+ GooglePubsubConsumer consumer = new GooglePubsubConsumer(endpoint,
processor);
+ consumer.start();
+ try {
+ assertTrue(enteredAwaitRunning.await(5, TimeUnit.SECONDS),
"subscriber never started");
+
+ // the subscriber is not yet in the consumer's list, so stopping
the consumer misses it
+ consumer.stop();
+ startupGate.countDown();
+
+ assertTrue(stopAsyncCalled.await(5, TimeUnit.SECONDS),
+ "subscriber that finished starting after the consumer
stopped was never stopped");
+ } finally {
+ startupGate.countDown();
+ consumer.stop();
+ }
+ }
+
+ @Test
+ void pullStillBeingIssuedWhenConsumerStopsIsCancelled() throws Exception {
+ CountDownLatch enteredPull = new CountDownLatch(1);
+ CountDownLatch pullGate = new CountDownLatch(1);
+ CountDownLatch cancelled = new CountDownLatch(1);
+ SettableApiFuture<PullResponse> pullResponseFuture =
SettableApiFuture.create();
+ pullResponseFuture.addListener(cancelled::countDown, Runnable::run);
+
+ when(endpoint.isSynchronousPull()).thenReturn(true);
+ SubscriberStub subscriberStub = mock();
+ @SuppressWarnings("unchecked")
+ UnaryCallable<PullRequest, PullResponse> pullCallable =
mock(UnaryCallable.class);
+ when(subscriberStub.pullCallable()).thenReturn(pullCallable);
+ doAnswer(invocation -> {
+ enteredPull.countDown();
+ pullGate.await();
+ return pullResponseFuture;
+ }).when(pullCallable).futureCall(any(PullRequest.class));
+ when(component.getSubscriberStub(any())).thenReturn(subscriberStub);
+
+ GooglePubsubConsumer consumer = new GooglePubsubConsumer(endpoint,
processor);
+ consumer.start();
+ try {
+ assertTrue(enteredPull.await(5, TimeUnit.SECONDS), "pull was never
issued");
+
+ // the pull future is not yet in the consumer's pending set, so
stopping the consumer misses it
+ consumer.stop();
+ pullGate.countDown();
+
+ assertTrue(cancelled.await(5, TimeUnit.SECONDS),
+ "pull issued while the consumer stopped was never
cancelled");
+ } finally {
+ pullGate.countDown();
+ pullResponseFuture.cancel(true);
+ consumer.stop();
+ }
+ }
+}
