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 dc7e05f5f470 CAMEL-26865: camel-cluster - ClusteredRoutePolicy starts
and stops each route on its own, and tries a failed start again (#29666)
dc7e05f5f470 is described below
commit dc7e05f5f4700f3a1acb79bcd0fb570f419ab5d1
Author: allthingssecurity <[email protected]>
AuthorDate: Thu Nov 12 14:30:54 2026 +0530
CAMEL-26865: camel-cluster - ClusteredRoutePolicy starts and stops each
route on its own, and tries a failed start again (#29666)
When this node takes the leadership, the policy starts its routes in
one try block: a route whose start failed (an endpoint that cannot
connect yet) was only logged, the routes after it in the loop were not
started, and nothing tried again, as the leadership was already taken.
The leader then ran none (or only some) of the routes until the
leadership changed again. Each route is now started on its own; a
route that fails stays with the stopped routes, and after a retry delay
(5 seconds, setRetryDelay) the policy thread applies the leadership
again: it starts the stopped routes if this node is still the leader.
Each failed attempt is numbered in the log. The retry runs on a timer
thread that ends once it fired, so a policy keeps no thread.
The stop on a leadership loss had the same single try: a route that
failed to stop kept the routes after it running, and stayed with the
started routes, so it was not started again on the next leadership.
Each route is now stopped on its own; a route that is no longer
running after a failed stop is recorded as stopped, and the stop of a
route still running is tried again after the retry delay.
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../camel/impl/cluster/ClusteredRoutePolicy.java | 185 ++++++++++++++++-----
.../camel/cluster/ClusteredEdgeCasesTest.java | 159 ++++++++++++++++++
.../user-manual/modules/ROOT/pages/clustering.adoc | 5 +
3 files changed, 306 insertions(+), 43 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 46d9db812017..1c0347d5ebd7 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
@@ -17,6 +17,7 @@
package org.apache.camel.impl.cluster;
import java.time.Duration;
+import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
@@ -63,6 +64,8 @@ public final class ClusteredRoutePolicy extends
RoutePolicySupport implements Ca
private static final String THREAD_NAME = "ClusteredRoutePolicy";
// how long the policy thread waits for another leadership change before
it exits
private static final long LEADERSHIP_THREAD_KEEP_ALIVE_MILLIS = 1000;
+ // the delay before the leader tries again to start a route whose start
failed
+ private static final Duration DEFAULT_RETRY_DELAY = Duration.ofSeconds(5);
private final AtomicBoolean leader;
private final Set<Route> autoStartupRoutes;
@@ -79,8 +82,10 @@ 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;
+ // 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
+ private final AtomicBoolean reapply;
+ private final AtomicBoolean retryScheduled;
private CamelClusterService clusterService;
private volatile boolean startManagedRoutesEarly;
private volatile boolean initialDelayElapsed;
@@ -88,6 +93,10 @@ public final class ClusteredRoutePolicy extends
RoutePolicySupport implements Ca
private Duration initialDelay;
private volatile ExecutorService leadershipExecutor;
private volatile ScheduledExecutorService initialDelayExecutor;
+ private volatile ScheduledExecutorService retryExecutor;
+ private volatile Duration retryDelay = DEFAULT_RETRY_DELAY;
+ // failed starts in a row per route id, for the log
+ private final Map<String, Integer> failedStarts = new
ConcurrentHashMap<>();
private CamelContext camelContext;
@@ -104,7 +113,8 @@ public final class ClusteredRoutePolicy extends
RoutePolicySupport implements Ca
this.lock = new ReentrantLock();
this.retainLock = new ReentrantLock();
this.leadershipChangePending = new AtomicBoolean();
- this.routeAdded = new AtomicBoolean();
+ this.reapply = new AtomicBoolean();
+ this.retryScheduled = 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();
@@ -162,6 +172,22 @@ public final class ClusteredRoutePolicy extends
RoutePolicySupport implements Ca
this.initialDelay = initialDelay;
}
+ public Duration getRetryDelay() {
+ return retryDelay;
+ }
+
+ /**
+ * The delay before this node, while it is the leader, tries again to
start a route whose start failed (5 seconds by
+ * default). It must be positive.
+ */
+ public void setRetryDelay(Duration retryDelay) {
+ ObjectHelper.notNull(retryDelay, "retryDelay");
+ if (retryDelay.toMillis() < 1) {
+ throw new IllegalArgumentException("retryDelay must be positive: "
+ retryDelay);
+ }
+ this.retryDelay = retryDelay;
+ }
+
// ****************************************************
// life-cycle
// ****************************************************
@@ -237,6 +263,7 @@ public final class ClusteredRoutePolicy extends
RoutePolicySupport implements Ca
retainLock.unlock();
}
autoStartupRoutes.remove(route);
+ failedStarts.remove(route.getId());
// a route added later with the same id must not be started or stopped
by this policy
startedRoutes.remove(route);
stoppedRoutes.remove(route);
@@ -269,6 +296,13 @@ public final class ClusteredRoutePolicy extends
RoutePolicySupport implements Ca
initialDelayExecutor = null;
camelContext.getExecutorServiceManager().shutdownNow(scheduler);
}
+ // no retry is scheduled from now on
+ retryScheduled.set(true);
+ ScheduledExecutorService retry = retryExecutor;
+ if (retry != null) {
+ retryExecutor = null;
+ camelContext.getExecutorServiceManager().shutdownNow(retry);
+ }
}
}
@@ -343,7 +377,7 @@ public final class ClusteredRoutePolicy extends
RoutePolicySupport implements Ca
}
// a route of this policy was added to the CamelContext since the
last change
- boolean added = routeAdded.getAndSet(false);
+ boolean added = reapply.getAndSet(false);
boolean isLeader = view.getLocalMember().isLeader();
if (isLeader && leader.compareAndSet(false, isLeader)) {
LOG.debug("Leadership taken");
@@ -351,10 +385,15 @@ 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();
+ } else if (added) {
+ // a route that could not be started when the leadership was
taken (not added to the CamelContext yet,
+ // or its start failed), or that failed to stop when it was
lost: doStartManagedRoutes only starts the
+ // routes that are still stopped, doStopManagedRoutes only
stops the routes that are still started
+ if (isLeader) {
+ startManagedRoutes();
+ } else {
+ stopManagedRoutes();
+ }
}
} finally {
lock.unlock();
@@ -396,36 +435,86 @@ public final class ClusteredRoutePolicy extends
RoutePolicySupport implements Ca
return;
}
- try {
- for (Route route : stoppedRoutes) {
- 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);
+ boolean failed = false;
+ // each route on its own: a route that fails to start stays with the
stopped routes, and is tried again while
+ // this node is the leader
+ for (Route route : stoppedRoutes) {
+ 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.isStartable() && autostart) {
- LOG.debug("Starting route '{}'", route.getId());
+ if (status.isStartable() && autostart) {
+ LOG.debug("Starting route '{}'", route.getId());
+ try {
camelContext.getRouteController().startRoute(route.getId());
+ } catch (Exception e) {
+ failed = true;
+ int attempt = failedStarts.merge(route.getId(), 1,
Integer::sum);
+ handleFailure("Failed to start route " + route.getId() + "
(attempt #" + attempt + "). Trying again in "
+ + retryDelay.toMillis() + " millis while
this node is the leader.",
+ e);
+ continue;
+ }
+ Integer failures = failedStarts.remove(route.getId());
+ if (failures != null) {
+ LOG.info("Route {} started after {} failed attempt(s)",
route.getId(), failures);
+ }
- 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);
- }
+ 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);
}
}
+ }
- stoppedRoutes.removeAll(startedRoutes);
- } catch (Exception e) {
- handleException(e);
+ stoppedRoutes.removeAll(startedRoutes);
+ if (failed) {
+ scheduleRetry();
+ }
+ }
+
+ // the policy thread applies the leadership to the routes again after the
retry delay (starts the stopped routes if
+ // this node is still the leader, stops the started ones if not). A timer
thread per retry, which ends once it fired,
+ // so a policy does not keep a thread (a policy per route with
ClusteredRoutePolicyFactory)
+ private void scheduleRetry() {
+ if (!isRunAllowed() || retryScheduled.getAndSet(true)) {
+ return;
+ }
+ ScheduledExecutorService scheduler
+ =
camelContext.getExecutorServiceManager().newSingleThreadScheduledExecutor(this,
THREAD_NAME + "Retry");
+ retryExecutor = scheduler;
+ try {
+ scheduler.schedule(() -> {
+ try {
+ retryExecutor = null;
+
camelContext.getExecutorServiceManager().shutdown(scheduler);
+ } finally {
+ retryScheduled.set(false);
+ reapply.set(true);
+ handOverLeadershipChange();
+ }
+ }, retryDelay.toMillis(), TimeUnit.MILLISECONDS);
+ } catch (RejectedExecutionException e) {
+ // the policy is being shut down
+ LOG.debug("Not trying again as ClusteredRoutePolicy for namespace
{} is shut down", namespace);
+ }
+ }
+
+ private void handleFailure(String message, Exception e) {
+ if (getExceptionHandler() != null) {
+ getExceptionHandler().handleException(message, e);
+ } else {
+ LOG.warn(message, e);
}
}
@@ -444,20 +533,30 @@ public final class ClusteredRoutePolicy extends
RoutePolicySupport implements Ca
return;
}
- try {
- for (Route route : startedRoutes) {
- ServiceStatus status = getStatus(route);
- if (status != null && status.isStoppable()) {
- LOG.debug("Stopping route '{}'", route.getId());
+ boolean stopFailed = false;
+ // each route on its own, so a route that fails to stop does not keep
the others running
+ for (Route route : startedRoutes) {
+ ServiceStatus status = getStatus(route);
+ if (status != null && status.isStoppable()) {
+ LOG.debug("Stopping route '{}'", route.getId());
+ try {
stopRoute(route);
-
- stoppedRoutes.add(route);
+ } catch (Exception e) {
+ handleFailure("Failed to stop route " + route.getId(), e);
+ if (getStatus(route).isStoppable()) {
+ // still running: it stays with the started routes,
and the stop is tried again
+ stopFailed = true;
+ continue;
+ }
}
+
+ stoppedRoutes.add(route);
}
+ }
- startedRoutes.removeAll(stoppedRoutes);
- } catch (Exception e) {
- handleException(e);
+ startedRoutes.removeAll(stoppedRoutes);
+ if (stopFailed) {
+ scheduleRetry();
}
}
@@ -560,7 +659,7 @@ public final class ClusteredRoutePolicy extends
RoutePolicySupport implements Ca
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);
+ reapply.set(true);
handOverLeadershipChange();
} else {
onCamelContextStarted();
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 a7121e12016c..6af19850ad56 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
@@ -19,11 +19,17 @@ package org.apache.camel.cluster;
import java.time.Duration;
import java.util.Collections;
import java.util.List;
+import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.camel.CamelContext;
+import org.apache.camel.Consumer;
+import org.apache.camel.Endpoint;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
import org.apache.camel.Route;
import org.apache.camel.ServiceStatus;
import org.apache.camel.builder.RouteBuilder;
@@ -31,6 +37,9 @@ 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.DefaultComponent;
+import org.apache.camel.support.DefaultConsumer;
+import org.apache.camel.support.DefaultEndpoint;
import org.apache.camel.support.RoutePolicySupport;
import org.apache.camel.support.cluster.AbstractCamelClusterService;
import org.apache.camel.support.cluster.AbstractCamelClusterView;
@@ -371,10 +380,160 @@ public class ClusteredEdgeCasesTest {
}
}
+ @Test
+ public void testRouteThatFailsToStartIsStartedAgain() throws Exception {
+ TestClusterService cs = new TestClusterService("cs");
+ CamelContext context = new DefaultCamelContext();
+ context.addService(cs);
+ // the first start of the consumer of route a fails (a broker that is
not reachable yet)
+ AtomicInteger failures = new AtomicInteger(1);
+ context.addComponent("failing", new FailingStartComponent(failures));
+ ClusteredRoutePolicy policy =
ClusteredRoutePolicy.forNamespace("my-ns");
+ policy.setRetryDelay(Duration.ofMillis(100));
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("failing:a").routeId("a").routePolicy(policy).to("mock:a");
+ from("seda:b").routeId("b").routePolicy(policy).to("mock:b");
+ }
+ });
+ context.start();
+ try {
+ cs.getView().setLeader(true);
+
+ // both routes run on the leader: the other one right away, the
failed one once its start is tried again
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
+ assertEquals(ServiceStatus.Started,
context.getRouteController().getRouteStatus("b"));
+ assertEquals(ServiceStatus.Started,
context.getRouteController().getRouteStatus("a"));
+ });
+ assertEquals(-1, failures.get(), "The start of route a did not
fail once");
+ } finally {
+ context.stop();
+ }
+ }
+
+ @Test
+ public void testRetryDoesNotStartARouteOnAFollower() throws Exception {
+ TestClusterService cs = new TestClusterService("cs");
+ CamelContext context = new DefaultCamelContext();
+ context.addService(cs);
+ // the consumer of route a cannot start for now
+ AtomicInteger failures = new AtomicInteger(Integer.MAX_VALUE);
+ FailingStartComponent component = new FailingStartComponent(failures);
+ context.addComponent("failing", component);
+ ClusteredRoutePolicy policy =
ClusteredRoutePolicy.forNamespace("my-ns");
+ policy.setRetryDelay(Duration.ofMillis(100));
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("failing:a").routeId("a").routePolicy(policy).to("mock:a");
+ }
+ });
+ context.start();
+ try {
+ cs.getView().setLeader(true);
+ // the leader tries again
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
component.attempts.get() >= 2);
+
+ // the node loses the leadership, and the consumer could start
now: the retry must not start the route
+ cs.getView().setLeader(false);
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
!policy.isLeader());
+ failures.set(0);
+ await().during(1, TimeUnit.SECONDS).atMost(5,
TimeUnit.SECONDS).untilAsserted(
+ () -> assertEquals(ServiceStatus.Stopped,
context.getRouteController().getRouteStatus("a")));
+ } finally {
+ context.stop();
+ }
+ }
+
+ @Test
+ public void testRouteThatFailsToStopIsStartedAgainOnTheNextLeadership()
throws Exception {
+ TestClusterService cs = new TestClusterService("cs");
+ CamelContext context = new DefaultCamelContext();
+ context.addService(cs);
+ ClusteredRoutePolicy policy =
ClusteredRoutePolicy.forNamespace("my-ns");
+ // the first stop of route a fails (in a route policy, after the route
stopped)
+ AtomicInteger stopFailures = new AtomicInteger(1);
+ RoutePolicySupport failingStop = new RoutePolicySupport() {
+ @Override
+ public void onStop(Route route) {
+ if ("a".equals(route.getId()) &&
stopFailures.getAndDecrement() > 0) {
+ throw new IllegalStateException("Cannot stop route a");
+ }
+ }
+ };
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("seda:a").routeId("a").routePolicy(policy,
failingStop).to("mock:a");
+ from("seda:b").routeId("b").routePolicy(policy,
failingStop).to("mock:b");
+ }
+ });
+ context.start();
+ try {
+ cs.getView().setLeader(true);
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
+ assertEquals(ServiceStatus.Started,
context.getRouteController().getRouteStatus("a"));
+ assertEquals(ServiceStatus.Started,
context.getRouteController().getRouteStatus("b"));
+ });
+
+ // the other route is stopped too
+ cs.getView().setLeader(false);
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
+ assertEquals(ServiceStatus.Stopped,
context.getRouteController().getRouteStatus("b"));
+ assertEquals(ServiceStatus.Stopped,
context.getRouteController().getRouteStatus("a"));
+ });
+ assertEquals(0, stopFailures.get(), "The stop of route a did not
fail once");
+
+ // and both are started on the next leadership
+ cs.getView().setLeader(true);
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
+ assertEquals(ServiceStatus.Started,
context.getRouteController().getRouteStatus("a"));
+ assertEquals(ServiceStatus.Started,
context.getRouteController().getRouteStatus("b"));
+ });
+ } finally {
+ context.stop();
+ }
+ }
+
// *********************************
// Helpers
// *********************************
+ // a consumer whose start fails as many times as failures says
+ private static final class FailingStartComponent extends DefaultComponent {
+ private final AtomicInteger failures;
+ private final AtomicInteger attempts = new AtomicInteger();
+
+ FailingStartComponent(AtomicInteger failures) {
+ this.failures = failures;
+ }
+
+ @Override
+ protected Endpoint createEndpoint(String uri, String remaining,
Map<String, Object> parameters) {
+ return new DefaultEndpoint(uri, this) {
+ @Override
+ public Producer createProducer() {
+ throw new UnsupportedOperationException("No producer");
+ }
+
+ @Override
+ public Consumer createConsumer(Processor processor) {
+ return new DefaultConsumer(this, processor) {
+ @Override
+ protected void doStart() throws Exception {
+ attempts.incrementAndGet();
+ if (failures.getAndDecrement() > 0) {
+ throw new IllegalStateException("Cannot start
the consumer yet");
+ }
+ super.doStart();
+ }
+ };
+ }
+ };
+ }
+ }
+
private static class TestClusterView extends AbstractCamelClusterView {
private volatile boolean leader;
private volatile boolean running;
diff --git a/docs/user-manual/modules/ROOT/pages/clustering.adoc
b/docs/user-manual/modules/ROOT/pages/clustering.adoc
index 806986ce5974..98a702b21282 100644
--- a/docs/user-manual/modules/ROOT/pages/clustering.adoc
+++ b/docs/user-manual/modules/ROOT/pages/clustering.adoc
@@ -213,6 +213,11 @@ context.addRoutes(new RouteBuilder {
});
----
+
+When this node is the leader and a route fails to start (for example its
endpoint cannot connect yet), the policy
+tries again to start it every 5 seconds while the node keeps the leadership,
and logs each failed attempt (at WARN with
+the default exception handler). The delay can be set with
`setRetryDelay(Duration)` on a `ClusteredRoutePolicy` created with
`forNamespace`; the
+policies of `ClusteredRoutePolicyFactory` and `ClusteredRouteController` use
the default.
++
To apply the same policy to all the routes a dedicated _RoutePolicyFactory_
can be used:
+
._Java-only: using ClusteredRoutePolicyFactory_
