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 8388b0ae4dd3 CAMEL-26756: camel-cluster - ClusteredRouteController:
stop the cluster service once the routes have stopped (#29531)
8388b0ae4dd3 is described below
commit 8388b0ae4dd3c7cbeecbd713117ac6365e6558dc
Author: allthingssecurity <[email protected]>
AuthorDate: Tue Nov 10 18:22:38 2026 +0530
CAMEL-26756: camel-cluster - ClusteredRouteController: stop the cluster
service once the routes have stopped (#29531)
When the CamelContext stops, it stops its route controller before the
routes.
ClusteredRouteController.doStop stopped the cluster service there, giving up
the leadership while the routes were still draining their in-flight
exchanges, so a follower could take over and run the same routes during a
graceful stop. Stop the cluster service once the routes have stopped, and
unregister the controller's listeners.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../impl/cluster/ClusteredRouteController.java | 86 ++++-
.../cluster/ClusteredRouteControllerStopTest.java | 350 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 10 +
3 files changed, 445 insertions(+), 1 deletion(-)
diff --git
a/core/camel-cluster/src/main/java/org/apache/camel/impl/cluster/ClusteredRouteController.java
b/core/camel-cluster/src/main/java/org/apache/camel/impl/cluster/ClusteredRouteController.java
index 87bf5f0a6af6..62b158581cc8 100644
---
a/core/camel-cluster/src/main/java/org/apache/camel/impl/cluster/ClusteredRouteController.java
+++
b/core/camel-cluster/src/main/java/org/apache/camel/impl/cluster/ClusteredRouteController.java
@@ -30,13 +30,19 @@ import java.util.concurrent.CopyOnWriteArraySet;
import org.apache.camel.CamelContext;
import org.apache.camel.NamedNode;
+import org.apache.camel.NonManagedService;
import org.apache.camel.Route;
import org.apache.camel.RuntimeCamelException;
import org.apache.camel.cluster.CamelClusterService;
import org.apache.camel.impl.engine.DefaultRouteController;
import org.apache.camel.model.RouteDefinition;
+import org.apache.camel.spi.CamelEvent;
+import org.apache.camel.spi.CamelEvent.CamelContextRoutesStoppedEvent;
+import org.apache.camel.spi.LifecycleStrategy;
import org.apache.camel.spi.RoutePolicy;
import org.apache.camel.spi.RoutePolicyFactory;
+import org.apache.camel.support.LifecycleStrategySupport;
+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.ServiceHelper;
@@ -55,6 +61,8 @@ public class ClusteredRouteController extends
DefaultRouteController {
private final List<ClusteredRouteFilter> filters;
private final PolicyFactory policyFactory;
private final ClusteredRouteConfiguration defaultConfiguration;
+ private final RoutesStoppedListener routesStoppedListener;
+ private final LifecycleStrategy contextStoppedStrategy;
private CamelClusterService clusterService;
private CamelClusterService.Selector clusterServiceSelector;
@@ -64,6 +72,8 @@ public class ClusteredRouteController extends
DefaultRouteController {
this.filters = new ArrayList<>();
this.clusterServiceSelector = ClusterServiceSelectors.DEFAULT_SELECTOR;
this.policyFactory = new PolicyFactory();
+ this.routesStoppedListener = new RoutesStoppedListener();
+ this.contextStoppedStrategy = new ContextStoppedStrategy();
this.defaultConfiguration = new ClusteredRouteConfiguration();
this.defaultConfiguration.setInitialDelay(Duration.ofMillis(0));
@@ -141,7 +151,8 @@ public class ClusteredRouteController extends
DefaultRouteController {
}
/**
- * Set the cluster service to use.
+ * Set the cluster service to use. The route controller starts it if it is
not started yet, and stops it when the
+ * CamelContext stops, once the routes have stopped.
*/
public void setClusterService(CamelClusterService clusterService) {
ObjectHelper.notNull(clusterService, "CamelClusterService");
@@ -193,11 +204,43 @@ public class ClusteredRouteController extends
DefaultRouteController {
clusterService.start();
}
+ if
(!context.getManagementStrategy().getEventNotifiers().contains(routesStoppedListener))
{
+
context.getManagementStrategy().addEventNotifier(routesStoppedListener);
+ }
+ // a fallback for an EventFactory that does not create the
CamelContextRoutesStoppedEvent
+ context.addLifecycleStrategy(contextStoppedStrategy);
+
super.doStart();
}
@Override
public void doStop() throws Exception {
+ if (getCamelContext() != null && getCamelContext().isStopping()) {
+ // The CamelContext stops the route controller before the routes.
The routes must keep the leadership until
+ // their consumers have stopped and their in-flight exchanges are
done, or another node would start them in
+ // the meantime, so the cluster service is stopped once the routes
have stopped.
+ LOGGER.debug("CamelContext is stopping: the cluster service is
stopped once the routes have stopped");
+ return;
+ }
+
+ stopClusterService();
+ }
+
+ @Override
+ protected void doShutdown() throws Exception {
+ final CamelContext context = getCamelContext();
+ if (context != null && !context.isStopping()) {
+ // The route controller is shut down while the CamelContext runs,
so it must no longer stop the cluster
+ // service when the CamelContext stops. While the CamelContext
stops, the listeners stay registered until it
+ // has stopped, as they stop the cluster service once the routes
have stopped.
+
context.getManagementStrategy().removeEventNotifier(routesStoppedListener);
+ context.getLifecycleStrategies().remove(contextStoppedStrategy);
+ }
+
+ super.doShutdown();
+ }
+
+ private void stopClusterService() {
if (ServiceHelper.isStarted(clusterService)) {
// Stop the cluster service.
clusterService.stop();
@@ -213,6 +256,47 @@ public class ClusteredRouteController extends
DefaultRouteController {
super.setCamelContext(camelContext);
}
+ /**
+ * Stops the cluster service once the routes have stopped, when the
CamelContext stops.
+ */
+ private final class RoutesStoppedListener extends
SimpleEventNotifierSupport implements NonManagedService {
+ RoutesStoppedListener() {
+ // only the CamelContext events are needed (the exchange and step
events are ignored by default)
+ setIgnoreRouteEvents(true);
+ setIgnoreServiceEvents(true);
+ }
+
+ @Override
+ public boolean isEnabled(CamelEvent event) {
+ return event instanceof CamelContextRoutesStoppedEvent;
+ }
+
+ @Override
+ public void notify(CamelEvent event) throws Exception {
+ stopClusterService();
+ }
+ }
+
+ /**
+ * Stops the cluster service when the CamelContext has stopped, if it is
still started. The cluster service is
+ * normally stopped earlier, once the routes have stopped, but a custom
EventFactory may not create the
+ * CamelContextRoutesStoppedEvent. Then unregisters the listeners of the
route controller, which registers them
+ * again when it is started again, so a route controller replaced while
the CamelContext was stopped leaves none
+ * behind.
+ */
+ private final class ContextStoppedStrategy extends
LifecycleStrategySupport {
+ @Override
+ public void onContextStopped(CamelContext context) {
+ try {
+ stopClusterService();
+ } catch (Exception e) {
+ LOGGER.warn("Error stopping the cluster service {}. This
exception is ignored.", clusterService, e);
+ }
+
context.getManagementStrategy().removeEventNotifier(routesStoppedListener);
+ context.getLifecycleStrategies().remove(this);
+ }
+ }
+
// *******************************
// Factories
// *******************************
diff --git
a/core/camel-core/src/test/java/org/apache/camel/cluster/ClusteredRouteControllerStopTest.java
b/core/camel-core/src/test/java/org/apache/camel/cluster/ClusteredRouteControllerStopTest.java
new file mode 100644
index 000000000000..d41457612527
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/cluster/ClusteredRouteControllerStopTest.java
@@ -0,0 +1,350 @@
+/*
+ * 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.cluster;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.camel.CamelContext;
+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.event.DefaultEventFactory;
+import org.apache.camel.spi.CamelEvent;
+import org.apache.camel.support.LifecycleStrategySupport;
+import org.apache.camel.support.cluster.AbstractCamelClusterService;
+import org.apache.camel.support.cluster.AbstractCamelClusterView;
+import org.junit.jupiter.api.Test;
+
+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.assertTrue;
+
+/**
+ * The CamelContext stops its route controller before the routes, and the
ClusteredRouteController must not stop the
+ * cluster service (giving up the leadership) before the consumers of the
routes have stopped.
+ */
+public class ClusteredRouteControllerStopTest {
+
+ @Test
+ public void testCamelContextClusterServiceStoppedAfterRoutes() throws
Exception {
+ TestClusterService cs = new TestClusterService("cs");
+ CamelContext context = new DefaultCamelContext();
+ context.addService(cs);
+ ClusteredRouteController controller = new ClusteredRouteController();
+
+ doTestStopWithInflightExchange(context, controller, cs);
+ }
+
+ @Test
+ public void testControllerClusterServiceStoppedAfterRoutes() throws
Exception {
+ TestClusterService cs = new TestClusterService("cs");
+ CamelContext context = new DefaultCamelContext();
+ ClusteredRouteController controller = new ClusteredRouteController();
+ // a cluster service that is not a service of the CamelContext, so the
route controller starts and stops it
+ controller.setClusterService(cs);
+
+ doTestStopWithInflightExchange(context, controller, cs);
+ }
+
+ @Test
+ public void testControllerStoppedOnItsOwnStopsClusterService() throws
Exception {
+ TestClusterService cs = new TestClusterService("cs");
+ CamelContext context = new DefaultCamelContext();
+ ClusteredRouteController controller = new ClusteredRouteController();
+ controller.setClusterService(cs);
+ controller.setNamespace("my-ns");
+ context.setRouteController(controller);
+ context.start();
+ try {
+ assertTrue(cs.isStarted());
+
+ // the CamelContext keeps running, so the cluster service is
stopped right away
+ controller.stop();
+ assertFalse(cs.isStarted());
+ } finally {
+ context.stop();
+ }
+ }
+
+ @Test
+ public void testControllerClusterServiceStoppedOnEachCamelContextStop()
throws Exception {
+ TestClusterService cs = new TestClusterService("cs");
+ CamelContext context = new DefaultCamelContext();
+ ClusteredRouteController controller = new ClusteredRouteController();
+ controller.setClusterService(cs);
+ controller.setNamespace("my-ns");
+ context.setRouteController(controller);
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("seda:foo").routeId("foo").to("mock:foo");
+ }
+ });
+
+ AtomicReference<Boolean> startedWhenContextStopped =
probeClusterServiceWhenContextStopped(context, cs);
+
+ try {
+ for (int i = 0; i < 2; i++) {
+ context.start();
+ assertTrue(cs.isStarted());
+ // the route controller registers its listeners once, also
when the CamelContext is started again
+ assertEquals(1, countControllerListeners(context));
+ startedWhenContextStopped.set(null);
+ context.stop();
+ assertFalse(cs.isStarted());
+ // stopped once the routes have stopped, not by the fallback
when the CamelContext has stopped
+ assertEquals(Boolean.FALSE, startedWhenContextStopped.get());
+ // and unregistered when the CamelContext has stopped
+ assertEquals(0, countControllerListeners(context));
+ }
+ } finally {
+ context.stop();
+ }
+ }
+
+ @Test
+ public void testControllerClusterServiceStoppedWithoutRoutesStoppedEvent()
throws Exception {
+ TestClusterService cs = new TestClusterService("cs");
+ CamelContext context = new DefaultCamelContext();
+ // a custom EventFactory that does not create the
CamelContextRoutesStoppedEvent
+ context.getManagementStrategy().setEventFactory(new
DefaultEventFactory() {
+ @Override
+ public CamelEvent
createCamelContextRoutesStoppedEvent(CamelContext context) {
+ return null;
+ }
+ });
+ ClusteredRouteController controller = new ClusteredRouteController();
+ // a cluster service that is not a service of the CamelContext, so
only the route controller stops it
+ controller.setClusterService(cs);
+ controller.setNamespace("my-ns");
+ context.setRouteController(controller);
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("seda:foo").routeId("foo").to("mock:foo");
+ }
+ });
+
+ try {
+ context.start();
+ assertTrue(cs.isStarted());
+ context.stop();
+ // the cluster service is stopped when the CamelContext has stopped
+ assertFalse(cs.isStarted());
+ } finally {
+ context.stop();
+ }
+ }
+
+ @Test
+ public void testControllerShutdownOnItsOwnUnregistersListeners() throws
Exception {
+ TestClusterService cs = new TestClusterService("cs");
+ CamelContext context = new DefaultCamelContext();
+ ClusteredRouteController controller = new ClusteredRouteController();
+ controller.setClusterService(cs);
+ controller.setNamespace("my-ns");
+ context.setRouteController(controller);
+
+ try {
+ context.start();
+ assertEquals(1, countControllerListeners(context));
+
+ // the route controller is shut down while the CamelContext keeps
running
+ controller.shutdown();
+ assertFalse(cs.isStarted());
+ assertEquals(0, countControllerListeners(context));
+
+ // the cluster service is used elsewhere: stopping the
CamelContext must not stop it through this controller
+ cs.start();
+ context.stop();
+ assertTrue(cs.isStarted());
+ } finally {
+ cs.stop();
+ context.stop();
+ }
+ }
+
+ /**
+ * Records whether the cluster service is still started when the
CamelContext has stopped, before the fallback of
+ * the route controller (whose lifecycle strategy is added after this one,
when the route controller starts).
+ */
+ private static AtomicReference<Boolean>
probeClusterServiceWhenContextStopped(
+ CamelContext context, TestClusterService cs) {
+ AtomicReference<Boolean> started = new AtomicReference<>();
+ context.addLifecycleStrategy(new LifecycleStrategySupport() {
+ @Override
+ public void onContextStopped(CamelContext context) {
+ started.set(cs.isStarted());
+ }
+ });
+ return started;
+ }
+
+ private static long countControllerListeners(CamelContext context) {
+ long notifiers =
context.getManagementStrategy().getEventNotifiers().stream()
+ .filter(n -> n.getClass().getEnclosingClass() ==
ClusteredRouteController.class)
+ .count();
+ long strategies = context.getLifecycleStrategies().stream()
+ .filter(s -> s.getClass().getEnclosingClass() ==
ClusteredRouteController.class)
+ .count();
+ assertEquals(notifiers, strategies);
+ return notifiers;
+ }
+
+ private void doTestStopWithInflightExchange(
+ CamelContext context, ClusteredRouteController controller,
TestClusterService cs)
+ throws Exception {
+ CountDownLatch processing = new CountDownLatch(1);
+ CountDownLatch release = new CountDownLatch(1);
+
+ controller.setNamespace("my-ns");
+ context.setRouteController(controller);
+ context.getShutdownStrategy().setTimeout(30);
+ AtomicReference<Boolean> startedWhenContextStopped =
probeClusterServiceWhenContextStopped(context, cs);
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("seda:foo").routeId("foo")
+ .process(e -> {
+ processing.countDown();
+ release.await(20, TimeUnit.SECONDS);
+ })
+ .to("mock:foo");
+ }
+ });
+ context.start();
+
+ Thread stopper = new Thread(context::stop, "CamelContextStopper");
+ try {
+ cs.getView().setLeader(true);
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(
+ () -> assertEquals(ServiceStatus.Started,
context.getRouteController().getRouteStatus("foo")));
+
+ context.createProducerTemplate().sendBody("seda:foo", "Hello");
+ assertTrue(processing.await(10, TimeUnit.SECONDS));
+
+ stopper.start();
+ // the CamelContext stops the route controller first, and then
waits for the in-flight exchange
+ await().atMost(10, TimeUnit.SECONDS).until(controller::isStopped);
+
+ // the route still processes its exchange, so this node must still
be the leader
+ assertTrue(cs.isStarted());
+ assertTrue(cs.getView().isRunAllowed());
+ assertTrue(cs.getView().getLocalMember().isLeader());
+ } finally {
+ release.countDown();
+ if (!stopper.isAlive()) {
+ // the test failed before the stop
+ context.stop();
+ }
+ stopper.join(TimeUnit.SECONDS.toMillis(30));
+ }
+
+ assertFalse(stopper.isAlive());
+ assertEquals(ServiceStatus.Stopped,
context.getRouteController().getRouteStatus("foo"));
+ // the cluster service is stopped once the routes have stopped
+ assertFalse(cs.isStarted());
+ assertFalse(cs.getView().isRunAllowed());
+ assertEquals(Boolean.FALSE, startedWhenContextStopped.get());
+ }
+
+ // *********************************
+ // Helpers
+ // *********************************
+
+ private static class TestClusterView extends AbstractCamelClusterView {
+ private volatile boolean leader;
+
+ TestClusterView(CamelClusterService cluster, String namespace) {
+ super(cluster, namespace);
+ }
+
+ @Override
+ public Optional<CamelClusterMember> getLeader() {
+ return leader ? Optional.of(getLocalMember()) : Optional.empty();
+ }
+
+ @Override
+ public CamelClusterMember getLocalMember() {
+ return new CamelClusterMember() {
+ @Override
+ public boolean isLeader() {
+ return leader;
+ }
+
+ @Override
+ public boolean isLocal() {
+ return true;
+ }
+
+ @Override
+ public String getId() {
+ return getClusterService().getId();
+ }
+ };
+ }
+
+ @Override
+ public List<CamelClusterMember> getMembers() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ protected void doStart() {
+ }
+
+ @Override
+ protected void doStop() {
+ // a view gives up the leadership when it stops
+ leader = false;
+ }
+
+ void setLeader(boolean leader) {
+ this.leader = leader;
+ if (isRunAllowed()) {
+ fireLeadershipChangedEvent(getLeader().orElse(null));
+ }
+ }
+ }
+
+ private static class TestClusterService extends
AbstractCamelClusterService<TestClusterView> {
+ private TestClusterView view;
+
+ TestClusterService(String id) {
+ super(id);
+ }
+
+ @Override
+ protected TestClusterView createView(String namespace) {
+ if (view == null) {
+ view = new TestClusterView(this, namespace);
+ }
+ return view;
+ }
+
+ TestClusterView getView() {
+ return view;
+ }
+ }
+}
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 6b8d57e16367..bc04938d7a7a 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
@@ -7307,6 +7307,16 @@ a route. `ErrorHandlerRedeliveryCustomizer` has a new
default method `useStepHoo
need no change. A custom error handler that wants a failure from the hook
handled like a failure of the step's
processor (with `onException` and redelivery) can override it.
+=== camel-cluster - ClusteredRouteController stops the cluster service once
the routes have stopped
+
+When the CamelContext stops, the `ClusteredRouteController` no longer stops
the cluster service when the route
+controller is stopped (which the CamelContext does before it stops the
routes), but once the routes have stopped. A node
+being stopped can keep (or still take) the leadership of its namespaces until
its routes have finished their in-flight
+exchanges, so another node no longer starts the clustered routes at the same
time. A graceful stop can therefore hand
+over the leadership later than before, by up to the shutdown timeout of the
CamelContext (45 seconds by default). To
+hand over sooner, lower the shutdown timeout (`camel.main.shutdownTimeout`, or
+`CamelContext.getShutdownStrategy().setTimeout`).
+
=== camel-quartz - a route that keeps its trigger fires again when it is added
again or after a restart
With `deleteJob=false` (as with `pauseJob=true`), a route pauses its trigger
when it stops, and only the same endpoint
