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 aca4248fddbb CAMEL-24315: camel-google-pubsub - stop subscribers that
finish starting after the consumer stopped
aca4248fddbb is described below
commit aca4248fddbb622abb8b761a14c769e342f84b31
Author: henrik242 <[email protected]>
AuthorDate: Mon Aug 3 07:48:19 2026 +0200
CAMEL-24315: camel-google-pubsub - stop subscribers that finish starting
after the consumer stopped
Fixes a race condition where a subscriber or synchronous pull that is
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 #25248
---
.../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 40a4f74c8998..ac94604b3cec 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
@@ -260,6 +260,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
@@ -328,6 +332,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();
+ }
+ }
+}
