(camel) 01/01: CAMEL-24315: camel-google-pubsub - stop subscribers that finish starting after the consumer stopped

[da...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

davsclaus pushed a commit to branch backport/25248-to-camel-4.18.x
in repository https://gitbox.apache.org/repos/asf/camel.git

commit 33806b1181d9c3c26429874bd087a3cdd8d59490
Author: henrik242 <he...@users.noreply.github.com>
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 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();
+        }
+    }
+}