CAMEL-26286: camel-disruptor - removing one of the routes that consume from a disruptor endpoint no longer shuts down its disruptor

components/camel-disruptor/src/main/java/org/apache/camel/component/disruptor/DisruptorConsumer.java

@@ -92,7 +92,7 @@ protected void doStop() throws Exception {
 
     @Override
     protected void doSuspend() throws Exception {
-        getEndpoint().onStopped(this);
+        getEndpoint().onSuspended(this);
     }
 
     @Override

components/camel-disruptor/src/main/java/org/apache/camel/component/disruptor/DisruptorEndpoint.java

@@ -76,6 +76,8 @@ public class DisruptorEndpoint extends DefaultEndpoint implements AsyncEndpoint,
 
     private final Set<DisruptorProducer> producers = new CopyOnWriteArraySet<>();
     private final Set<DisruptorConsumer> consumers = new CopyOnWriteArraySet<>();
+    // the consumers of suspended routes: they are not in consumers, but they still use this endpoint
+    private final Set<DisruptorConsumer> suspendedConsumers = new CopyOnWriteArraySet<>();
     private final DisruptorReference disruptorReference;
 
     @UriPath(description = "Name of queue")
@@ -279,6 +281,41 @@ protected void doInit() throws Exception {
         disruptorReference.addEndpoint(this);
     }
 
+    @Override
+    public void stop() {
+        // the routes that consume from the same uri share this endpoint, and removing one of them stops and shuts down
+        // the endpoint of its consumer: that must not shut down the disruptor of the other consumers, also not of the
+        // consumers of suspended routes, which are resumed on this endpoint. The check runs under the lock that a
+        // consumer takes to register (onStarted), so a consumer starting concurrently is either seen here or registers
+        // after the endpoint stopped
+        lock.lock();
+        try {
+            if (consumers.isEmpty() && suspendedConsumers.isEmpty()) {
+                super.stop();
+            } else {
+                LOGGER.debug("Not stopping endpoint {} as it still has {} consumer(s) and {} suspended consumer(s)",
+                        getEndpointUri(), consumers.size(), suspendedConsumers.size());
+            }
+        } finally {
+            lock.unlock();
+        }
+    }
+
+    @Override
+    public void shutdown() {
+        lock.lock();
+        try {
+            if (consumers.isEmpty() && suspendedConsumers.isEmpty()) {
+                super.shutdown();
+            } else {
+                LOGGER.debug("Not shutting down endpoint {} as it still has {} consumer(s) and {} suspended consumer(s)",
+                        getEndpointUri(), consumers.size(), suspendedConsumers.size());
+            }
+        } finally {
+            lock.unlock();
+        }
+    }
+
     @Override
     protected void doStop() throws Exception {
         // notify reference we are shutting down this endpoint
@@ -304,6 +341,7 @@ public DisruptorComponent getComponent() {
     void onStarted(final DisruptorConsumer consumer) throws Exception {
         lock.lock();
         try {
+            suspendedConsumers.remove(consumer);
             // validate multiple consumers have been enabled is necessary
             if (!consumers.isEmpty() && !isMultipleConsumersSupported()) {
                 throw new IllegalStateException(
@@ -321,9 +359,20 @@ void onStarted(final DisruptorConsumer consumer) throws Exception {
         }
     }
 
+    void onSuspended(final DisruptorConsumer consumer) throws Exception {
+        lock.lock();
+        try {
+            onStopped(consumer);
+            suspendedConsumers.add(consumer);
+        } finally {
+            lock.unlock();
+        }
+    }
+
     void onStopped(final DisruptorConsumer consumer) throws Exception {
         lock.lock();
         try {
+            suspendedConsumers.remove(consumer);
             if (consumers.remove(consumer)) {
                 LOGGER.debug("Stopping consumer {} on endpoint {}", consumer, getEndpointUri());
                 getDisruptor().reconfigure();

components/camel-disruptor/src/test/java/org/apache/camel/component/disruptor/DisruptorRemoveRouteOfSharedEndpointTest.java

@@ -0,0 +1,152 @@
+/*
+ * 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.disruptor;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.ReentrantLock;
+
+import org.apache.camel.Consumer;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.Test;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Removing one of the routes that consume from a disruptor endpoint must not shut down the disruptor of the other
+ * routes that consume from it, also when they are suspended.
+ */
+class DisruptorRemoveRouteOfSharedEndpointTest extends CamelTestSupport {
+
+    @Test
+    void removingOneOfSeveralConsumersKeepsTheOthersConsuming() throws Exception {
+        getMockEndpoint("mock:a").expectedBodiesReceived("before", "after");
+        getMockEndpoint("mock:b").expectedBodiesReceived("before");
+        getMockEndpoint("mock:c").expectedBodiesReceived("before", "after");
+
+        template.sendBody("disruptor:foo?multipleConsumers=true", "before");
+        getMockEndpoint("mock:b").assertIsSatisfied();
+
+        context.getRouteController().stopRoute("b");
+        context.removeRoute("b");
+
+        template.sendBody("disruptor:foo?multipleConsumers=true", "after");
+
+        MockEndpoint.assertIsSatisfied(context);
+    }
+
+    @Test
+    void removingAConsumerRouteKeepsTheSuspendedOnes() throws Exception {
+        getMockEndpoint("mock:a").expectedBodiesReceived("after");
+        getMockEndpoint("mock:c").expectedBodiesReceived("after");
+
+        context.getRouteController().suspendRoute("a");
+        context.getRouteController().suspendRoute("c");
+        context.getRouteController().stopRoute("b");
+        context.removeRoute("b");
+        context.getRouteController().resumeRoute("a");
+        context.getRouteController().resumeRoute("c");
+
+        template.sendBody("disruptor:foo?multipleConsumers=true", "after");
+
+        MockEndpoint.assertIsSatisfied(context);
+    }
+
+    @Test
+    void consumerRouteRemovedAndAddedAgainWhileAProducerRouteUsesTheQueue() throws Exception {
+        getMockEndpoint("mock:bar").expectedBodiesReceived("before", "after");
+
+        template.sendBody("direct:start", "before");
+
+        // such as a reload of the consumer route
+        context.getRouteController().stopRoute("barConsumer");
+        context.removeRoute("barConsumer");
+        context.addRoutes(new RouteBuilder() {
+            @Override
+            public void configure() {
+                from("disruptor:bar").routeId("barConsumer").to("mock:bar");
+            }
+        });
+
+        template.sendBody("direct:start", "after");
+
+        MockEndpoint.assertIsSatisfied(context);
+    }
+
+    @Test
+    void consumerRegisteringWhileTheEndpointIsAskedToStopKeepsTheEndpointStarted() throws Exception {
+        DisruptorEndpoint shared = context.getEndpoint("disruptor:race", DisruptorEndpoint.class);
+        LockExposingEndpoint endpoint = new LockExposingEndpoint(shared);
+        endpoint.start();
+        Consumer consumer = endpoint.createConsumer(exchange -> {
+        });
+        ReentrantLock lock = endpoint.lock();
+        Thread stopper = new Thread(endpoint::stop, "stopper");
+
+        // the test thread holds the lock that a consumer holds while it registers on the endpoint (onStarted), and
+        // the stop asked meanwhile (such as removing the last other route) must wait for it before checking for
+        // consumers
+        lock.lock();
+        try {
+            stopper.start();
+            await().atMost(10, TimeUnit.SECONDS).until(() -> lock.hasQueuedThread(stopper));
+            consumer.start();
+        } finally {
+            lock.unlock();
+        }
+        stopper.join(TimeUnit.SECONDS.toMillis(10));
+
+        assertFalse(stopper.isAlive());
+        assertTrue(endpoint.isStarted(), "the endpoint has a consumer again, so the stop must leave it started");
+
+        consumer.stop();
+        endpoint.stop();
+        assertTrue(endpoint.isStopped());
+    }
+
+    /**
+     * Shares the disruptor of an endpoint and exposes the service lock, so a test can hold it.
+     */
+    private static final class LockExposingEndpoint extends DisruptorEndpoint {
+
+        LockExposingEndpoint(DisruptorEndpoint shared) {
+            super(shared.getEndpointUri(), shared.getComponent(), shared.getDisruptor());
+        }
+
+        ReentrantLock lock() {
+            return (ReentrantLock) getInternalLock();
+        }
+    }
+
+    @Override
+    protected RouteBuilder createRouteBuilder() {
+        return new RouteBuilder() {
+            @Override
+            public void configure() {
+                from("disruptor:foo?multipleConsumers=true").routeId("a").to("mock:a");
+                from("disruptor:foo?multipleConsumers=true").routeId("b").to("mock:b");
+                from("disruptor:foo?multipleConsumers=true").routeId("c").to("mock:c");
+
+                from("direct:start").routeId("barProducer").to("disruptor:bar");
+                from("disruptor:bar").routeId("barConsumer").to("mock:bar");
+            }
+        };
+    }
+}