davsclaus commented on code in PR #28260:
URL: https://github.com/apache/camel/pull/28260#discussion_r4344924913
##########
components/camel-kamelet/src/main/java/org/apache/camel/component/kamelet/KameletComponent.java:
##########
@@ -484,6 +499,26 @@ protected void doInit() throws Exception {
super.doInit();
}
+ /**
+ * Stops and 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).
+ */
+ private void removeCreatedRoute(String routeId) {
+ CamelContext context = getCamelContext();
+ if (context == null || !context.isStarted() || context.isStopping() ||
context.getRoute(routeId) == null) {
Review Comment:
Not blocking (point 2): `getRoute(routeId) != null` does not prove this is
the route this endpoint created. With a user-given route id
(`kamelet:prefix/myRoute`), every endpoint shares the id, and a newer endpoint
replaces the route. Remembering the `Route` instance returned after creation,
and removing it only while `context.getRoute(id)` is still that instance, would
keep one endpoint from removing a route that another in-flight exchange uses.
##########
components/camel-kamelet/src/main/java/org/apache/camel/component/kamelet/KameletComponent.java:
##########
@@ -484,6 +499,26 @@ protected void doInit() throws Exception {
super.doInit();
}
+ /**
+ * Stops and 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).
+ */
+ private void removeCreatedRoute(String routeId) {
+ CamelContext context = getCamelContext();
+ if (context == null || !context.isStarted() || context.isStopping() ||
context.getRoute(routeId) == null) {
+ // the context removes all the routes when it stops
+ 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);
Review Comment:
Blocking (point 1): this runs on the thread that finishes the exchange, from
inside the toD callback, and `stopRoute` holds the CamelContext lock while it
runs. If the Kamelet's route finishes on its own thread, that thread shuts down
its own pool. With `threads()`, `shutdownNow` interrupts the current thread.
With an async kafka producer, `shutdownGraceful(workerPool, 30000)` runs from a
worker of that same pool, so it waits 30s and then interrupts itself. Please
run the stop and remove on a separate thread that the component owns.
##########
components/camel-kamelet/src/test/java/org/apache/camel/component/kamelet/KameletToDRoutesNotKeptTest.java:
##########
@@ -0,0 +1,110 @@
+/*
+ * 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 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.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * 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);
+
+ for (int i = 0; i < MESSAGES; i++) {
+ template.sendBodyAndHeader("direct:prototype", "body", "id", "id"
+ i);
+ }
+ mock.assertIsSatisfied();
+ for (int i = 0; i < MESSAGES; i++) {
+ assertEquals("id" + i + "-body",
mock.getReceivedExchanges().get(i).getMessage().getBody());
+ }
+
+ assertEquals(0, kameletRoutes(), "routes created by the Kamelet and
still kept after " + MESSAGES + " calls");
+ assertEquals(3, context.getEndpointRegistry().size(),
+ "endpoints after " + MESSAGES + " calls: " +
context.getEndpointRegistry().values());
+ }
+
+ @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++) {
+ assertEquals("same-body" + i,
mock.getReceivedExchanges().get(i).getMessage().getBody());
+ }
+
+ assertEquals(0, kameletRoutes(), "routes created by the Kamelet and
still kept after " + MESSAGES + " calls");
+ }
+
+ @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
+ assertEquals(2, kameletRoutes());
+ }
+
+ 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}");
+
+ from("direct:prototype").routeId("prototype")
+ .toD("kamelet:prefix?value=${header.id}", -1)
+ .to("mock:result");
Review Comment:
Optional: please add a variant whose template uses `threads()` (the calling
route must not continue on an interrupted thread), and one that sends from
several threads at once. These cover points 1 and 2 of the review.
##########
components/camel-kamelet/src/main/java/org/apache/camel/component/kamelet/KameletComponent.java:
##########
@@ -484,6 +499,26 @@ protected void doInit() throws Exception {
super.doInit();
}
+ /**
+ * Stops and 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).
+ */
Review Comment:
Not blocking (point 3): this is also reached when a route that consumes from
a Kamelet (`from("kamelet:my-source")`) and was added after the context started
is removed, because `RouteService.doShutdown` shuts down the route's input
endpoint. That route is then removed as well, nested inside `removeRoute`. It
is probably what we want, but please say so here and cover it with a test.
--
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]
