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 6086054c4ca5 CAMEL-25500: camel-core - stopping or removing all routes
is one graceful shutdown
6086054c4ca5 is described below
commit 6086054c4ca5a877c539590c0fa6bd354c769d7b
Author: Claus Ibsen <[email protected]>
AuthorDate: Fri Oct 9 23:10:54 2026 +0200
CAMEL-25500: camel-core - stopping or removing all routes is one graceful
shutdown
stopAllRoutes and removeAllRoutes (used by the route reload in dev mode)
stopped the routes one by one, each as its own graceful shutdown. The
ShutdownStrategy defers a direct consumer until the routes sending to it
have drained, but only within one batch. Stopped one at a time, the
downstream route stopped as soon as its own inflight count was 0, a split
upstream then blocked on the direct endpoint, and its exchanges were cut
off at the shutdown timeout.
All the running routes are now stopped as one batch through the
ShutdownStrategy, as when the CamelContext stops (the same pattern as
doShutdownRouteGroup), then their route services are stopped. The
shutdown timeout now applies to all routes together instead of to each
route in turn, which is noted in the 4.23 upgrade guide.
Closes #27632
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
---
.../camel/impl/engine/AbstractCamelContext.java | 63 +++++++++++----
.../apache/camel/impl/StopAllRoutesDrainTest.java | 90 ++++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_23.adoc | 10 +++
3 files changed, 148 insertions(+), 15 deletions(-)
diff --git
a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/AbstractCamelContext.java
b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/AbstractCamelContext.java
index 00655128efbb..03c671ef9c25 100644
---
a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/AbstractCamelContext.java
+++
b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/AbstractCamelContext.java
@@ -1249,26 +1249,59 @@ public abstract class AbstractCamelContext extends
BaseService
}
private void doStopRoutes(RouteController controller,
Comparator<RouteStartupOrder> comparator) throws Exception {
- List<RouteStartupOrder> routesOrdered = new
ArrayList<>(camelContextExtension.getRouteStartupOrder());
- routesOrdered.sort(comparator);
- for (RouteStartupOrder order : routesOrdered) {
- Route route = order.getRoute();
- var status = controller.getRouteStatus(route.getRouteId());
- boolean stopped = status == null || status.isStopped();
- if (!stopped) {
- stopRoute(route.getRouteId(), LoggingLevel.DEBUG);
+ lock.lock();
+ try {
+ List<RouteStartupOrder> routesOrdered = new ArrayList<>();
+ Set<String> ids = new HashSet<>();
+ for (RouteStartupOrder order :
camelContextExtension.getRouteStartupOrder()) {
+ String id = order.getRoute().getRouteId();
+ if (isRouteRunning(controller, id) && ids.add(id)) {
+ routesOrdered.add(order);
+ }
}
- }
- // stop any remainder routes
- for (Route route : getRoutes()) {
- var status = controller.getRouteStatus(route.getRouteId());
- boolean stopped = status == null || status.isStopped();
- if (!stopped) {
- stopRoute(route.getRouteId(), LoggingLevel.DEBUG);
+ // any remainder routes (not in the startup order) are stopped last
+ for (Route route : getRoutes()) {
+ String id = route.getRouteId();
+ RouteService routeService = routeServices.get(id);
+ if (routeService != null && isRouteRunning(controller, id) &&
ids.add(id)) {
+ routesOrdered.add(new DefaultRouteStartupOrder(0, route,
routeService));
+ }
+ }
+ if (routesOrdered.isEmpty()) {
+ return;
+ }
+ routesOrdered.sort(comparator);
+ for (String id : ids) {
+ DefaultRouteError.reset(this, id);
}
+ try {
+ // all the routes as one graceful shutdown, as when
CamelContext stops: a route that consumes from
+ // another (direct) keeps running until the routes sending to
it have completed their inflight
+ // exchanges; stopped one by one, it stopped first and those
exchanges were cut off (CAMEL-25500)
+ getShutdownStrategy().shutdown(this, routesOrdered,
getShutdownStrategy().getTimeout(),
+ getShutdownStrategy().getTimeUnit());
+ for (RouteStartupOrder order : routesOrdered) {
+ RouteService routeService =
routeServices.get(order.getRoute().getRouteId());
+ if (routeService != null) {
+ stopRouteService(routeService, false,
LoggingLevel.DEBUG);
+ }
+ }
+ } catch (Exception e) {
+ for (String id : ids) {
+ DefaultRouteError.set(this, id, Phase.STOP, e);
+ }
+ throw e;
+ }
+ } finally {
+ lock.unlock();
}
}
+ private static boolean isRouteRunning(RouteController controller, String
routeId) {
+ var status = controller.getRouteStatus(routeId);
+ return status != null && !status.isStopped();
+ }
+
public void stopAllRoutes() throws Exception {
RouteController controller = getRouteController();
if (controller == null) {
diff --git
a/core/camel-core/src/test/java/org/apache/camel/impl/StopAllRoutesDrainTest.java
b/core/camel-core/src/test/java/org/apache/camel/impl/StopAllRoutesDrainTest.java
new file mode 100644
index 000000000000..3a3f3b48e643
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/impl/StopAllRoutesDrainTest.java
@@ -0,0 +1,90 @@
+/*
+ * 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.impl;
+
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.ServiceStatus;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.junit.jupiter.api.BeforeEach;
+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.assertTrue;
+
+/**
+ * CAMEL-25500: stopping or removing all routes (as a route reload does) is
one graceful shutdown, so a route that sends
+ * to another over direct completes its inflight exchanges, instead of being
cut off when the route it sends to was
+ * stopped first.
+ */
+public class StopAllRoutesDrainTest extends ContextTestSupport {
+
+ private MockEndpoint shipped;
+
+ @Override
+ @BeforeEach
+ public void setUp() throws Exception {
+ super.setUp();
+ context.getShutdownStrategy().setTimeout(10);
+ context.getShutdownStrategy().setTimeUnit(TimeUnit.SECONDS);
+ shipped = getMockEndpoint("mock:shipped");
+ template.sendBody("seda:start", List.of("A", "B", "C", "D"));
+ // the split is still sending when the routes are stopped
+ await().atMost(5, TimeUnit.SECONDS).until(() ->
shipped.getReceivedCounter() >= 1);
+ }
+
+ @Test
+ public void removeAllRoutesDrainsTheSplit() throws Exception {
+ context.getRouteController().removeAllRoutes();
+
+ assertEquals(List.of("A", "B", "C", "D"), shippedBodies());
+ assertTrue(context.getRoutes().isEmpty());
+ }
+
+ @Test
+ public void stopAllRoutesDrainsTheSplit() throws Exception {
+ context.getRouteController().stopAllRoutes();
+
+ assertEquals(List.of("A", "B", "C", "D"), shippedBodies());
+ assertEquals(ServiceStatus.Stopped,
context.getRouteController().getRouteStatus("picked"));
+ assertEquals(ServiceStatus.Stopped,
context.getRouteController().getRouteStatus("shipment"));
+ }
+
+ private List<Object> shippedBodies() {
+ return shipped.getReceivedExchanges().stream().map(e ->
e.getMessage().getBody()).toList();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("seda:start").routeId("picked")
+ .split(body())
+ .delay(300)
+ .to("direct:shipment")
+ .end();
+ from("direct:shipment").routeId("shipment")
+ .to("mock:shipped");
+ }
+ };
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
index 003caaa164bb..8195e780d257 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
@@ -1550,6 +1550,16 @@ reloaded", as nothing in the route failed, and a
consumer that rolls back, such
A level lower than WARN configured for exhausted logging is kept. A forced
stop of the CamelContext is logged as
before. The exception set on the exchange is unchanged.
+=== camel-core - stopping or removing all routes is one graceful shutdown
+
+`RouteController.stopAllRoutes()` and `removeAllRoutes()` (which a route
reload in dev mode uses) stopped the routes
+one at a time, each as its own graceful shutdown. A route that consumes from
another route, such as over `direct`,
+was then stopped as soon as its own exchanges had completed, while the route
sending to it could still have work for
+it (a split, a loop). Those exchanges waited for the stopped route until the
shutdown timeout, and were cut off. All
+the routes are now stopped as one graceful shutdown, as when the CamelContext
stops, so a route consuming from
+another keeps running until the routes sending to it have completed. The
shutdown timeout now applies to all the
+routes together, instead of to each route in turn.
+
=== camel-core - RoutesLoader.updateRoutes runs one call at a time
Concurrent `RoutesLoader.updateRoutes` calls on one CamelContext now run one
at a time instead of failing with
