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 5a84225089f9 CAMEL-26799: camel-cluster - ClusteredRoutePolicy starts
a route when the leadership was taken before the route could be started (#29589)
5a84225089f9 is described below
commit 5a84225089f915316b7a0c64e8635d4f3083f1c8
Author: allthingssecurity <[email protected]>
AuthorDate: Wed Nov 11 12:21:40 2026 +0530
CAMEL-26799: camel-cluster - ClusteredRoutePolicy starts a route when the
leadership was taken before the route could be started (#29589)
* CAMEL-26799: camel-cluster - ClusteredRoutePolicy starts a route when the
leadership was taken before the route could be started
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../camel/impl/cluster/ClusteredRoutePolicy.java | 51 +++++++-
.../camel/cluster/ClusteredEdgeCasesTest.java | 134 +++++++++++++++++++++
2 files changed, 181 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 db96dd245bfd..46d9db812017 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,11 +43,13 @@ import org.apache.camel.cluster.CamelClusterService;
import org.apache.camel.cluster.CamelClusterView;
import org.apache.camel.spi.CamelEvent;
import org.apache.camel.spi.CamelEvent.CamelContextStartedEvent;
+import org.apache.camel.spi.CamelEvent.RouteAddedEvent;
import org.apache.camel.spi.ThreadPoolProfile;
import org.apache.camel.support.RoutePolicySupport;
import org.apache.camel.support.SimpleEventNotifierSupport;
import org.apache.camel.support.cluster.ClusterServiceHelper;
import org.apache.camel.support.cluster.ClusterServiceSelectors;
+import org.apache.camel.support.service.BaseService;
import org.apache.camel.util.ObjectHelper;
import org.apache.camel.util.ReferenceCount;
import org.apache.camel.util.concurrent.ThreadPoolRejectedPolicy;
@@ -77,6 +79,8 @@ public final class ClusteredRoutePolicy extends
RoutePolicySupport implements Ca
private final Lock retainLock;
private final AtomicReference<CamelClusterView> clusterView;
private final AtomicBoolean leadershipChangePending;
+ // a route of this policy was added to the CamelContext, so the leader may
now start it
+ private final AtomicBoolean routeAdded;
private CamelClusterService clusterService;
private volatile boolean startManagedRoutesEarly;
private volatile boolean initialDelayElapsed;
@@ -100,6 +104,7 @@ public final class ClusteredRoutePolicy extends
RoutePolicySupport implements Ca
this.lock = new ReentrantLock();
this.retainLock = new ReentrantLock();
this.leadershipChangePending = new AtomicBoolean();
+ this.routeAdded = new AtomicBoolean();
// the routes are started and stopped on the policy thread, while
routes are added and removed by the caller
this.stoppedRoutes = ConcurrentHashMap.newKeySet();
this.startedRoutes = ConcurrentHashMap.newKeySet();
@@ -337,6 +342,8 @@ public final class ClusteredRoutePolicy extends
RoutePolicySupport implements Ca
return;
}
+ // a route of this policy was added to the CamelContext since the
last change
+ boolean added = routeAdded.getAndSet(false);
boolean isLeader = view.getLocalMember().isLeader();
if (isLeader && leader.compareAndSet(false, isLeader)) {
LOG.debug("Leadership taken");
@@ -344,6 +351,10 @@ public final class ClusteredRoutePolicy extends
RoutePolicySupport implements Ca
} else if (!isLeader && leader.getAndSet(isLeader)) {
LOG.debug("Leadership lost");
stopManagedRoutes();
+ } else if (isLeader && added) {
+ // a route that could not be started when the leadership was
taken, as it was not added to the
+ // CamelContext yet (doStartManagedRoutes only starts the
routes that are still stopped)
+ startManagedRoutes();
}
} finally {
lock.unlock();
@@ -375,19 +386,40 @@ public final class ClusteredRoutePolicy extends
RoutePolicySupport implements Ca
}
if (!isRunAllowed()) {
+ if (isNew() || isBuild() || isInit()) {
+ // the leadership was taken before this policy started (a
cluster service started on its own, with a
+ // view that is already the leader): start the routes once the
CamelContext has started. A policy that
+ // is stopping or stopped does not start them
+ LOG.debug("Will defer starting managed routes until the policy
and the camel context are started");
+ startManagedRoutesEarly = true;
+ }
return;
}
try {
for (Route route : stoppedRoutes) {
- ServiceStatus status = getStatus(route);
+ ServiceStatus status =
camelContext.getRouteController().getRouteStatus(route.getId());
+ if (status == null) {
+ // the route is not added to the CamelContext yet:
starting it now would do nothing, so it is
+ // started when it is added (RouteAddedEvent)
+ LOG.debug("Will defer starting route '{}' until it is
added to the CamelContext", route.getId());
+ continue;
+ }
boolean autostart = autoStartupRoutes.contains(route);
- if (status != null && status.isStartable() && autostart) {
+ if (status.isStartable() && autostart) {
LOG.debug("Starting route '{}'", route.getId());
camelContext.getRouteController().startRoute(route.getId());
startedRoutes.add(route);
+ if (camelContext.getRoute(route.getId()) != route
+ || route instanceof BaseService service &&
service.isShutdown()) {
+ // the route was removed while it was started
(startRoute waited for the removal, and did
+ // nothing): a route added later with the same id must
not be stopped by this policy. Checked
+ // after the add, as a removed route is shut down
before onRemove forgets it
+ LOG.debug("Route '{}' was removed while it was being
started", route.getId());
+ startedRoutes.remove(route);
+ }
}
}
@@ -525,12 +557,23 @@ public final class ClusteredRoutePolicy extends
RoutePolicySupport implements Ca
implements ExtendedStartupListener, NonManagedService {
@Override
public void notify(CamelEvent event) throws Exception {
- onCamelContextStarted();
+ 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)
+ routeAdded.set(true);
+ handOverLeadershipChange();
+ } else {
+ onCamelContextStarted();
+ }
}
@Override
public boolean isEnabled(CamelEvent event) {
- return event instanceof CamelContextStartedEvent;
+ // a route added while the CamelContext starts is started (or
deferred) with the others. The event also comes
+ // when a stopped route is warmed up again; the extra pass then
has nothing to do
+ return event instanceof CamelContextStartedEvent
+ || event instanceof RouteAddedEvent rae &&
camelContext.isStarted()
+ && stoppedRoutes.contains(rae.getRoute());
}
@Override
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 20907a4bc322..a7121e12016c 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
@@ -20,14 +20,18 @@ import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.Optional;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.apache.camel.CamelContext;
+import org.apache.camel.Route;
import org.apache.camel.ServiceStatus;
import org.apache.camel.builder.RouteBuilder;
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.support.RoutePolicySupport;
import org.apache.camel.support.cluster.AbstractCamelClusterService;
import org.apache.camel.support.cluster.AbstractCamelClusterView;
import org.apache.camel.support.cluster.ClusterServiceSelectors;
@@ -237,6 +241,136 @@ public class ClusteredEdgeCasesTest {
}
}
+ @Test
+ public void testLeadershipTakenBeforeTheCamelContextStarted() throws
Exception {
+ // a cluster service started on its own (not by the CamelContext),
whose view is already the leader
+ TestClusterService cs = new TestClusterService("cs");
+ cs.start();
+ // creates the view
+ CamelClusterView view = cs.getView("my-ns");
+ CamelContext context = new DefaultCamelContext();
+ try {
+ cs.getView().setLeader(true);
+ ClusteredRoutePolicy policy =
ClusteredRoutePolicy.forNamespace(cs, "my-ns");
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("seda:foo").routeId("foo").routePolicy(policy).to("mock:foo");
+ }
+ });
+ context.start();
+
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(
+ () -> assertEquals(ServiceStatus.Started,
context.getRouteController().getRouteStatus("foo")));
+ } finally {
+ context.stop();
+ cs.releaseView(view);
+ cs.stop();
+ }
+ }
+
+ @Test
+ public void testLeadershipTakenWhileARouteIsAdded() throws Exception {
+ TestClusterService cs = new TestClusterService("cs");
+ CamelContext context = new DefaultCamelContext();
+ context.addService(cs);
+ // tells when the policy thread has tried to start the route
+ CountDownLatch startTried = new CountDownLatch(1);
+ context.setRouteController(new DefaultRouteController(context) {
+ @Override
+ public void startRoute(String routeId) throws Exception {
+ super.startRoute(routeId);
+ startTried.countDown();
+ }
+ });
+ context.start();
+ try {
+ ClusteredRoutePolicy policy =
ClusteredRoutePolicy.forNamespace("my-ns");
+ // listed after the clustered policy: this node becomes the leader
while the route is created, before the
+ // route is added to the CamelContext
+ RoutePolicySupport leaderWhileAdding = new RoutePolicySupport() {
+ @Override
+ public void onInit(Route route) {
+ cs.getView().setLeader(true);
+ try {
+ // the policy thread tries to start the route now (the
old code did); the fixed policy does
+ // not, as the route is not added yet, so this wait
ends after its timeout of one second
+ startTried.await(1, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ };
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("seda:foo").routeId("foo").routePolicy(policy,
leaderWhileAdding).to("mock:foo");
+ }
+ });
+
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(
+ () -> assertEquals(ServiceStatus.Started,
context.getRouteController().getRouteStatus("foo")));
+ } finally {
+ context.stop();
+ }
+ }
+
+ @Test
+ public void testRouteRemovedWhileThePolicyStartsIt() throws Exception {
+ TestClusterService cs = new TestClusterService("cs");
+ CamelContext context = new DefaultCamelContext();
+ context.addService(cs);
+ CountDownLatch starting = new CountDownLatch(1);
+ CountDownLatch removed = new CountDownLatch(1);
+ // the start of the route by the policy thread waits until the route
was removed
+ context.setRouteController(new DefaultRouteController(context) {
+ @Override
+ public void startRoute(String routeId) throws Exception {
+ if ("foo".equals(routeId) && starting.getCount() > 0) {
+ starting.countDown();
+ // on the policy thread: the test checks the outcome
(bounded wait, the test fails anyway)
+ removed.await(20, TimeUnit.SECONDS);
+ }
+ super.startRoute(routeId);
+ }
+ });
+ ClusteredRoutePolicy policy =
ClusteredRoutePolicy.forNamespace("my-ns");
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("seda:foo").routeId("foo").routePolicy(policy).to("mock:foo");
+
from("seda:bar").routeId("bar").routePolicy(policy).to("mock:bar");
+ }
+ });
+ context.start();
+ try {
+ cs.getView().setLeader(true);
+ assertTrue(starting.await(10, TimeUnit.SECONDS), "The policy did
not start the route");
+ context.removeRoute("foo");
+ removed.countDown();
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(
+ () -> assertEquals(ServiceStatus.Started,
context.getRouteController().getRouteStatus("bar")));
+
+ // a route that is not clustered with the same id
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("seda:foo2").routeId("foo").to("mock:foo2");
+ }
+ });
+ assertEquals(ServiceStatus.Started,
context.getRouteController().getRouteStatus("foo"));
+
+ // the policy must not stop it when this node loses the leadership
+ cs.getView().setLeader(false);
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(
+ () -> assertEquals(ServiceStatus.Stopped,
context.getRouteController().getRouteStatus("bar")));
+ await().during(1, TimeUnit.SECONDS).atMost(2,
TimeUnit.SECONDS).untilAsserted(
+ () -> assertEquals(ServiceStatus.Started,
context.getRouteController().getRouteStatus("foo")));
+ } finally {
+ context.stop();
+ }
+ }
+
// *********************************
// Helpers
// *********************************
