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 867b5d053e96 CAMEL-26880: camel-cluster - ClusteredRoutePolicy applies
the leadership again when the CamelContext resumes (#29784)
867b5d053e96 is described below
commit 867b5d053e968f188205b6dd210363fdc3f796f0
Author: allthingssecurity <[email protected]>
AuthorDate: Fri Nov 13 15:02:02 2026 +0530
CAMEL-26880: camel-cluster - ClusteredRoutePolicy applies the leadership
again when the CamelContext resumes (#29784)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../camel/impl/cluster/ClusteredRoutePolicy.java | 32 +++-
.../camel/cluster/ClusteredEdgeCasesTest.java | 175 +++++++++++++++++++++
2 files changed, 203 insertions(+), 4 deletions(-)
diff --git
a/core/camel-cluster/src/main/java/org/apache/camel/impl/cluster/ClusteredRoutePolicy.java
b/core/camel-cluster/src/main/java/org/apache/camel/impl/cluster/ClusteredRoutePolicy.java
index 1c0347d5ebd7..862bc7e607e4 100644
---
a/core/camel-cluster/src/main/java/org/apache/camel/impl/cluster/ClusteredRoutePolicy.java
+++
b/core/camel-cluster/src/main/java/org/apache/camel/impl/cluster/ClusteredRoutePolicy.java
@@ -43,6 +43,7 @@ import org.apache.camel.cluster.CamelClusterMember;
import org.apache.camel.cluster.CamelClusterService;
import org.apache.camel.cluster.CamelClusterView;
import org.apache.camel.spi.CamelEvent;
+import org.apache.camel.spi.CamelEvent.CamelContextResumedEvent;
import org.apache.camel.spi.CamelEvent.CamelContextStartedEvent;
import org.apache.camel.spi.CamelEvent.RouteAddedEvent;
import org.apache.camel.spi.ThreadPoolProfile;
@@ -83,12 +84,14 @@ public final class ClusteredRoutePolicy extends
RoutePolicySupport implements Ca
private final AtomicReference<CamelClusterView> clusterView;
private final AtomicBoolean leadershipChangePending;
// the leadership is applied again to the routes (and not only on a
change): a route of this policy was added to
- // the CamelContext, or a route failed to start or stop
+ // the CamelContext, a route failed to start or stop, or the CamelContext
resumed
private final AtomicBoolean reapply;
private final AtomicBoolean retryScheduled;
private CamelClusterService clusterService;
private volatile boolean startManagedRoutesEarly;
private volatile boolean initialDelayElapsed;
+ // the CamelContext was resumed once (a policy attached after the start
does not get the started event)
+ private volatile boolean contextResumed;
private Duration initialDelay;
private volatile ExecutorService leadershipExecutor;
@@ -413,10 +416,23 @@ public final class ClusteredRoutePolicy extends
RoutePolicySupport implements Ca
private void doStartManagedRoutes() {
// if we are currently starting up Camel context then defer starting
routes till its fully started
if (camelContext.isStarting()) {
+ if (contextStarted.get() || contextResumed) {
+ // the CamelContext is resuming (its status is starting until
the resume completed, and the resumed
+ // event comes before that): the started event does not come
again, so try again later
+ LOG.debug("Will start managed routes once camel context has
resumed");
+ scheduleRetry();
+ return;
+ }
LOG.debug("Will defer starting managed routes until camel context
is fully started");
startManagedRoutesEarly = true;
return;
}
+ // a route cannot be started while the CamelContext is suspended
(startRoute does nothing then): the leadership is
+ // applied again when the CamelContext resumes
(CamelContextResumedEvent)
+ if (camelContext.isSuspending() || camelContext.isSuspended()) {
+ LOG.debug("Will defer starting managed routes until camel context
is resumed");
+ return;
+ }
// and if the initial delay has not elapsed yet, then defer starting
routes till it has
if (!initialDelayElapsed && initialDelay != null &&
initialDelay.toMillis() > 0) {
LOG.debug("Will defer starting managed routes until the initial
delay {} has elapsed", initialDelay);
@@ -656,7 +672,14 @@ public final class ClusteredRoutePolicy extends
RoutePolicySupport implements Ca
implements ExtendedStartupListener, NonManagedService {
@Override
public void notify(CamelEvent event) throws Exception {
- if (event instanceof RouteAddedEvent) {
+ if (event instanceof CamelContextResumedEvent) {
+ contextResumed = true;
+ // a leadership taken while the CamelContext was suspended
could not start the routes, and the resume
+ // starts (or resumes) the routes that were running when it
was suspended, also on a node that lost the
+ // leadership meanwhile: the policy thread applies the
leadership again
+ reapply.set(true);
+ handOverLeadershipChange();
+ } else if (event instanceof RouteAddedEvent) {
// on the thread that adds the route, which holds the
CamelContext route lock: the policy thread starts
// the route if this node is the leader (the route is known to
the CamelContext from now on)
reapply.set(true);
@@ -668,9 +691,10 @@ public final class ClusteredRoutePolicy extends
RoutePolicySupport implements Ca
@Override
public boolean isEnabled(CamelEvent event) {
- // a route added while the CamelContext starts is started (or
deferred) with the others. The event also comes
+ // the resumed event: see notify. A route added while the
CamelContext starts is started (or deferred) with
+ // the others. The added event also comes
// when a stopped route is warmed up again; the extra pass then
has nothing to do
- return event instanceof CamelContextStartedEvent
+ return event instanceof CamelContextStartedEvent || event
instanceof CamelContextResumedEvent
|| event instanceof RouteAddedEvent rae &&
camelContext.isStarted()
&& stoppedRoutes.contains(rae.getRoute());
}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/cluster/ClusteredEdgeCasesTest.java
b/core/camel-core/src/test/java/org/apache/camel/cluster/ClusteredEdgeCasesTest.java
index 6af19850ad56..54935235cc4a 100644
---
a/core/camel-core/src/test/java/org/apache/camel/cluster/ClusteredEdgeCasesTest.java
+++
b/core/camel-core/src/test/java/org/apache/camel/cluster/ClusteredEdgeCasesTest.java
@@ -37,9 +37,11 @@ import org.apache.camel.impl.DefaultCamelContext;
import org.apache.camel.impl.cluster.ClusteredRouteController;
import org.apache.camel.impl.cluster.ClusteredRoutePolicy;
import org.apache.camel.impl.engine.DefaultRouteController;
+import org.apache.camel.spi.CamelEvent;
import org.apache.camel.support.DefaultComponent;
import org.apache.camel.support.DefaultConsumer;
import org.apache.camel.support.DefaultEndpoint;
+import org.apache.camel.support.EventNotifierSupport;
import org.apache.camel.support.RoutePolicySupport;
import org.apache.camel.support.cluster.AbstractCamelClusterService;
import org.apache.camel.support.cluster.AbstractCamelClusterView;
@@ -446,6 +448,179 @@ public class ClusteredEdgeCasesTest {
}
}
+ @Test
+ public void testLeadershipTakenWhileTheCamelContextIsSuspended() throws
Exception {
+ TestClusterService cs = new TestClusterService("cs");
+ CamelContext context = new DefaultCamelContext();
+ context.addService(cs);
+ ClusteredRoutePolicy policy =
ClusteredRoutePolicy.forNamespace("my-ns");
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("seda:foo").routeId("foo").routePolicy(policy).to("mock:foo");
+ }
+ });
+ context.start();
+ try {
+ context.suspend();
+ // the node becomes the leader while the CamelContext is suspended
(from JMX, for example)
+ cs.getView().setLeader(true);
+ await().atMost(10, TimeUnit.SECONDS).until(policy::isLeader);
+
+ context.resume();
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(
+ () -> assertEquals(ServiceStatus.Started,
context.getRouteController().getRouteStatus("foo")));
+ } finally {
+ context.stop();
+ }
+ }
+
+ @Test
+ public void
testLeadershipTakenWhileTheCamelContextIsSuspendedAndASlowResume() throws
Exception {
+ TestClusterService cs = new TestClusterService("cs");
+ CamelContext context = new DefaultCamelContext();
+ context.addService(cs);
+ ClusteredRoutePolicy policy =
ClusteredRoutePolicy.forNamespace("my-ns");
+ policy.setRetryDelay(Duration.ofMillis(100));
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("seda:foo").routeId("foo").routePolicy(policy).to("mock:foo");
+ }
+ });
+ context.start();
+ // an event notifier after the one of the policy holds the resume, so
the CamelContext is still starting when
+ // the policy thread applies the leadership after the resumed event
+ CountDownLatch neverReleased = new CountDownLatch(1);
+ context.getManagementStrategy().addEventNotifier(new
EventNotifierSupport() {
+ @Override
+ public void notify(CamelEvent event) throws Exception {
+ neverReleased.await(500, TimeUnit.MILLISECONDS);
+ }
+
+ @Override
+ public boolean isEnabled(CamelEvent event) {
+ return event instanceof CamelEvent.CamelContextResumedEvent;
+ }
+ });
+ try {
+ context.suspend();
+ cs.getView().setLeader(true);
+ await().atMost(10, TimeUnit.SECONDS).until(policy::isLeader);
+
+ context.resume();
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(
+ () -> assertEquals(ServiceStatus.Started,
context.getRouteController().getRouteStatus("foo")));
+ } finally {
+ context.stop();
+ }
+ }
+
+ @Test
+ public void testRouteAddedAfterStartAndASlowResume() throws Exception {
+ TestClusterService cs = new TestClusterService("cs");
+ CamelContext context = new DefaultCamelContext();
+ context.addService(cs);
+ ClusteredRoutePolicy policy =
ClusteredRoutePolicy.forNamespace("my-ns");
+ policy.setRetryDelay(Duration.ofMillis(100));
+ context.start();
+ // the policy is attached after the CamelContext started, so it never
gets the started event
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("seda:foo").routeId("foo").routePolicy(policy).to("mock:foo");
+ }
+ });
+ CountDownLatch neverReleased = new CountDownLatch(1);
+ context.getManagementStrategy().addEventNotifier(new
EventNotifierSupport() {
+ @Override
+ public void notify(CamelEvent event) throws Exception {
+ neverReleased.await(500, TimeUnit.MILLISECONDS);
+ }
+
+ @Override
+ public boolean isEnabled(CamelEvent event) {
+ return event instanceof CamelEvent.CamelContextResumedEvent;
+ }
+ });
+ try {
+ context.suspend();
+ cs.getView().setLeader(true);
+ await().atMost(10, TimeUnit.SECONDS).until(policy::isLeader);
+
+ context.resume();
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(
+ () -> assertEquals(ServiceStatus.Started,
context.getRouteController().getRouteStatus("foo")));
+ } finally {
+ context.stop();
+ }
+ }
+
+ @Test
+ public void
testLeadershipLostWhileTheCamelContextIsSuspendedWithASuspendableConsumer()
throws Exception {
+ TestClusterService cs = new TestClusterService("cs");
+ CamelContext context = new DefaultCamelContext();
+ context.addService(cs);
+ ClusteredRoutePolicy policy =
ClusteredRoutePolicy.forNamespace("my-ns");
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("seda:foo").routeId("foo").routePolicy(policy).to("mock:foo");
+ }
+ });
+ context.start();
+ try {
+ cs.getView().setLeader(true);
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(
+ () -> assertEquals(ServiceStatus.Started,
context.getRouteController().getRouteStatus("foo")));
+
+ context.suspend();
+ cs.getView().setLeader(false);
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
!policy.isLeader());
+
+ context.resume();
+ await().during(1, TimeUnit.SECONDS).atMost(10,
TimeUnit.SECONDS).untilAsserted(
+ () -> assertEquals(ServiceStatus.Stopped,
context.getRouteController().getRouteStatus("foo")));
+ } finally {
+ context.stop();
+ }
+ }
+
+ @Test
+ public void testLeadershipLostWhileTheCamelContextIsSuspended() throws
Exception {
+ TestClusterService cs = new TestClusterService("cs");
+ CamelContext context = new DefaultCamelContext();
+ context.addService(cs);
+ // a consumer that does not support suspension (the failing component
with no failure): the suspend of the
+ // CamelContext stops its route, and the resume starts it again
+ context.addComponent("failing", new FailingStartComponent(new
AtomicInteger()));
+ ClusteredRoutePolicy policy =
ClusteredRoutePolicy.forNamespace("my-ns");
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("failing:foo").routeId("foo").routePolicy(policy).to("mock:foo");
+ }
+ });
+ context.start();
+ try {
+ cs.getView().setLeader(true);
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(
+ () -> assertEquals(ServiceStatus.Started,
context.getRouteController().getRouteStatus("foo")));
+
+ context.suspend();
+ // the node loses the leadership while the CamelContext is
suspended
+ cs.getView().setLeader(false);
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
!policy.isLeader());
+
+ // the resume starts the routes it stopped: this node must not
keep running the route
+ context.resume();
+ await().during(1, TimeUnit.SECONDS).atMost(10,
TimeUnit.SECONDS).untilAsserted(
+ () -> assertEquals(ServiceStatus.Stopped,
context.getRouteController().getRouteStatus("foo")));
+ } finally {
+ context.stop();
+ }
+ }
+
@Test
public void testRouteThatFailsToStopIsStartedAgainOnTheNextLeadership()
throws Exception {
TestClusterService cs = new TestClusterService("cs");
