allthingssecurity commented on code in PR #28260:
URL: https://github.com/apache/camel/pull/28260#discussion_r4345005702
##########
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:
Done in c2f8f4c057b9. The stop and remove now run on a single thread the
component owns (`newSingleThreadExecutor` from the `ExecutorServiceManager`,
created in `doStart`, shut down in `doShutdown`).
Handing it off alone was not enough for `threads()`: the calling route goes
on running on the pool thread after the toD callback, so a removal on another
thread can still `shutdownNow` that pool while the caller uses it. So when the
endpoint is shut down while a call is completing (toD does it inside the
producer callback), the removal waits for that exchange to be done (an
onCompletion on it) and is then submitted to the component thread. New tests:
`templateWithThreadsDoesNotInterruptTheCallingRoute` fails with the inline
removal (`Expecting value to be false but was true`), and
`kameletRouteIsKeptWhileTheCallingRouteRunsOnItsThread` fails when the removal
does not wait for the exchange.
_Claude Code on behalf of allthingssecurity_
##########
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:
Done in c2f8f4c057b9. The endpoint keeps the `Route` instance and removes it
only while `context.getRoute(id)` is still that instance. On top of that it
removes only a route whose id was generated for this endpoint. With an id given
in the uri, every endpoint replaces the previous one's route (`DefaultModel`,
same template), so there is never more than one such route. Leaving it alone
also closes the gap between the identity check and `stopRoute(id)`.
`routeIdGivenInTheUriIsLeftAlone` pins that only `myRoute` is kept after 50
calls. `sendsFromSeveralThreadsAtOnce` sends 200 requests from 8 threads and
checks every reply.
_Claude Code on behalf of allthingssecurity_
##########
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:
Yes, it is intended. The comment in `doShutdown` and the javadoc of
`removeCreatedRoute` now say so, and
`routeOfKameletIsRemovedWithTheRouteAddedAtRuntimeThatConsumesFromIt` covers
it: it adds `from("kamelet:source-of?name=a")` at runtime, routes a message,
removes the route, and the Kamelet's route goes too. Since the removal now runs
on the component thread, it is no longer nested inside `removeRoute`; it waits
for the context lock until that call is done. Routes created when the context
starts are not changed: their endpoint does not record the route (it is created
later, in `onContextInitialized`). The PR description says this now as well.
_Claude Code on behalf of allthingssecurity_
##########
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:
Both added in c2f8f4c057b9:
`templateWithThreadsDoesNotInterruptTheCallingRoute` (50 messages through a
template with `threads(2)`; the processor after the toD records the thread name
and `isInterrupted()`),
`kameletRouteIsKeptWhileTheCallingRouteRunsOnItsThread`, and
`sendsFromSeveralThreadsAtOnce` (8 threads, 200 requests). The test class now
uses AssertJ.
_Claude Code on behalf of allthingssecurity_
--
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]
