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 d9a95c67223a CAMEL-25849: camel-disruptor - a disruptor-vm consumer
keeps consuming when the CamelContext that first used the queue stops (#28322)
d9a95c67223a is described below
commit d9a95c67223a2fdffb207dc3809bc5864d296ce0
Author: allthingssecurity <[email protected]>
AuthorDate: Fri Oct 23 13:52:18 2026 +0530
CAMEL-25849: camel-disruptor - a disruptor-vm consumer keeps consuming when
the CamelContext that first used the queue stops (#28322)
The consumer threads of a disruptor-vm queue belonged to the CamelContext
that first created the queue, so stopping that CamelContext stopped consumption
for all other CamelContexts. The threads now belong to a CamelContext that uses
the queue, and move to another one when the last endpoint of their CamelContext
is removed.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../component/disruptor/DisruptorReference.java | 56 +++++++++++++++++--
.../vm/DisruptorVmFirstContextStopTest.java | 65 ++++++++++++++++++++++
2 files changed, 115 insertions(+), 6 deletions(-)
diff --git
a/components/camel-disruptor/src/main/java/org/apache/camel/component/disruptor/DisruptorReference.java
b/components/camel-disruptor/src/main/java/org/apache/camel/component/disruptor/DisruptorReference.java
index 4317c87a047f..67427a5fc416 100644
---
a/components/camel-disruptor/src/main/java/org/apache/camel/component/disruptor/DisruptorReference.java
+++
b/components/camel-disruptor/src/main/java/org/apache/camel/component/disruptor/DisruptorReference.java
@@ -39,7 +39,9 @@ import java.util.concurrent.locks.ReentrantLock;
import com.lmax.disruptor.InsufficientCapacityException;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
+import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
+import org.apache.camel.RuntimeCamelException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -79,6 +81,8 @@ public class DisruptorReference {
//access guarded by this
private ExecutorService executor;
+ // the CamelContext whose ExecutorServiceManager created the executor,
which shuts it down when it stops
+ private CamelContext executorContext;
private LifecycleAwareExchangeEventHandler[] handlers = new
LifecycleAwareExchangeEventHandler[0];
@@ -256,17 +260,42 @@ public class DisruptorReference {
}
}
+ /**
+ * The CamelContext to create the threads of the event handlers with: the
CamelContext of an endpoint that uses this
+ * Disruptor (with disruptor-vm, the endpoints can belong to several
CamelContexts), else that of the component that
+ * created it.
+ */
+ private CamelContext resolveExecutorContext() {
+ for (DisruptorEndpoint endpoint : endpoints) {
+ CamelContext context = endpoint.getCamelContext();
+ if (context != null) {
+ return context;
+ }
+ }
+ return component.getCamelContext();
+ }
+
+ private boolean isExecutorContextInUse() {
+ for (DisruptorEndpoint endpoint : endpoints) {
+ if (endpoint.getCamelContext() == executorContext) {
+ return true;
+ }
+ }
+ return false;
+ }
+
private void resizeThreadPoolExecutor(final int newSize) {
if (executor == null && newSize > 0) {
LOGGER.debug("Creating new executor with {} threads", newSize);
//no thread pool executor yet, create a new one
- executor =
component.getCamelContext().getExecutorServiceManager().newFixedThreadPool(this,
uri,
- newSize);
+ executorContext = resolveExecutorContext();
+ executor =
executorContext.getExecutorServiceManager().newFixedThreadPool(this, uri,
newSize);
} else if (executor != null && newSize <= 0) {
LOGGER.debug("Shutting down executor");
//we need to shut down our executor
-
component.getCamelContext().getExecutorServiceManager().shutdown(executor);
+ executorContext.getExecutorServiceManager().shutdown(executor);
executor = null;
+ executorContext = null;
} else if (executor instanceof ThreadPoolExecutor threadPoolExecutor) {
LOGGER.debug("Resizing existing executor to {} threads", newSize);
//our thread pool executor is of type ThreadPoolExecutor, we know
how to resize it
@@ -283,10 +312,10 @@ public class DisruptorReference {
} else if (newSize > 0) {
LOGGER.debug("Shutting down old and creating new executor with {}
threads", newSize);
//hmmm...no idea what kind of executor this is...just kill it and
start fresh
-
component.getCamelContext().getExecutorServiceManager().shutdown(executor);
+ executorContext.getExecutorServiceManager().shutdown(executor);
- executor =
component.getCamelContext().getExecutorServiceManager().newFixedThreadPool(this,
uri,
- newSize);
+ executorContext = resolveExecutorContext();
+ executor =
executorContext.getExecutorServiceManager().newFixedThreadPool(this, uri,
newSize);
}
}
@@ -404,6 +433,21 @@ public class DisruptorReference {
}
endpoints.remove(disruptorEndpoint);
LOGGER.debug("Endpoint removed: {}, new total endpoints {}",
disruptorEndpoint, getEndpointCount());
+ if (executor != null && !endpoints.isEmpty() &&
!isExecutorContextInUse()) {
+ // a disruptor-vm queue shared with other CamelContexts: the
CamelContext of the threads of the event
+ // handlers no longer uses the queue, and shuts the threads
down when it stops, so move them to the
+ // CamelContext of an endpoint that still uses it, or its
consumers stop consuming for good
+ LOGGER.info("Moving the event handlers of disruptor {} from
CamelContext {}, which no longer uses it, to"
+ + " CamelContext {}",
+ uri, executorContext.getName(),
resolveExecutorContext().getName());
+ shutdownDisruptor(true);
+ shutdownExecutor();
+ try {
+ start();
+ } catch (Exception e) {
+ throw RuntimeCamelException.wrapRuntimeCamelException(e);
+ }
+ }
} finally {
lock.unlock();
}
diff --git
a/components/camel-disruptor/src/test/java/org/apache/camel/component/disruptor/vm/DisruptorVmFirstContextStopTest.java
b/components/camel-disruptor/src/test/java/org/apache/camel/component/disruptor/vm/DisruptorVmFirstContextStopTest.java
new file mode 100644
index 000000000000..28a460159a36
--- /dev/null
+++
b/components/camel-disruptor/src/test/java/org/apache/camel/component/disruptor/vm/DisruptorVmFirstContextStopTest.java
@@ -0,0 +1,65 @@
+/*
+ * 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.vm;
+
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.junit.jupiter.api.Test;
+
+/**
+ * The consumer of a disruptor-vm queue in one CamelContext keeps consuming
when the CamelContext that first used the
+ * queue is stopped.
+ */
+class DisruptorVmFirstContextStopTest extends AbstractVmTestSupport {
+
+ @Test
+ void consumerKeepsConsumingAfterTheFirstContextStops() throws Exception {
+ MockEndpoint mock = context2.getEndpoint("mock:result",
MockEndpoint.class);
+ mock.expectedBodiesReceived("first");
+ template.sendBody("direct:test", "first");
+ mock.assertIsSatisfied();
+ mock.reset();
+
+ // the CamelContext that created the queue goes away, the one with the
consumer goes on
+ context.stop();
+
+ mock.expectedBodiesReceived("second");
+ template2.sendBody("direct:other", "second");
+ mock.assertIsSatisfied();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:test").to("disruptor-vm:foo");
+ }
+ };
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilderForSecondContext() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("disruptor-vm:foo").to("mock:result");
+ from("direct:other").to("disruptor-vm:foo");
+ }
+ };
+ }
+}
