davsclaus commented on code in PR #28260:
URL: https://github.com/apache/camel/pull/28260#discussion_r4345470820
##########
components/camel-kamelet/src/main/java/org/apache/camel/component/kamelet/KameletComponent.java:
##########
@@ -484,13 +536,86 @@ protected void doInit() throws Exception {
super.doInit();
}
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+ ExecutorService executor = routeRemover;
+ if (executor == null || executor.isShutdown()) {
+ // the threads are created on first use
+ routeRemover =
getCamelContext().getExecutorServiceManager().newSingleThreadExecutor(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, after
the exchange whose call is completing is
+ * done: 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. Stopping
a route also holds the lock of the
+ * CamelContext while the route shuts down gracefully.
+ *
+ * @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);
+ }
+ });
+ } else {
+ submitRemoveRoute(route);
Review Comment:
This path is taken when the endpoint is shut down outside the producer
callback: recipientList `releaseProducer()`, pollEnrich, and routingSlip when
the copy has no unit of work. The removal then starts at once, so with a
`threads()` template the calling route can still be on that pool when it is
shut down. A test with recipientList `cacheSize=-1` and the `pooled` template
would show whether this matters. If it does not get fixed here, a comment
explaining the gap would help.
##########
components/camel-kamelet/src/test/java/org/apache/camel/component/kamelet/KameletToDRoutesNotKeptTest.java:
##########
@@ -0,0 +1,282 @@
+/*
+ * 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
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(() ->
assertThat(kameletRoutes()).isZero());
+ // and their endpoints (kamelet:source) with them
+ assertThat(context.getEndpointRegistry().values()).hasSize(endpoints);
Review Comment:
This can flake. `removeRoute` takes the route out of `getRoutes()` while it
shuts down the route service, and removes the route's private endpoints
(`kamelet:source`) only after that, in `doRemove`. So the await on
`kameletRoutes()` can pass while the last endpoint is still in the registry.
Please move this assertion into the `untilAsserted` block, or into its own
await.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
