diff --git a/components/camel-disruptor/src/main/java/org/apache/camel/component/disruptor/DisruptorConsumer.java b/components/camel-disruptor/src/main/java/org/apache/camel/component/disruptor/DisruptorConsumer.java index adf7823fa77d0..f20145996c078 100644 --- a/components/camel-disruptor/src/main/java/org/apache/camel/component/disruptor/DisruptorConsumer.java +++ b/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 diff --git a/components/camel-disruptor/src/main/java/org/apache/camel/component/disruptor/DisruptorEndpoint.java b/components/camel-disruptor/src/main/java/org/apache/camel/component/disruptor/DisruptorEndpoint.java index 511a49a3c79bb..858733c08fea4 100644 --- a/components/camel-disruptor/src/main/java/org/apache/camel/component/disruptor/DisruptorEndpoint.java +++ b/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 producers = new CopyOnWriteArraySet<>(); private final Set consumers = new CopyOnWriteArraySet<>(); + // the consumers of suspended routes: they are not in consumers, but they still use this endpoint + private final Set 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(); diff --git a/components/camel-disruptor/src/test/java/org/apache/camel/component/disruptor/DisruptorRemoveRouteOfSharedEndpointTest.java b/components/camel-disruptor/src/test/java/org/apache/camel/component/disruptor/DisruptorRemoveRouteOfSharedEndpointTest.java new file mode 100644 index 0000000000000..777adde5b395f --- /dev/null +++ b/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"); + } + }; + } +}