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 c69301b930d CAMEL-21793: camel-kamelet - Creating dynamic kamelets
should not happen concurrently (use lock)
c69301b930d is described below
commit c69301b930d94634f92b6fce86169bf0e0d56e63
Author: Claus Ibsen <[email protected]>
AuthorDate: Thu Feb 27 10:27:13 2025 +0100
CAMEL-21793: camel-kamelet - Creating dynamic kamelets should not happen
concurrently (use lock)
---
.../camel/component/kamelet/KameletComponent.java | 14 ++++
.../kamelet/KameletConcurrencyIssueTest.java | 86 ++++++++++++++++++++++
.../kamelet/KameletMultiThreadedTest.java | 70 ------------------
3 files changed, 100 insertions(+), 70 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 af5265df24b..ea7afb02747 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
@@ -64,6 +64,8 @@ public class KameletComponent extends DefaultComponent {
private static final Logger LOG =
LoggerFactory.getLogger(KameletComponent.class);
+ private final ReentrantLock lock = new ReentrantLock();
+
private final LifecycleHandler lifecycleHandler = new LifecycleHandler();
// active consumers
@@ -454,6 +456,18 @@ public class KameletComponent extends DefaultComponent {
public void createRouteForEndpoint(KameletEndpoint endpoint, String
parentRouteId, String parentProcessorId)
throws Exception {
+ // creating dynamic routes from kamelets should not happen
concurrently so we use locking
+ lock.lock();
+ try {
+ doCreateRouteForEndpoint(endpoint, parentRouteId,
parentProcessorId);
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ protected void doCreateRouteForEndpoint(KameletEndpoint endpoint,
String parentRouteId, String parentProcessorId)
+ throws Exception {
+
final ModelCamelContext context = (ModelCamelContext)
getCamelContext();
final String templateId = endpoint.getTemplateId();
final String routeId = endpoint.getRouteId();
diff --git
a/components/camel-kamelet/src/test/java/org/apache/camel/component/kamelet/KameletConcurrencyIssueTest.java
b/components/camel-kamelet/src/test/java/org/apache/camel/component/kamelet/KameletConcurrencyIssueTest.java
new file mode 100644
index 00000000000..fff35e69492
--- /dev/null
+++
b/components/camel-kamelet/src/test/java/org/apache/camel/component/kamelet/KameletConcurrencyIssueTest.java
@@ -0,0 +1,86 @@
+/*
+ * 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.util.ArrayList;
+import java.util.List;
+
+import org.apache.camel.LoggingLevel;
+import org.apache.camel.RoutesBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.test.junit5.CamelTestSupport;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+
+@Disabled("Manual test")
+public class KameletConcurrencyIssueTest extends CamelTestSupport {
+
+ @Test
+ public void testConcurrency() throws Exception {
+ // check there are no exception throw during creating kamelets
+ Thread.sleep(120000);
+ }
+
+ @Override
+ protected RoutesBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ List<Integer> splitSpam = new ArrayList<>();
+ for (int i = 0; i < 500; i++) {
+ splitSpam.add(i);
+ }
+
+ from("timer://spamroute?fixedRate=true&period=5")
+ .routeId("BreakingRoute")
+ .process(exchange -> {
+ exchange.getIn().setBody(splitSpam);
+ exchange.setVariable("numberMs",
System.currentTimeMillis() % 1000);
+ })
+ .split(body())
+ .parallelProcessing()
+ .process(exchange -> {
+ exchange.setVariable("number",
exchange.getIn().getBody(Integer.class));
+ })
+
.toD("kamelet:spamRoute?number=${variable.number}&differentValue=${variable.numberMs}")
+ .end()
+ .log(LoggingLevel.INFO, "Ending");
+
+ routeTemplate("spamRoute")
+ .templateParameter("number", "0")
+ .templateParameter("differentValue", "0")
+ .from("kamelet:source")
+ .routeId("spamRoute")
+
.toD("kamelet:spamRoute2?numberAgain={{number}}&differentValueAgain={{differentValue}}&time="
+ + System.currentTimeMillis() % 1000)
+
.toD("kamelet:spamRoute2?numberAgain={{number}}&differentValueAgain=5&time="
+ + System.currentTimeMillis() % 1000)
+
.toD("kamelet:spamRoute2?numberAgain={{number}}&differentValueAgain=10&time="
+ + System.currentTimeMillis() % 1000);
+
+ routeTemplate("spamRoute2")
+ .templateParameter("numberAgain")
+ .templateParameter("differentValueAgain")
+ .templateParameter("time")
+ .from("kamelet:source")
+ .routeId("spamRoute2")
+ .log("Spam route {{numberAgain}}
{{differentValueAgain}} {{time}}")
+ .toD("log:valuespam");
+ }
+ };
+ }
+}
diff --git
a/components/camel-kamelet/src/test/java/org/apache/camel/component/kamelet/KameletMultiThreadedTest.java
b/components/camel-kamelet/src/test/java/org/apache/camel/component/kamelet/KameletMultiThreadedTest.java
deleted file mode 100644
index 73d47e2cd6f..00000000000
---
a/components/camel-kamelet/src/test/java/org/apache/camel/component/kamelet/KameletMultiThreadedTest.java
+++ /dev/null
@@ -1,70 +0,0 @@
-/*
- * 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.util.concurrent.CountDownLatch;
-
-import org.apache.camel.RoutesBuilder;
-import org.apache.camel.builder.RouteBuilder;
-import org.apache.camel.component.mock.MockEndpoint;
-import org.apache.camel.test.junit5.CamelTestSupport;
-import org.junit.jupiter.api.Test;
-
-import static org.apache.camel.component.kamelet.Kamelet.templateToRoute;
-
-public class KameletMultiThreadedTest extends CamelTestSupport {
-
- @Test
- public void
createSameKameletTwiceInParallel_KameletConsumerNotAvailableExceptionThrown()
throws InterruptedException {
- var latch = new CountDownLatch(2);
- context.addRouteTemplateDefinitionConverter("*", (in, parameters) -> {
- try {
- return templateToRoute(in, parameters);
- } finally {
- latch.countDown();
- latch.await();
- }
- });
- getMockEndpoint("mock:foo").expectedMessageCount(2);
-
- template.sendBody("seda:route", null);
- template.requestBody("seda:route", ((Object) null));
-
- MockEndpoint.assertIsSatisfied(context);
- }
-
- // **********************************************
- //
- // test set-up
- //
- // **********************************************
-
- @Override
- protected RoutesBuilder createRouteBuilder() {
- return new RouteBuilder() {
- @Override
- public void configure() {
- from("seda:route?concurrentConsumers=2")
- .toD("kamelet:-");
-
- routeTemplate("-"). // This is a workaround for "*" to be
iterated before templateId at
org.apache.camel.impl.DefaultModel#addRouteFromTemplate (line 460)
- from("kamelet:source")
- .to("mock:foo");
- }
- };
- }
-}
