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 bdec091863e4 CAMEL-25093: camel-core - Run concurrent updateRoutes
calls one at a time (#27524)
bdec091863e4 is described below
commit bdec091863e46b89609c09a13be6562147944d29
Author: allthingssecurity <[email protected]>
AuthorDate: Thu Oct 8 18:36:26 2026 +0530
CAMEL-25093: camel-core - Run concurrent updateRoutes calls one at a time
(#27524)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../java/org/apache/camel/spi/RoutesLoader.java | 6 +
.../camel/impl/engine/DefaultRoutesLoader.java | 19 ++-
.../ROOT/pages/camel-4x-upgrade-guide-4_23.adoc | 6 +
.../dsl/xml/io/XmlConcurrentUpdateRoutesTest.java | 137 +++++++++++++++++++++
4 files changed, 166 insertions(+), 2 deletions(-)
diff --git
a/core/camel-api/src/main/java/org/apache/camel/spi/RoutesLoader.java
b/core/camel-api/src/main/java/org/apache/camel/spi/RoutesLoader.java
index 743a7e219bf9..9e023931409e 100644
--- a/core/camel-api/src/main/java/org/apache/camel/spi/RoutesLoader.java
+++ b/core/camel-api/src/main/java/org/apache/camel/spi/RoutesLoader.java
@@ -110,6 +110,8 @@ public interface RoutesLoader extends CamelContextAware {
* If a route is loaded with a route id for an existing route, then the
existing route is stopped and remove, so it
* can be updated.
*
+ * Camel's {@code DefaultRoutesLoader} runs concurrent calls one at a
time, see {@link #updateRoutes(Collection)}.
+ *
* @param resources the resources to be loaded or updated.
* @return route ids for the routes that was loaded or updated.
*/
@@ -124,6 +126,10 @@ public interface RoutesLoader extends CamelContextAware {
* If a route is loaded with a route id for an existing route, then the
existing route is stopped and remove, so it
* can be updated.
*
+ * Camel's {@code DefaultRoutesLoader} runs concurrent calls of the {@code
updateRoutes} methods one at a time, so
+ * they can be called from several threads. They are not serialized with
other ways of adding routes, such as
+ * {@link #loadRoutes(Resource...)}.
+ *
* @param resources the resources to be loaded or updated.
* @return route ids for the routes that was loaded or updated.
*/
diff --git
a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultRoutesLoader.java
b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultRoutesLoader.java
index fc462f70087a..2b523d3a6528 100644
---
a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultRoutesLoader.java
+++
b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/DefaultRoutesLoader.java
@@ -24,6 +24,8 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
import org.apache.camel.CamelContext;
import org.apache.camel.CamelContextAware;
@@ -59,6 +61,9 @@ public class DefaultRoutesLoader extends ServiceSupport
implements RoutesLoader,
public static final String ROUTES_LOADER_KEY_PREFIX =
"routes-builder-loader-";
private final Map<String, RoutesBuilderLoader> loaders;
+ // the loaders keep state between pre-parsing and loading the resources
(such as beans that are registered again
+ // when the routes are configured), and the model is not updated
atomically, so updates run one at a time
+ private final Lock updateLock = new ReentrantLock();
private CamelContext camelContext;
private boolean ignoreLoadingError;
@@ -292,6 +297,18 @@ public class DefaultRoutesLoader extends ServiceSupport
implements RoutesLoader,
return answer;
}
+ // the lock is held while the routes being replaced are stopped: an
updateRoutes call made from an exchange of
+ // such a route therefore waits here until the shutdown timeout forces
that route to stop (see the upgrade guide)
+ updateLock.lockInterruptibly();
+ try {
+ doUpdateRoutes(resources, answer);
+ } finally {
+ updateLock.unlock();
+ }
+ return answer;
+ }
+
+ private void doUpdateRoutes(Collection<Resource> resources, Set<String>
answer) throws Exception {
Collection<RoutesBuilder> builders = findRoutesBuilders(resources);
for (RoutesBuilder builder : builders) {
// update any existing route configurations first
@@ -312,8 +329,6 @@ public class DefaultRoutesLoader extends ServiceSupport
implements RoutesLoader,
}
answer.addAll(ids);
}
-
- return answer;
}
protected RoutesBuilderLoader resolveRoutesBuilderLoader(Resource
resource, boolean optional) throws Exception {
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 034bcf893831..4ed981d11a4f 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
@@ -1484,6 +1484,12 @@ A project whose routes are all in one file is covered
too: the content that last
broken save goes back to the version that was running, and the log says how
many routes came from it. The file on
disk is untouched; its next save is loaded as usual.
+=== 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
+`ConcurrentModificationException`. A route whose exchange calls `updateRoutes`
while another `updateRoutes` call
+replaces that route now makes the stop wait until the shutdown timeout, after
which the route is forced to stop.
+
=== camel-core - the recursive file watcher watches directories created while
it runs
The file watcher reload strategy (`camel run --dev`, the route watcher reload
strategy with a recursive directory)
diff --git
a/dsl/camel-xml-io-dsl/src/test/java/org/apache/camel/dsl/xml/io/XmlConcurrentUpdateRoutesTest.java
b/dsl/camel-xml-io-dsl/src/test/java/org/apache/camel/dsl/xml/io/XmlConcurrentUpdateRoutesTest.java
new file mode 100644
index 000000000000..7fd1f4408d73
--- /dev/null
+++
b/dsl/camel-xml-io-dsl/src/test/java/org/apache/camel/dsl/xml/io/XmlConcurrentUpdateRoutesTest.java
@@ -0,0 +1,137 @@
+/*
+ * 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.dsl.xml.io;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.spi.Resource;
+import org.apache.camel.spi.RoutesLoader;
+import org.apache.camel.support.PluginHelper;
+import org.apache.camel.support.ResourceHelper;
+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.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Calling updateRoutes from several threads on the same CamelContext
(CAMEL-25093).
+ */
+class XmlConcurrentUpdateRoutesTest {
+
+ private static final AtomicInteger GATED_CALLS = new AtomicInteger();
+ private static final AtomicInteger OTHER_CALLS = new AtomicInteger();
+ private static volatile CountDownLatch gatedEntered;
+ private static volatile CountDownLatch gatedRelease;
+
+ @BeforeEach
+ void reset() {
+ GATED_CALLS.set(0);
+ OTHER_CALLS.set(0);
+ gatedEntered = new CountDownLatch(1);
+ gatedRelease = new CountDownLatch(1);
+ }
+
+ /**
+ * Fails the first time (so it is registered again when the routes are
configured), and the second time waits for
+ * the test.
+ */
+ public static class GatedBean {
+ public GatedBean() throws InterruptedException {
+ int call = GATED_CALLS.incrementAndGet();
+ if (call == 1) {
+ throw new IllegalStateException("Not yet");
+ } else if (call == 2) {
+ gatedEntered.countDown();
+ gatedRelease.await(20, TimeUnit.SECONDS);
+ }
+ }
+ }
+
+ /**
+ * Fails the first time (so it is registered again when the routes are
configured).
+ */
+ public static class OtherBean {
+ public OtherBean() {
+ if (OTHER_CALLS.incrementAndGet() == 1) {
+ throw new IllegalStateException("Not yet");
+ }
+ }
+ }
+
+ @Test
+ void testConcurrentUpdateRoutes() throws Exception {
+ Resource first = ResourceHelper.fromString("first.xml", camel("gated",
GatedBean.class, "first"));
+ Resource second = ResourceHelper.fromString("second.xml",
camel("other", OtherBean.class, "second"));
+
+ try (CamelContext context = new DefaultCamelContext()) {
+ context.start();
+ RoutesLoader loader = PluginHelper.getRoutesLoader(context);
+
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+ try {
+ // the first update waits while it registers the beans that
failed when the resource was pre-parsed
+ Future<?> one = executor.submit(() ->
loader.updateRoutes(first));
+ assertTrue(gatedEntered.await(20, TimeUnit.SECONDS));
+
+ // the second update runs meanwhile (or waits for the first)
+ AtomicReference<Thread> thread = new AtomicReference<>();
+ Future<?> two = executor.submit(() -> {
+ thread.set(Thread.currentThread());
+ return loader.updateRoutes(second);
+ });
+ await().atMost(20, TimeUnit.SECONDS).until(() -> two.isDone()
+ || OTHER_CALLS.get() == 0 && thread.get() != null
+ && thread.get().getState() ==
Thread.State.WAITING);
+ gatedRelease.countDown();
+
+ assertDoesNotThrow(() -> one.get(20, TimeUnit.SECONDS), "The
first updateRoutes should not fail");
+ assertDoesNotThrow(() -> two.get(20, TimeUnit.SECONDS), "The
second updateRoutes should not fail");
+ } finally {
+ gatedRelease.countDown();
+ executor.shutdownNow();
+ }
+
+ assertNotNull(context.getRoute("first"), "Route first should be
added");
+ assertNotNull(context.getRoute("second"), "Route second should be
added");
+ assertNotNull(context.getRegistry().lookupByName("gated"), "Bean
gated should be registered");
+ assertNotNull(context.getRegistry().lookupByName("other"), "Bean
other should be registered");
+ }
+ }
+
+ private static String camel(String bean, Class<?> type, String route) {
+ return """
+ <camel>
+ <bean name="%s" type="%s"/>
+ <route id="%s">
+ <from uri="direct:%s"/>
+ <to uri="mock:%s"/>
+ </route>
+ </camel>
+ """.formatted(bean, type.getName(), route, route, route);
+ }
+}
