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 ae7355f5de27 CAMEL-26535: camel-kamelet - fix deadlock between
addRoutes and a toD creating a Kamelet route (#29119)
ae7355f5de27 is described below
commit ae7355f5de2734558953d1aca263d65f882072ca
Author: allthingssecurity <[email protected]>
AuthorDate: Wed Nov 4 13:44:07 2026 +0530
CAMEL-26535: camel-kamelet - fix deadlock between addRoutes and a toD
creating a Kamelet route (#29119)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../camel/component/kamelet/KameletComponent.java | 6 +-
.../kamelet/KameletAddRoutesDeadlockTest.java | 203 +++++++++++++++++++++
.../org/apache/camel/impl/DefaultCamelContext.java | 6 +
.../java/org/apache/camel/impl/DefaultModel.java | 5 +
.../main/java/org/apache/camel/model/Model.java | 13 ++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 7 +
6 files changed, 237 insertions(+), 3 deletions(-)
diff --git
a/components/camel-kamelet/src/main/java/org/apache/camel/component/kamelet/KameletComponent.java
b/components/camel-kamelet/src/main/java/org/apache/camel/component/kamelet/KameletComponent.java
index c871d2e3f1a5..11f77ee8f520 100644
---
a/components/camel-kamelet/src/main/java/org/apache/camel/component/kamelet/KameletComponent.java
+++
b/components/camel-kamelet/src/main/java/org/apache/camel/component/kamelet/KameletComponent.java
@@ -82,8 +82,6 @@ public class KameletComponent extends DefaultComponent {
// how long the removal of a route waits before it checks again whether
the thread pools of the route are idle
private static final long ROUTE_REMOVAL_RETRY_MILLIS = 100;
- private final ReentrantLock lock = new ReentrantLock();
-
private final LifecycleHandler lifecycleHandler = new LifecycleHandler();
// stops and removes the routes of Kamelet endpoints that are shut down,
off the threads that route the messages
@@ -683,7 +681,9 @@ public class KameletComponent extends DefaultComponent {
public String createRouteForEndpoint(KameletEndpoint endpoint, String
parentRouteId, String parentProcessorId)
throws Exception {
- // creating dynamic routes from kamelets should not happen
concurrently so we use locking
+ // creating dynamic routes from kamelets should not happen
concurrently, and we use the model lock
+ // (taken first) to avoid deadlock with concurrent 'addRoutes'
calls that create kamelet routes
+ Lock lock = ((ModelCamelContext) getCamelContext()).getModelLock();
lock.lock();
try {
return doCreateRouteForEndpoint(endpoint, parentRouteId,
parentProcessorId);
diff --git
a/components/camel-kamelet/src/test/java/org/apache/camel/component/kamelet/KameletAddRoutesDeadlockTest.java
b/components/camel-kamelet/src/test/java/org/apache/camel/component/kamelet/KameletAddRoutesDeadlockTest.java
new file mode 100644
index 000000000000..744144efa958
--- /dev/null
+++
b/components/camel-kamelet/src/test/java/org/apache/camel/component/kamelet/KameletAddRoutesDeadlockTest.java
@@ -0,0 +1,203 @@
+/*
+ * 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.component.kamelet;
+
+import java.lang.management.ManagementFactory;
+import java.lang.management.ThreadInfo;
+import java.lang.management.ThreadMXBean;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.LockSupport;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.stream.Collectors;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.model.ModelCamelContext;
+import org.apache.camel.model.ModelLifecycleStrategySupport;
+import org.apache.camel.model.RouteDefinition;
+import org.apache.camel.model.RouteTemplateDefinition;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.fail;
+import static org.awaitility.Awaitility.await;
+
+/**
+ * Adding a route that uses a Kamelet while the CamelContext runs, at the same
time as an exchange creates the route of
+ * another Kamelet (toD), must not deadlock.
+ * <p/>
+ * Adding the routes holds the lock of the model while the new route is
created, and creating the route of its Kamelet
+ * takes the lock of the Kamelet component. The toD creates the route of its
Kamelet holding the lock of the Kamelet
+ * component, and adding that route takes the lock of the model.
+ */
+class KameletAddRoutesDeadlockTest {
+
+ private CamelContext context;
+ private boolean hung;
+
+ @AfterEach
+ void stopContext() {
+ // the threads of a deadlock never release the locks, so the context
cannot be stopped
+ if (context != null && !hung) {
+ context.stop();
+ }
+ }
+
+ @Test
+ void addRoutesWhileToDCreatesTheRouteOfAKamelet() throws Exception {
+ CountDownLatch addHoldsModelLock = new CountDownLatch(1);
+ CountDownLatch toDHoldsKameletLock = new CountDownLatch(1);
+ AtomicReference<Thread> toDThread = new AtomicReference<>();
+
+ context = createContext();
+ ModelCamelContext mcc = (ModelCamelContext) context;
+ // runs while adding the route "added", holding the lock of the model:
wait until the toD holds the lock of
+ // the Kamelet component (or waits for a lock)
+ mcc.addModelLifecycleStrategy(new ModelLifecycleStrategySupport() {
+ @Override
+ public void onAddRouteDefinition(RouteDefinition routeDefinition) {
+ if ("added".equals(routeDefinition.getRouteId())) {
+ addHoldsModelLock.countDown();
+ awaitOrBlocked(toDHoldsKameletLock, toDThread);
+ }
+ }
+ });
+ // runs while the toD creates the route of Kamelet b, holding the lock
of the Kamelet component
+ mcc.addRouteTemplateDefinitionConverter("b", (in, parameters) -> {
+ toDHoldsKameletLock.countDown();
+ return
RouteTemplateDefinition.Converter.DEFAULT_CONVERTER.apply(in, parameters);
+ });
+ context.start();
+
+ FutureTask<Void> add = new FutureTask<>(() -> {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:added").routeId("added").to("kamelet:a");
+ }
+ });
+ return null;
+ });
+ Thread addThread = new Thread(add, "add-routes");
+ addThread.setDaemon(true);
+ addThread.start();
+ assertThat(addHoldsModelLock.await(10, TimeUnit.SECONDS)).as("the
route is being added").isTrue();
+
+ ProducerTemplate template = context.createProducerTemplate();
+ FutureTask<Object> toD = new FutureTask<>(() ->
template.requestBodyAndHeader("direct:dynamic", "Hello", "x", "1"));
+ Thread thread = new Thread(toD, "toD-kamelet");
+ thread.setDaemon(true);
+ toDThread.set(thread);
+ thread.start();
+
+ try {
+ add.get(20, TimeUnit.SECONDS);
+ assertThat(toD.get(20, TimeUnit.SECONDS)).isEqualTo("b-1-Hello");
+ } catch (TimeoutException e) {
+ hung = true;
+ ThreadMXBean threads = ManagementFactory.getThreadMXBean();
+ long[] ids = threads.findDeadlockedThreads();
+ List<ThreadInfo> deadlocked = ids == null
+ ? List.of()
+ : Arrays.stream(threads.getThreadInfo(ids))
+ .filter(i -> i.getThreadId() ==
addThread.threadId() || i.getThreadId() == thread.threadId())
+ .toList();
+ assertThat(deadlocked).as("deadlocked: %s",
describe(deadlocked)).isEmpty();
+ fail("add-routes or toD-kamelet did not complete");
+ }
+
+ MockEndpoint result = context.getEndpoint("mock:result",
MockEndpoint.class);
+ result.expectedBodiesReceived("a-Hello");
+ template.sendBody("direct:added", "Hello");
+ assertThat(template.requestBodyAndHeader("direct:dynamic", "Hello",
"x", "2")).isEqualTo("b-2-Hello");
+ MockEndpoint.assertIsSatisfied(context);
+ }
+
+ /**
+ * Control: the same two operations one after the other.
+ */
+ @Test
+ void addRoutesAfterToDCreatedTheRouteOfAKamelet() throws Exception {
+ context = createContext();
+ context.start();
+
+ ProducerTemplate template = context.createProducerTemplate();
+ assertThat(template.requestBodyAndHeader("direct:dynamic", "Hello",
"x", "1")).isEqualTo("b-1-Hello");
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:added").routeId("added").to("kamelet:a");
+ }
+ });
+
+ MockEndpoint result = context.getEndpoint("mock:result",
MockEndpoint.class);
+ result.expectedBodiesReceived("a-Hello");
+ template.sendBody("direct:added", "Hello");
+ MockEndpoint.assertIsSatisfied(context);
+ }
+
+ private static CamelContext createContext() throws Exception {
+ CamelContext context = new DefaultCamelContext();
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ routeTemplate("a")
+ .from("kamelet:source")
+ .setBody().simple("a-${body}")
+ .to("mock:result");
+
+ routeTemplate("b")
+ .templateParameter("x")
+ .from("kamelet:source")
+ .setBody().simple("b-{{x}}-${body}");
+
+ from("direct:dynamic").routeId("dynamic")
+ .toD("kamelet:b?x=${header.x}");
+ }
+ });
+ return context;
+ }
+
+ private static String describe(List<ThreadInfo> infos) {
+ return infos.stream()
+ .map(i -> i.getThreadName() + " waits for " + i.getLockName()
+ " held by " + i.getLockOwnerName())
+ .collect(Collectors.joining("; "));
+ }
+
+ /**
+ * Waits until the latch is counted down, or the thread waits for a {@link
ReentrantLock} (as the toD does when it
+ * takes the lock of the model first). Fails the adding of the route (and
so the test) if neither happens.
+ */
+ private static void awaitOrBlocked(CountDownLatch latch,
AtomicReference<Thread> thread) {
+ await().atMost(10, TimeUnit.SECONDS).until(() -> latch.getCount() == 0
|| waitsForLock(thread.get()));
+ }
+
+ private static boolean waitsForLock(Thread thread) {
+ Object blocker = thread != null ? LockSupport.getBlocker(thread) :
null;
+ return blocker != null &&
blocker.getClass().getName().startsWith(ReentrantLock.class.getName());
+ }
+}
diff --git
a/core/camel-core-engine/src/main/java/org/apache/camel/impl/DefaultCamelContext.java
b/core/camel-core-engine/src/main/java/org/apache/camel/impl/DefaultCamelContext.java
index 0a2f8fab2927..16036421cefc 100644
---
a/core/camel-core-engine/src/main/java/org/apache/camel/impl/DefaultCamelContext.java
+++
b/core/camel-core-engine/src/main/java/org/apache/camel/impl/DefaultCamelContext.java
@@ -24,6 +24,7 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
+import java.util.concurrent.locks.Lock;
import java.util.function.Function;
import org.apache.camel.CamelContext;
@@ -233,6 +234,11 @@ public class DefaultCamelContext extends
SimpleCamelContext implements ModelCame
return resolver;
}
+ @Override
+ public Lock getModelLock() {
+ return model.getModelLock();
+ }
+
@Override
public void addModelLifecycleStrategy(ModelLifecycleStrategy
modelLifecycleStrategy) {
model.addModelLifecycleStrategy(modelLifecycleStrategy);
diff --git
a/core/camel-core-engine/src/main/java/org/apache/camel/impl/DefaultModel.java
b/core/camel-core-engine/src/main/java/org/apache/camel/impl/DefaultModel.java
index b1849c57b4b4..1f84b2491194 100644
---
a/core/camel-core-engine/src/main/java/org/apache/camel/impl/DefaultModel.java
+++
b/core/camel-core-engine/src/main/java/org/apache/camel/impl/DefaultModel.java
@@ -108,6 +108,11 @@ public class DefaultModel implements Model {
return lock;
}
+ @Override
+ public Lock getModelLock() {
+ return lock;
+ }
+
@Override
public void addModelLifecycleStrategy(ModelLifecycleStrategy
modelLifecycleStrategy) {
// avoid adding double which can happen with spring xml on spring boot
diff --git
a/core/camel-core-model/src/main/java/org/apache/camel/model/Model.java
b/core/camel-core-model/src/main/java/org/apache/camel/model/Model.java
index 1f7aab42cfe3..669a093b0594 100644
--- a/core/camel-core-model/src/main/java/org/apache/camel/model/Model.java
+++ b/core/camel-core-model/src/main/java/org/apache/camel/model/Model.java
@@ -19,6 +19,7 @@ package org.apache.camel.model;
import java.util.Collection;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.locks.Lock;
import java.util.function.Function;
import org.apache.camel.CamelContext;
@@ -497,4 +498,16 @@ public interface Model {
*/
List<BeanFactoryDefinition<?>> getCustomBeans();
+ /**
+ * Gets the lock that is held while route definitions are added to or
removed from this model (including creating
+ * and starting the added routes).
+ * <p/>
+ * Code that adds routes while it holds a lock of its own must take this
lock first, to avoid a deadlock with routes
+ * being added at the same time.
+ *
+ * @return the lock
+ * @since 4.24
+ */
+ Lock getModelLock();
+
}
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 dda0dccee5f1..6ce3498cfcdb 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
@@ -127,6 +127,13 @@ sibling package whose name merely started with the scanned
package name, such as
and its sub packages, so a `RoutesBuilder` or `@Configuration` class in such a
sibling package is no longer
discovered when running from a JAR; list that package explicitly if it should
be scanned.
+=== camel-core - Model has a new getModelLock method
+
+The `org.apache.camel.model.Model` interface (and so `ModelCamelContext`) has
a new method `getModelLock()`, which
+returns the lock held while routes are added to or removed from the model. A
custom implementation of `Model` must
+implement it (classes that extend `DefaultCamelContext` need no change). The
Kamelet component now creates the routes of
+Kamelets holding this lock.
+
=== camel-core - Route.clearRouteModel() is deprecated
The `Route.clearRouteModel()` method is deprecated and is now a no-op. It will
be removed in a later release.
