This is an automated email from the ASF dual-hosted git repository. luigidemasi pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/camel.git
commit cdc2d5bc8ce547cda534f220a6e9ea69364b25e3 Author: Luigi De Masi <[email protected]> AuthorDate: Fri Oct 9 19:34:43 2026 +0200 CAMEL-25491: Deregister semantic evaluation executors on stop Shut down the connector and developer-console pools through Camel's ExecutorServiceManager so pool tracking and lifecycle management are removed while the context remains running. Extend shutdown-race and console restart tests to assert pool removal. Co-authored-by: Codex <[email protected]> --- .../apache/camel/semantic/SemanticEvaluateConsole.java | 2 +- .../org/apache/camel/semantic/SemanticConsoleTest.java | 18 ++++++++++++++++++ .../apache/camel/cli/connector/LocalCliConnector.java | 2 +- .../LocalCliConnectorSemanticLifecycleTest.java | 18 +++++++++++++++++- 4 files changed, 37 insertions(+), 3 deletions(-) diff --git a/components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/SemanticEvaluateConsole.java b/components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/SemanticEvaluateConsole.java index 2c185cba03a9..5f305998a818 100644 --- a/components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/SemanticEvaluateConsole.java +++ b/components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/SemanticEvaluateConsole.java @@ -82,7 +82,7 @@ public class SemanticEvaluateConsole extends AbstractDevConsole { protected synchronized void doStop() throws Exception { stopped = true; if (executor != null) { - executor.shutdownNow(); + getCamelContext().getExecutorServiceManager().shutdownNow(executor); executor = null; } super.doStop(); diff --git a/components/camel-ai/camel-semantic/src/test/java/org/apache/camel/semantic/SemanticConsoleTest.java b/components/camel-ai/camel-semantic/src/test/java/org/apache/camel/semantic/SemanticConsoleTest.java index 1f494a8a83e3..88314c0bf58f 100644 --- a/components/camel-ai/camel-semantic/src/test/java/org/apache/camel/semantic/SemanticConsoleTest.java +++ b/components/camel-ai/camel-semantic/src/test/java/org/apache/camel/semantic/SemanticConsoleTest.java @@ -19,9 +19,11 @@ package org.apache.camel.semantic; import java.util.List; import java.util.Map; import java.util.concurrent.ExecutorService; +import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; +import org.apache.camel.CamelContext; import org.apache.camel.console.DevConsole; import org.apache.camel.impl.DefaultCamelContext; import org.apache.camel.impl.engine.DefaultExecutorServiceManager; @@ -29,6 +31,7 @@ import org.apache.camel.language.semantic.SemanticLanguage; import org.apache.camel.semantic.SemanticExpert.InputType; import org.apache.camel.semantic.SemanticExpert.ResultType; import org.apache.camel.spi.ThreadPoolProfile; +import org.apache.camel.support.LifecycleStrategySupport; import org.apache.camel.support.PluginHelper; import org.apache.camel.util.json.JsonObject; import org.junit.jupiter.api.Test; @@ -217,6 +220,18 @@ class SemanticConsoleTest { @Test void stoppedConsoleRejectsCallsUntilRestarted() throws Exception { try (var context = new DefaultCamelContext()) { + var removals = new AtomicInteger(); + context.addLifecycleStrategy(new LifecycleStrategySupport() { + @Override + public void onThreadPoolRemove(CamelContext context, ThreadPoolExecutor executor) { + onThreadPoolRemove(context, (ExecutorService) executor); + } + + @Override + public void onThreadPoolRemove(CamelContext context, ExecutorService executor) { + removals.incrementAndGet(); + } + }); var expert = new Scorer(); context.getRegistry().bind("risk", expert); context.start(); @@ -225,6 +240,8 @@ class SemanticConsoleTest { // Resolver-created consoles can be called before explicit lifecycle startup. assertThat(call(console, options)).containsEntry("status", "success"); console.stop(); + assertThat(context.isStarted()).isTrue(); + assertThat(removals).hasValue(1); assertThat(call(console, options)).containsEntry("status", "failed") .containsEntry("error", "Semantic evaluation console is stopping or stopped"); assertThat(expert.calls).hasValue(1); @@ -235,6 +252,7 @@ class SemanticConsoleTest { } finally { console.stop(); } + assertThat(removals).hasValue(2); } } diff --git a/dsl/camel-cli-connector/src/main/java/org/apache/camel/cli/connector/LocalCliConnector.java b/dsl/camel-cli-connector/src/main/java/org/apache/camel/cli/connector/LocalCliConnector.java index 8318887b1594..2e0325483b3f 100644 --- a/dsl/camel-cli-connector/src/main/java/org/apache/camel/cli/connector/LocalCliConnector.java +++ b/dsl/camel-cli-connector/src/main/java/org/apache/camel/cli/connector/LocalCliConnector.java @@ -1928,7 +1928,7 @@ public class LocalCliConnector extends ServiceSupport // Stop the transport before taking this monitor: file dispatch may hold it while the transport stops. synchronized (this) { if (semanticExecutor != null) { - semanticExecutor.shutdownNow(); + camelContext.getExecutorServiceManager().shutdownNow(semanticExecutor); semanticExecutor = null; } } diff --git a/dsl/camel-cli-connector/src/test/java/org/apache/camel/cli/connector/LocalCliConnectorSemanticLifecycleTest.java b/dsl/camel-cli-connector/src/test/java/org/apache/camel/cli/connector/LocalCliConnectorSemanticLifecycleTest.java index 95f145ab412f..c8ba1664616f 100644 --- a/dsl/camel-cli-connector/src/test/java/org/apache/camel/cli/connector/LocalCliConnectorSemanticLifecycleTest.java +++ b/dsl/camel-cli-connector/src/test/java/org/apache/camel/cli/connector/LocalCliConnectorSemanticLifecycleTest.java @@ -21,6 +21,7 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.FutureTask; +import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; @@ -29,6 +30,7 @@ import org.apache.camel.impl.DefaultCamelContext; import org.apache.camel.impl.engine.DefaultExecutorServiceManager; import org.apache.camel.spi.CliConnectorFactory; import org.apache.camel.spi.ThreadPoolProfile; +import org.apache.camel.support.LifecycleStrategySupport; import org.apache.camel.util.json.JsonObject; import org.junit.jupiter.api.Test; @@ -43,6 +45,18 @@ class LocalCliConnectorSemanticLifecycleTest { var created = new CountDownLatch(1); var release = new CountDownLatch(1); var pool = new AtomicReference<ExecutorService>(); + var removed = new AtomicReference<ExecutorService>(); + context.addLifecycleStrategy(new LifecycleStrategySupport() { + @Override + public void onThreadPoolRemove(CamelContext context, ThreadPoolExecutor executor) { + onThreadPoolRemove(context, (ExecutorService) executor); + } + + @Override + public void onThreadPoolRemove(CamelContext context, ExecutorService executor) { + removed.set(executor); + } + }); context.setExecutorServiceManager(new DefaultExecutorServiceManager(context) { @Override public ExecutorService newThreadPool(Object source, String name, ThreadPoolProfile profile) { @@ -82,13 +96,15 @@ class LocalCliConnectorSemanticLifecycleTest { dispatch.get(5, TimeUnit.SECONDS).cancel(true); assertThat(connector.isStopped()).isTrue(); assertThat(pool.get().isShutdown()).isTrue(); + assertThat(context.isStarted()).isTrue(); + assertThat(removed.get()).isSameAs(pool.get()); } finally { release.countDown(); caller.join(5000); stopper.join(5000); connector.stop(); if (pool.get() != null) { - pool.get().shutdownNow(); + context.getExecutorServiceManager().shutdownNow(pool.get()); } } }
