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 481428816a13 CAMEL-25839: camel-kamelet - remove the route of a
Kamelet endpoint that toD discards after use (#28260)
481428816a13 is described below
commit 481428816a13f3446ebb9052189f945fb69fd721
Author: allthingssecurity <[email protected]>
AuthorDate: Fri Oct 23 10:54:40 2026 +0530
CAMEL-25839: camel-kamelet - remove the route of a Kamelet endpoint that
toD discards after use (#28260)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../camel/component/kamelet/KameletComponent.java | 203 ++++++++++++-
.../camel/component/kamelet/KameletEndpoint.java | 11 +
.../camel/component/kamelet/KameletProducer.java | 4 +-
.../kamelet/KameletToDRoutesNotKeptTest.java | 325 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 11 +
5 files changed, 545 insertions(+), 9 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 580b85b105b9..c871d2e3f1a5 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
@@ -26,15 +26,22 @@ import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
+import org.apache.camel.AsyncCallback;
import org.apache.camel.CamelContext;
import org.apache.camel.Endpoint;
+import org.apache.camel.Exchange;
import org.apache.camel.Processor;
+import org.apache.camel.Route;
import org.apache.camel.RuntimeCamelException;
import org.apache.camel.ServiceStatus;
import org.apache.camel.VetoCamelContextStartException;
@@ -46,6 +53,7 @@ import org.apache.camel.spi.annotations.Component;
import org.apache.camel.support.DefaultComponent;
import org.apache.camel.support.LifecycleStrategySupport;
import org.apache.camel.support.RouteTemplateHelper;
+import org.apache.camel.support.SynchronizationAdapter;
import org.apache.camel.support.service.ServiceHelper;
import org.apache.camel.util.StopWatch;
import org.apache.camel.util.URISupport;
@@ -71,10 +79,18 @@ public class KameletComponent extends DefaultComponent {
private static final Logger LOG =
LoggerFactory.getLogger(KameletComponent.class);
+ // 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
+ private volatile ScheduledExecutorService routeRemover;
+ // the thread pools that the processors of each route created (such as
threads()), by route id
+ private final Map<String, List<ThreadPoolExecutor>> routeThreadPools = new
ConcurrentHashMap<>();
+
// active consumers
private final Map<String, KameletConsumer> consumers = new HashMap<>();
private final Lock consumersLock = new ReentrantLock();
@@ -137,6 +153,8 @@ public class KameletComponent extends DefaultComponent {
final String uuid = Kamelet.extractUuid();
final String routeId = Kamelet.extractRouteId(getCamelContext(),
remaining, parameters, uuid);
final String loc = Kamelet.extractLocation(getCamelContext(),
parameters);
+ // a route id that is not given in the uri is unique to this endpoint
+ final boolean ownRouteId = routeId.equals(templateId + "-" + uuid);
parameters.remove(PARAM_TEMPLATE_ID);
parameters.remove(PARAM_ROUTE_ID);
@@ -188,6 +206,12 @@ public class KameletComponent extends DefaultComponent {
setProperties(endpoint, parameters);
} else {
endpoint = new KameletEndpoint(uri, this, templateId, routeId) {
+ // the route this endpoint created from the template while the
context runs (such as from toD), with
+ // a route id of its own, which is removed when this endpoint
is shut down
+ private volatile Route createdRoute;
+ // the exchange whose call of this endpoint is completing on
the thread that runs the callback
+ private volatile Exchange completing;
+
@Override
protected void doInit() throws Exception {
super.doInit();
@@ -197,7 +221,42 @@ public class KameletComponent extends DefaultComponent {
//
String routeId =
getCamelContext().getCamelContextExtension().getCreateRoute();
String processorId =
getCamelContext().getCamelContextExtension().getCreateProcessor();
- lifecycleHandler.track(this, routeId, processorId);
+ String id = lifecycleHandler.track(this, routeId,
processorId);
+ if (id != null && ownRouteId) {
+ // a route id given in the uri is shared by all the
endpoints that use it, and each of them
+ // replaces the route of the previous one: such a
route is left alone
+ createdRoute = getCamelContext().getRoute(id);
+ }
+ }
+
+ @Override
+ AsyncCallback completingCallback(Exchange exchange,
AsyncCallback callback) {
+ if (createdRoute == null) {
+ return callback;
+ }
+ return doneSync -> {
+ completing = exchange;
+ try {
+ callback.done(doneSync);
+ } finally {
+ completing = null;
+ }
+ };
+ }
+
+ @Override
+ protected void doShutdown() throws Exception {
+ // An endpoint used once and discarded (toD or
recipientList with cacheSize=-1) is shut down after
+ // use: remove the route it created, otherwise every call
keeps one more route for ever.
+ // The endpoint of a route that consumes from a Kamelet
(from("kamelet:my-source")) is shut down
+ // as well when that route is removed: when that route was
added while the context runs, the
+ // Kamelet's route is removed with it.
+ Route route = createdRoute;
+ createdRoute = null;
+ if (route != null) {
+ removeCreatedRoute(route, completing);
+ }
+ super.doShutdown();
}
};
@@ -484,6 +543,106 @@ public class KameletComponent extends DefaultComponent {
super.doInit();
}
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+ ScheduledExecutorService executor = routeRemover;
+ if (executor == null || executor.isShutdown()) {
+ // the thread is created on first use
+ routeRemover =
getCamelContext().getExecutorServiceManager().newSingleThreadScheduledExecutor(this,
+ "KameletRouteRemover");
+ }
+ }
+
+ /**
+ * Removes the route a Kamelet endpoint created from its template while
the context runs, when that endpoint is shut
+ * down (it is not shut down while the context runs unless it was used
once and discarded, or its route consumed
+ * from the Kamelet and was removed).
+ * <p/>
+ * The route is stopped and removed on a thread of this component:
stopping a route holds the lock of the
+ * CamelContext while the route shuts down gracefully. The calling route
can continue on a thread that the Kamelet's
+ * route owns (threads(), an asynchronous producer), and stopping the
route shuts that thread pool down, which
+ * interrupts that thread. So the removal waits:
+ * <ul>
+ * <li>until the exchange whose call is completing on this thread is done,
when the endpoint is shut down inside the
+ * producer callback (toD);</li>
+ * <li>and until no thread of the thread pools that the route's processors
created (threads(), and the other EIPs
+ * with an executor service) runs a task. This covers an endpoint that is
shut down where the exchange of the
+ * calling route is not known: recipientList with cacheSize=-1 (the
endpoint is shut down when the producer is
+ * released, after the call), routingSlip with cacheSize=-1 (its copy of
the exchange has no unit of work) and
+ * pollEnrich. The route is used only by the endpoint that created it, so
its thread pools stay idle once the
+ * calling route has left them.</li>
+ * </ul>
+ * Not covered: a thread pool that a component of the route creates for
itself (such as the worker pool of an
+ * asynchronous Kafka producer, which waits for its running tasks when it
stops, up to its shutdown timeout), and an
+ * executor service that is not a {@link ThreadPoolExecutor} (virtual
threads), which the removal cannot tell idle.
+ *
+ * @param route the route the endpoint created
+ * @param exchange the exchange whose call of the endpoint is completing
on this thread, or {@code null}
+ */
+ private void removeCreatedRoute(Route route, Exchange exchange) {
+ if (exchange != null && exchange.getUnitOfWork() != null) {
+ exchange.getExchangeExtension().addOnCompletion(new
SynchronizationAdapter() {
+ @Override
+ public void onDone(Exchange done) {
+ submitRemoveRoute(route, 0);
+ }
+ });
+ } else {
+ submitRemoveRoute(route, 0);
+ }
+ }
+
+ private void submitRemoveRoute(Route route, long delayMillis) {
+ ScheduledExecutorService executor = routeRemover;
+ if (executor == null) {
+ // the context removes all the routes when it stops
+ return;
+ }
+ try {
+ executor.schedule(() -> doRemoveRoute(route), delayMillis,
TimeUnit.MILLISECONDS);
+ } catch (RejectedExecutionException e) {
+ // the component is shut down, and the context removes all the
routes when it stops
+ LOG.debug("Cannot remove route with id={} as the component is shut
down", route.getRouteId());
+ }
+ }
+
+ private void doRemoveRoute(Route route) {
+ CamelContext context = getCamelContext();
+ String routeId = route.getRouteId();
+ if (!context.isStarted() || context.isStopping() ||
context.getRoute(routeId) != route) {
+ // the context removes all the routes when it stops, and a route
that is gone or was replaced is not ours
+ return;
+ }
+ if (isThreadPoolActive(routeId)) {
+ // the calling route may still run on a thread of the route's
thread pools: stopping the route now would
+ // shut the pool down under it
+ LOG.trace("Route with id={} still has an active thread: removing
it later", routeId);
+ submitRemoveRoute(route, ROUTE_REMOVAL_RETRY_MILLIS);
+ return;
+ }
+ try {
+ LOG.debug("Removing route with id={} created by a Kamelet endpoint
that is shut down", routeId);
+ context.getRouteController().stopRoute(routeId);
+ context.removeRoute(routeId);
+ } catch (Exception e) {
+ LOG.warn("Error removing route {} created by a Kamelet endpoint
that is shut down. This exception is ignored.",
+ routeId, e);
+ }
+ }
+
+ private boolean isThreadPoolActive(String routeId) {
+ List<ThreadPoolExecutor> pools = routeThreadPools.get(routeId);
+ if (pools != null) {
+ for (ThreadPoolExecutor pool : pools) {
+ if (pool.getActiveCount() > 0) {
+ return true;
+ }
+ }
+ }
+ return false;
+ }
+
@Override
protected void doShutdown() throws Exception {
getCamelContext().getLifecycleStrategies().remove(lifecycleHandler);
@@ -491,6 +650,13 @@ public class KameletComponent extends DefaultComponent {
ServiceHelper.stopAndShutdownService(consumers);
consumers.clear();
kameletEips.clear();
+ ScheduledExecutorService executor = routeRemover;
+ routeRemover = null;
+ if (executor != null) {
+ // the context removes all the routes when it stops, so a removal
that has not started is not needed
+ getCamelContext().getExecutorServiceManager().shutdown(executor);
+ }
+ routeThreadPools.clear();
super.doShutdown();
}
@@ -515,18 +681,18 @@ public class KameletComponent extends DefaultComponent {
this.initialized = new AtomicBoolean();
}
- public void createRouteForEndpoint(KameletEndpoint endpoint, String
parentRouteId, String parentProcessorId)
+ public String 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);
+ return doCreateRouteForEndpoint(endpoint, parentRouteId,
parentProcessorId);
} finally {
lock.unlock();
}
}
- protected void doCreateRouteForEndpoint(KameletEndpoint endpoint,
String parentRouteId, String parentProcessorId)
+ protected String doCreateRouteForEndpoint(KameletEndpoint endpoint,
String parentRouteId, String parentProcessorId)
throws Exception {
final String templateId = endpoint.getTemplateId();
final String loc = endpoint.getLocation() != null ?
endpoint.getLocation() : getLocation();
@@ -548,7 +714,7 @@ public class KameletComponent extends DefaultComponent {
}
creating.push(templateId);
try {
- doCreateRouteForEndpoint(endpoint, parentRouteId,
parentProcessorId, templateId, loc);
+ return doCreateRouteForEndpoint(endpoint, parentRouteId,
parentProcessorId, templateId, loc);
} finally {
creating.pop();
if (creating.isEmpty()) {
@@ -557,7 +723,7 @@ public class KameletComponent extends DefaultComponent {
}
}
- private void doCreateRouteForEndpoint(
+ private String doCreateRouteForEndpoint(
KameletEndpoint endpoint, String parentRouteId, String
parentProcessorId, String templateId, String loc)
throws Exception {
@@ -584,6 +750,7 @@ public class KameletComponent extends DefaultComponent {
}
LOG.debug("Route with id={} created from template={}", id,
templateId);
+ return id;
} catch (Exception e) {
throw new FailedToCreateKameletException(templateId, loc, e);
}
@@ -605,20 +772,40 @@ public class KameletComponent extends DefaultComponent {
}
}
+ @Override
+ public void onThreadPoolAdd(
+ CamelContext camelContext, ThreadPoolExecutor threadPool,
String id, String sourceId, String routeId,
+ String threadPoolProfileId) {
+ if (routeId != null) {
+ routeThreadPools.computeIfAbsent(routeId, k -> new
CopyOnWriteArrayList<>()).add(threadPool);
+ }
+ }
+
+ @Override
+ public void onThreadPoolRemove(CamelContext camelContext,
ThreadPoolExecutor threadPool) {
+ routeThreadPools.values().removeIf(pools ->
pools.remove(threadPool) && pools.isEmpty());
+ }
+
public void setInitialized(boolean initialized) {
this.initialized.set(initialized);
}
- public void track(KameletEndpoint endpoint, String parentRouteId,
String parentProcessorId) {
+ /**
+ * Creates the route of the endpoint now if the context is
initialized, or when it is.
+ *
+ * @return the id of the route created now, or {@code null} if it is
created later
+ */
+ public String track(KameletEndpoint endpoint, String parentRouteId,
String parentProcessorId) {
if (this.initialized.get()) {
try {
- createRouteForEndpoint(endpoint, parentRouteId,
parentProcessorId);
+ return createRouteForEndpoint(endpoint, parentRouteId,
parentProcessorId);
} catch (Exception e) {
throw RuntimeCamelException.wrapRuntimeException(e);
}
} else {
LOG.debug("Tracking route template={} and id={}",
endpoint.getTemplateId(), endpoint.getRouteId());
this.endpoints.add(new Tuple(endpoint, parentRouteId,
parentProcessorId));
+ return null;
}
}
}
diff --git
a/components/camel-kamelet/src/main/java/org/apache/camel/component/kamelet/KameletEndpoint.java
b/components/camel-kamelet/src/main/java/org/apache/camel/component/kamelet/KameletEndpoint.java
index 879ad17ef9ca..e1cd6110630c 100644
---
a/components/camel-kamelet/src/main/java/org/apache/camel/component/kamelet/KameletEndpoint.java
+++
b/components/camel-kamelet/src/main/java/org/apache/camel/component/kamelet/KameletEndpoint.java
@@ -21,9 +21,11 @@ import java.util.HashMap;
import java.util.Map;
import org.apache.camel.AfterPropertiesConfigured;
+import org.apache.camel.AsyncCallback;
import org.apache.camel.CamelContext;
import org.apache.camel.Category;
import org.apache.camel.Consumer;
+import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.Producer;
import org.apache.camel.spi.BodyType;
@@ -201,6 +203,15 @@ public class KameletEndpoint extends DefaultEndpoint
implements AfterPropertiesC
}
}
+ /**
+ * The callback the producer calls when the Kamelet is done with the
exchange, and the calling route continues. The
+ * default is the given callback. An endpoint that removes the route it
created when it is shut down uses it to know
+ * the exchange whose call is completing when it is shut down (as toD does
inside this callback).
+ */
+ AsyncCallback completingCallback(Exchange exchange, AsyncCallback
callback) {
+ return callback;
+ }
+
@Override
public Producer createProducer() throws Exception {
return new KameletProducer(this, key);
diff --git
a/components/camel-kamelet/src/main/java/org/apache/camel/component/kamelet/KameletProducer.java
b/components/camel-kamelet/src/main/java/org/apache/camel/component/kamelet/KameletProducer.java
index 3715297bc09f..ee9296a87284 100644
---
a/components/camel-kamelet/src/main/java/org/apache/camel/component/kamelet/KameletProducer.java
+++
b/components/camel-kamelet/src/main/java/org/apache/camel/component/kamelet/KameletProducer.java
@@ -55,7 +55,9 @@ final class KameletProducer extends DefaultAsyncProducer
implements RouteIdAware
}
@Override
- public boolean process(Exchange exchange, AsyncCallback callback) {
+ public boolean process(Exchange exchange, AsyncCallback originalCallback) {
+ // the endpoint may want to know the exchange whose call is completing
when the calling route continues
+ final AsyncCallback callback = endpoint.completingCallback(exchange,
originalCallback);
try {
final KameletConsumer consumer = getConsumer();
if (consumer == null) {
diff --git
a/components/camel-kamelet/src/test/java/org/apache/camel/component/kamelet/KameletToDRoutesNotKeptTest.java
b/components/camel-kamelet/src/test/java/org/apache/camel/component/kamelet/KameletToDRoutesNotKeptTest.java
new file mode 100644
index 000000000000..418141e9e92e
--- /dev/null
+++
b/components/camel-kamelet/src/test/java/org/apache/camel/component/kamelet/KameletToDRoutesNotKeptTest.java
@@ -0,0 +1,325 @@
+/*
+ * 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 java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.Route;
+import org.apache.camel.RoutesBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.awaitility.core.ConditionTimeoutException;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+/**
+ * A Kamelet called with toD and cacheSize=-1 (the documented setting for a
dynamic endpoint that is different for every
+ * message) gets a new endpoint for every call, which creates a route from the
template. The endpoint is discarded after
+ * the call, so its route must be removed as well, otherwise every call keeps
one more route.
+ */
+class KameletToDRoutesNotKeptTest extends CamelTestSupport {
+
+ private static final int MESSAGES = 50;
+
+ @Test
+ void routesOfDiscardedKameletEndpointsAreRemoved() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMessageCount(MESSAGES);
+ int endpoints = context.getEndpointRegistry().size();
+
+ for (int i = 0; i < MESSAGES; i++) {
+ template.sendBodyAndHeader("direct:prototype", "body", "id", "id"
+ i);
+ }
+ mock.assertIsSatisfied();
+ for (int i = 0; i < MESSAGES; i++) {
+
assertThat(mock.getReceivedExchanges().get(i).getMessage().getBody()).isEqualTo("id"
+ i + "-body");
+ }
+
+ // the routes are removed on a thread of the component, once the
exchange is done, and their endpoints
+ // (kamelet:source) with them: removeRoute takes the route out of
getRoutes() before it removes its endpoints
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
+ assertThat(kameletRoutes()).isZero();
+
assertThat(context.getEndpointRegistry().values()).hasSize(endpoints);
+ });
+ }
+
+ @Test
+ void sameValueEveryTimeDoesNotKeepRoutes() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMessageCount(MESSAGES);
+
+ for (int i = 0; i < MESSAGES; i++) {
+ template.sendBodyAndHeader("direct:prototype", "body" + i, "id",
"same");
+ }
+ mock.assertIsSatisfied();
+ for (int i = 0; i < MESSAGES; i++) {
+
assertThat(mock.getReceivedExchanges().get(i).getMessage().getBody()).isEqualTo("same-body"
+ i);
+ }
+
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(() ->
assertThat(kameletRoutes()).isZero());
+ }
+
+ @Test
+ void cachedKameletEndpointsKeepTheirRoutes() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("a-1", "b-2", "a-3", "b-4");
+
+ String[] ids = { "a", "b", "a", "b" };
+ for (int i = 0; i < ids.length; i++) {
+ template.sendBodyAndHeader("direct:cached", String.valueOf(i + 1),
"id", ids[i]);
+ }
+ mock.assertIsSatisfied();
+
+ // the default toD keeps the endpoints and so their routes for the
next messages
+ assertThat(kameletRoutes()).isEqualTo(2);
+ }
+
+ @Test
+ void templateWithThreadsDoesNotInterruptTheCallingRoute() throws Exception
{
+ MockEndpoint mock = getMockEndpoint("mock:threads");
+ mock.expectedMessageCount(MESSAGES);
+
+ for (int i = 0; i < MESSAGES; i++) {
+ template.sendBodyAndHeader("direct:threads", "body", "id", "id" +
i);
+ }
+ mock.assertIsSatisfied();
+ for (int i = 0; i < MESSAGES; i++) {
+ Exchange exchange = mock.getReceivedExchanges().get(i);
+ assertThat(exchange.getMessage().getBody()).isEqualTo("id" + i +
"-body");
+ // the calling route continues on a thread of the pool of the
Kamelet's route
+ assertThat(exchange.getMessage().getHeader("thread",
String.class)).contains("Threads");
+ assertThat(exchange.getMessage().getHeader("interrupted",
Boolean.class)).isFalse();
+ // the Kamelet's route is removed only after the exchange is done
+ assertThat(exchange.getMessage().getHeader("kameletRouteKept",
Boolean.class)).isTrue();
+ }
+
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(() ->
assertThat(kameletRoutes()).isZero());
+ }
+
+ @Test
+ void kameletRouteIsKeptWhileTheCallingRouteRunsOnItsThread() throws
Exception {
+ MockEndpoint mock = getMockEndpoint("mock:slow");
+ mock.expectedBodiesReceived("x-body");
+
+ template.sendBodyAndHeader("direct:slow", "body", "id", "x");
+ mock.assertIsSatisfied();
+
+ Exchange exchange = mock.getReceivedExchanges().get(0);
+ assertThat(exchange.getMessage().getHeader("thread",
String.class)).contains("Threads");
+ assertThat(exchange.getMessage().getHeader("kameletRouteKept",
Boolean.class)).isTrue();
+ assertThat(exchange.getMessage().getHeader("interrupted",
Boolean.class)).isFalse();
+
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(() ->
assertThat(kameletRoutes()).isZero());
+ }
+
+ @Test
+ void
kameletRouteIsKeptWhileTheCallingRouteRunsOnItsThreadAfterARecipientList()
throws Exception {
+ // recipientList with cacheSize=-1 shuts the endpoint down when it
releases the producer, after the call has
+ // completed: the exchange of the calling route is not known there
+
assertKameletRouteIsKeptWhileTheCallingRouteRunsOnItsThread("direct:recipientList");
+ }
+
+ @Test
+ void
kameletRouteIsKeptWhileTheCallingRouteRunsOnItsThreadAfterARoutingSlip() throws
Exception {
+ // routingSlip with cacheSize=-1 shuts the endpoint down in its
callback, with a copy of the exchange that has
+ // no unit of work
+
assertKameletRouteIsKeptWhileTheCallingRouteRunsOnItsThread("direct:routingSlip");
+ }
+
+ private void
assertKameletRouteIsKeptWhileTheCallingRouteRunsOnItsThread(String caller)
throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:slow");
+ mock.expectedBodiesReceived("x-body");
+
+ template.sendBodyAndHeader(caller, "body", "id", "x");
+ mock.assertIsSatisfied();
+
+ Exchange exchange = mock.getReceivedExchanges().get(0);
+ assertThat(exchange.getMessage().getHeader("thread",
String.class)).contains("Threads");
+ assertThat(exchange.getMessage().getHeader("kameletRouteKept",
Boolean.class)).isTrue();
+ assertThat(exchange.getMessage().getHeader("interrupted",
Boolean.class)).isFalse();
+
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(() ->
assertThat(kameletRoutes()).isZero());
+ }
+
+ @Test
+ void sendsFromSeveralThreadsAtOnce() throws Exception {
+ int threads = 8;
+ int perThread = 25;
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMessageCount(threads * perThread);
+
+ ExecutorService executor = Executors.newFixedThreadPool(threads);
+ try {
+ List<Future<Object>> replies = new ArrayList<>();
+ for (int t = 0; t < threads; t++) {
+ for (int i = 0; i < perThread; i++) {
+ String id = "t" + t + "-" + i;
+ replies.add(executor.submit(() ->
template.requestBodyAndHeader("direct:prototype", "body", "id", id)));
+ }
+ }
+ List<Object> bodies = new ArrayList<>();
+ for (Future<Object> reply : replies) {
+ bodies.add(reply.get(30, TimeUnit.SECONDS));
+ }
+ List<String> expected = new ArrayList<>();
+ for (int t = 0; t < threads; t++) {
+ for (int i = 0; i < perThread; i++) {
+ expected.add("t" + t + "-" + i + "-body");
+ }
+ }
+ assertThat(bodies).containsExactlyElementsOf(expected);
+ } finally {
+ executor.shutdownNow();
+ assertThat(executor.awaitTermination(10,
TimeUnit.SECONDS)).isTrue();
+ }
+ mock.assertIsSatisfied();
+
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(() ->
assertThat(kameletRoutes()).isZero());
+ }
+
+ @Test
+ void routeIdGivenInTheUriIsLeftAlone() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMessageCount(MESSAGES);
+
+ for (int i = 0; i < MESSAGES; i++) {
+ template.sendBodyAndHeader("direct:myRoute", "body", "id", "id" +
i);
+ }
+ mock.assertIsSatisfied();
+ for (int i = 0; i < MESSAGES; i++) {
+
assertThat(mock.getReceivedExchanges().get(i).getMessage().getBody()).isEqualTo("id"
+ i + "-body");
+ }
+
+ // every endpoint replaces the route of the previous one, as they
share the route id: the last one is kept
+
assertThat(context.getRoutes()).filteredOn(Route::isCreatedByKamelet).extracting(Route::getRouteId)
+ .containsExactly("myRoute");
+ }
+
+ @Test
+ void routeOfKameletIsRemovedWithTheRouteAddedAtRuntimeThatConsumesFromIt()
throws Exception {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("kamelet:source-of?name=a").routeId("consumer")
+ .to("mock:consumer");
+ }
+ });
+ assertThat(kameletRoutes()).isEqualTo(1);
+
+ MockEndpoint mock = getMockEndpoint("mock:consumer");
+ mock.expectedBodiesReceived("from a");
+ template.sendBody("direct:source-a", "from a");
+ mock.assertIsSatisfied();
+
+ // removing the route shuts down its input endpoint, which removes the
route it created from the template
+ context.getRouteController().stopRoute("consumer");
+ assertThat(context.removeRoute("consumer")).isTrue();
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(() ->
assertThat(kameletRoutes()).isZero());
+ }
+
+ private long kameletRoutes() {
+ return
context.getRoutes().stream().filter(Route::isCreatedByKamelet).count();
+ }
+
+ @Override
+ protected RoutesBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ routeTemplate("prefix")
+ .templateParameter("value")
+ .from("kamelet:source")
+ .setBody().simple("{{value}}-${body}");
+
+ routeTemplate("pooled")
+ .templateParameter("value")
+ .from("kamelet:source")
+ .threads(2)
+ .setHeader("kameletRouteId").simple("${routeId}")
+ .setBody().simple("{{value}}-${body}");
+
+ routeTemplate("source-of")
+ .templateParameter("name")
+ .from("direct:source-{{name}}")
+ .to("kamelet:sink");
+
+ from("direct:prototype").routeId("prototype")
+ .toD("kamelet:prefix?value=${header.id}", -1)
+ .to("mock:result");
+
+ from("direct:cached").routeId("cached")
+ .toD("kamelet:prefix?value=${header.id}")
+ .to("mock:result");
+
+ from("direct:myRoute").routeId("myRouteCaller")
+ .toD("kamelet:prefix/myRoute?value=${header.id}", -1)
+ .to("mock:result");
+
+ from("direct:threads").routeId("threadsCaller")
+ .toD("kamelet:pooled?value=${header.id}", -1)
+ .process(e -> {
+ e.getMessage().setHeader("thread",
Thread.currentThread().getName());
+ e.getMessage().setHeader("interrupted",
Thread.currentThread().isInterrupted());
+ String id =
e.getMessage().getHeader("kameletRouteId", String.class);
+ e.getMessage().setHeader("kameletRouteKept",
context.getRoute(id) != null);
+ })
+ .to("mock:threads");
+
+ from("direct:slow").routeId("slowCaller")
+ .toD("kamelet:pooled?value=${header.id}", -1)
+ .process(e -> recordWhetherTheKameletRouteIsKept(e))
+ .to("mock:slow");
+
+ from("direct:recipientList").routeId("recipientListCaller")
+
.recipientList(simple("kamelet:pooled?value=${header.id}")).cacheSize(-1)
+ .process(e -> recordWhetherTheKameletRouteIsKept(e))
+ .to("mock:slow");
+
+ from("direct:routingSlip").routeId("routingSlipCaller")
+
.routingSlip(simple("kamelet:pooled?value=${header.id}")).cacheSize(-1)
+ .process(e -> recordWhetherTheKameletRouteIsKept(e))
+ .to("mock:slow");
+ }
+ };
+ }
+
+ private void recordWhetherTheKameletRouteIsKept(Exchange e) {
+ e.getMessage().setHeader("thread", Thread.currentThread().getName());
+ String id = e.getMessage().getHeader("kameletRouteId", String.class);
+ boolean kept;
+ try {
+ // the rest of the calling route runs on a thread of the Kamelet's
route, which must not be stopped
+ // (shutting down its thread pool) before the calling route has
left that thread
+ await().during(500, TimeUnit.MILLISECONDS).atMost(2,
TimeUnit.SECONDS)
+ .until(() -> context.getRoute(id) != null);
+ kept = true;
+ } catch (ConditionTimeoutException ex) {
+ kept = false;
+ }
+ e.getMessage().setHeader("kameletRouteKept", kept);
+ e.getMessage().setHeader("interrupted",
Thread.currentThread().isInterrupted());
+ }
+}
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 fedf1ea33b06..78e56e371433 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
@@ -1560,6 +1560,17 @@ placeholder or a route template parameter in them never
matched (Switch) or did
They are now resolved at startup like the other EIP options; a placeholder of
a property that does not exist now
fails the route at startup.
+=== camel-kamelet - the route of a Kamelet endpoint is removed with the
endpoint
+
+A Kamelet endpoint creates a route from the Kamelet's template. When the
endpoint is created while the CamelContext
+runs and is shut down later, its route is now stopped and removed as well;
before, it stayed behind and kept running.
+This applies to an endpoint that is used once and discarded (`toD`,
`recipientList` or `routingSlip` with
+`cacheSize=-1`), which kept one more route for every message. It also applies
to a route added at runtime (such as a
+dev-mode reload or `CamelContext.addRoutes`) that consumes from a Kamelet
(`from("kamelet:my-source")`): removing that
+route now also removes the route of the Kamelet. A route id given in the uri
(`kamelet:my-source/myRoute` or
+`routeId=`) is shared by the endpoints that use it, so such a route is not
removed. Routes created when the
+CamelContext starts are not changed.
+
=== camel-undertow - the producer refuses a request target or header value
with control characters
The undertow producer writes the path, query and header values of the outgoing
request to the socket one character at
