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 3e9a631827b7 CAMEL-26800: camel-master - MasterConsumer stops its
delegate before it gives up the leadership (#29581)
3e9a631827b7 is described below
commit 3e9a631827b728570c70e84ccd388bb09dcf4005
Author: allthingssecurity <[email protected]>
AuthorDate: Wed Nov 11 12:21:19 2026 +0530
CAMEL-26800: camel-master - MasterConsumer stops its delegate before it
gives up the leadership (#29581)
* CAMEL-26800: camel-master - MasterConsumer stops its delegate before it
gives up the leadership
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../camel/component/master/MasterConsumer.java | 38 +++++++-----
.../master/MasterConsumerLeadershipTest.java | 72 ++++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 10 +++
3 files changed, 106 insertions(+), 14 deletions(-)
diff --git
a/components/camel-master/src/main/java/org/apache/camel/component/master/MasterConsumer.java
b/components/camel-master/src/main/java/org/apache/camel/component/master/MasterConsumer.java
index f9d9f79685f2..f6250afe46fd 100644
---
a/components/camel-master/src/main/java/org/apache/camel/component/master/MasterConsumer.java
+++
b/components/camel-master/src/main/java/org/apache/camel/component/master/MasterConsumer.java
@@ -114,8 +114,6 @@ public class MasterConsumer extends DefaultConsumer
implements ResumeAware<Resum
@Override
protected void doStop() throws Exception {
- super.doStop();
-
leadershipLock.lock();
try {
// a start can still be pending, cancel it first so it cannot
start the delegated consumer
@@ -126,19 +124,31 @@ public class MasterConsumer extends DefaultConsumer
implements ResumeAware<Resum
leadershipLock.unlock();
}
- // note: removeEventListener below needs the write lock of the cluster
view, while an event dispatch
- // takes the read lock of the view and then leadershipLock. This
thread must not hold leadershipLock
- // here, or the two orders deadlock
-
- if (view != null) {
- view.removeEventListener(leadershipListener);
- clusterService.releaseView(view);
- view = null;
+ try {
+ // the delegated consumer is stopped while this node still holds
the leadership, and before the processor
+ // of the route, so no other node starts consuming while it stops,
and the messages it still routes while
+ // it stops (a consumer may wait in its stop for the messages it
processes) reach a started route. A
+ // leadership event that comes meanwhile is ignored, as this
consumer is stopping
+ Consumer delegate = delegatedConsumer;
+ delegatedConsumer = null;
+ ServiceHelper.stopAndShutdownServices(delegate, delegatedEndpoint);
+ // the start task was cancelled above
+
getEndpoint().getCamelContext().getExecutorServiceManager().shutdown(leaderPool);
+ } finally {
+ try {
+ super.doStop();
+ } finally {
+ // note: removeEventListener below needs the write lock of the
cluster view, while an event dispatch
+ // takes the read lock of the view and then leadershipLock.
This thread must not hold leadershipLock
+ // here, or the two orders deadlock
+ CamelClusterView current = view;
+ view = null;
+ if (current != null) {
+ current.removeEventListener(leadershipListener);
+ clusterService.releaseView(current);
+ }
+ }
}
-
-
getEndpoint().getCamelContext().getExecutorServiceManager().shutdown(leaderPool);
- ServiceHelper.stopAndShutdownServices(delegatedConsumer,
delegatedEndpoint);
- delegatedConsumer = null;
}
@Override
diff --git
a/components/camel-master/src/test/java/org/apache/camel/component/master/MasterConsumerLeadershipTest.java
b/components/camel-master/src/test/java/org/apache/camel/component/master/MasterConsumerLeadershipTest.java
index f39510bfbf14..a42388473ed5 100644
---
a/components/camel-master/src/test/java/org/apache/camel/component/master/MasterConsumerLeadershipTest.java
+++
b/components/camel-master/src/test/java/org/apache/camel/component/master/MasterConsumerLeadershipTest.java
@@ -28,10 +28,13 @@ import java.util.concurrent.atomic.AtomicReference;
import org.apache.camel.Consumer;
import org.apache.camel.Endpoint;
+import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.Producer;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.cluster.CamelClusterMember;
+import org.apache.camel.cluster.CamelClusterView;
+import org.apache.camel.component.mock.MockEndpoint;
import org.apache.camel.impl.DefaultCamelContext;
import org.apache.camel.support.DefaultComponent;
import org.apache.camel.support.DefaultConsumer;
@@ -48,6 +51,7 @@ import org.junit.jupiter.api.Timeout;
import static org.awaitility.Awaitility.await;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
/**
@@ -290,6 +294,54 @@ public class MasterConsumerLeadershipTest {
return registry.getTasks().stream().anyMatch(task ->
"Leadership".equals(task.getName()));
}
+ @Test
+ @Timeout(60)
+ void testDelegateIsStoppedBeforeTheLeadershipIsGivenUp() throws Exception {
+ TestClusterView view = clusterService.getTestView();
+ view.setLeader(true);
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(() ->
assertEquals(1, probe.started.get()));
+
+ // while the delegated consumer stops, this node must still hold the
view, and a message it still routes (as a
+ // consumer that completes its messages in its stop does) must reach a
started route
+ MockEndpoint mock = context.getEndpoint("mock:result",
MockEndpoint.class);
+ mock.expectedBodiesReceived("in flight");
+ AtomicBoolean viewHeldWhileStopping = new AtomicBoolean();
+ AtomicReference<Exception> routingFailure = new AtomicReference<>();
+ probe.stopHook.set(consumer -> {
+ viewHeldWhileStopping.set(view.isRunAllowed());
+ Exchange exchange = consumer.createExchange(false);
+ exchange.getMessage().setBody("in flight");
+ try {
+ consumer.getProcessor().process(exchange);
+ } catch (Exception e) {
+ exchange.setException(e);
+ }
+ routingFailure.set(exchange.getException());
+ });
+ context.getRouteController().stopRoute("master-route");
+
+ assertEquals(1, probe.stopped.get());
+ assertTrue(viewHeldWhileStopping.get(), "The leadership was given up
before the delegated consumer stopped");
+ assertNull(routingFailure.get(), "A message routed while the delegated
consumer stops failed");
+ mock.assertIsSatisfied();
+ assertFalse(view.isRunAllowed(), "The view is not released after the
consumer stopped");
+ }
+
+ @Test
+ @Timeout(60)
+ void testDelegateIsStoppedWhenReleasingTheViewFails() throws Exception {
+ TestClusterView view = clusterService.getTestView();
+ view.setLeader(true);
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(() ->
assertEquals(1, probe.started.get()));
+
+ // the failure is logged by the shutdown strategy
+ clusterService.failRelease = true;
+ context.getRouteController().stopRoute("master-route");
+
+ assertEquals(1, probe.stopped.get(), "The delegated consumer was not
stopped");
+ assertFalse(view.isRunAllowed(), "The view is not released");
+ }
+
// ************************************
// Delegated endpoint under observation
// ************************************
@@ -303,6 +355,8 @@ public class MasterConsumerLeadershipTest {
private final AtomicReference<CountDownLatch> startGate = new
AtomicReference<>();
private final AtomicReference<CountDownLatch> suspendEntered = new
AtomicReference<>();
private final AtomicReference<CountDownLatch> suspendGate = new
AtomicReference<>();
+ // runs when a delegated consumer starts to stop
+ private final AtomicReference<StopHook> stopHook = new
AtomicReference<>();
@Override
protected Endpoint createEndpoint(String uri, String remaining,
Map<String, Object> parameters) {
@@ -319,6 +373,11 @@ public class MasterConsumerLeadershipTest {
}
}
+ @FunctionalInterface
+ private interface StopHook {
+ void stopping(ProbeConsumer consumer);
+ }
+
private static final class ProbeEndpoint extends DefaultEndpoint {
private final ProbeComponent component;
@@ -369,6 +428,10 @@ public class MasterConsumerLeadershipTest {
@Override
protected void doStop() throws Exception {
+ StopHook hook = component.stopHook.get();
+ if (hook != null) {
+ hook.stopping(this);
+ }
super.doStop();
component.stopped.incrementAndGet();
}
@@ -393,6 +456,7 @@ public class MasterConsumerLeadershipTest {
private static final class TestClusterService extends
AbstractCamelClusterService<TestClusterView> {
private volatile TestClusterView view;
+ private volatile boolean failRelease;
TestClusterService() {
super("test-cluster-service");
@@ -407,6 +471,14 @@ public class MasterConsumerLeadershipTest {
view = new TestClusterView(this, namespace);
return view;
}
+
+ @Override
+ public void releaseView(CamelClusterView view) throws Exception {
+ super.releaseView(view);
+ if (failRelease) {
+ throw new IllegalStateException("Simulated failure to release
the view");
+ }
+ }
}
private static final class TestClusterView extends
AbstractCamelClusterView {
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index bb0dc8cbdd84..42c70d790bfb 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -193,6 +193,16 @@ Apache MINA SSHD leaves it off by default, so a small
request, such as one that
listing, could wait for the server to acknowledge the previous reply. The
`org.apache.sshd.config.tcp-nodelay` system
property no longer applies to the connections of `mina-sftp`.
+=== camel-master - the consumer keeps the leadership until its delegated
consumer has stopped
+
+When a `master:` consumer stops (its route stops, or the CamelContext shuts
down), it now stops the delegated consumer
+first, then the processor of the route, and only then releases the cluster
view, so the leadership moves to another node
+once this node no longer consumes. Before, it released the view first, so
another node could start consuming while this
+one was still stopping its consumer (a consumer may wait in its stop for the
messages it is processing, as JMS does, and
+these messages now reach a started route). The hand-over on a graceful stop
can therefore come later, by up to the time
+the delegated consumer takes to stop (a stop that blocks holds back the
failover until it returns). Exchanges in flight
+that the delegated consumer does not wait for in its stop may still be in
flight when the leadership moves, as before.
+
=== camel-ftp - sftp checks whether a file exists with one lstat request
The `sftp` component now checks whether a remote file exists
(`fileExist=Ignore|Fail|Move`, `tempFileName`,
