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 5e08a7e3b8eaa2e0319601cdd440f2dacd3c876b Author: Luigi De Masi <[email protected]> AuthorDate: Fri Oct 9 13:37:15 2026 +0200 CAMEL-25491: Fix semantic evaluation lifecycle and diagnostics Serialize CLI semantic executor initialization, submission and shutdown, and reject late requests once stopping begins. Stop transport polling before acquiring the executor monitor so file dispatch can finish safely. Preserve error details for message-less exceptions on both console failure paths, and reject calls after console shutdown until it is restarted. Throttle missing semantic-console discovery to five-second retries while preserving late registration. Restore server-side action failure warnings. Add regressions for executor initialization racing with shutdown, late calls, failure details, and discovery retry behavior. Co-authored-by: Codex <[email protected]> --- .../camel/semantic/SemanticEvaluateConsole.java | 16 ++- .../apache/camel/semantic/SemanticConsoleTest.java | 61 +++++++++ .../cli/connector/FileCliConnectorTransport.java | 1 + .../camel/cli/connector/LocalCliConnector.java | 39 ++++-- .../LocalCliConnectorSemanticLifecycleTest.java | 144 +++++++++++++++++++++ .../LocalCliConnectorSemanticMetadataTest.java | 26 +++- 6 files changed, 273 insertions(+), 14 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 6ff582a75ff8..2c185cba03a9 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 @@ -58,8 +58,13 @@ public class SemanticEvaluateConsole extends AbstractDevConsole { public static final String TIMEOUT = "timeout"; private ExecutorService executor; + // A resolver-created console may be called without starting it; only an explicit stop rejects calls. + private boolean stopped; private synchronized ExecutorService executor() throws Exception { + if (stopped || isStopping()) { + throw new IllegalStateException("Semantic evaluation console is stopping or stopped"); + } if (executor == null) { executor = new ThreadPoolBuilder(getCamelContext()).poolSize(2).maxPoolSize(2).maxQueueSize(0) .rejectedPolicy(ThreadPoolRejectedPolicy.Abort).build(this, "SemanticEvaluation"); @@ -67,8 +72,15 @@ public class SemanticEvaluateConsole extends AbstractDevConsole { return executor; } + @Override + protected synchronized void doStart() throws Exception { + stopped = false; + super.doStart(); + } + @Override protected synchronized void doStop() throws Exception { + stopped = true; if (executor != null) { executor.shutdownNow(); executor = null; @@ -124,7 +136,7 @@ public class SemanticEvaluateConsole extends AbstractDevConsole { } catch (RejectedExecutionException e) { failure = "Busy: semantic evaluations are already running"; } catch (Exception e) { - failure = e.getMessage(); + failure = e.getMessage() != null ? e.getMessage() : e.getClass().getSimpleName(); } finally { if (task != null && !task.isDone()) { task.cancel(true); @@ -186,7 +198,7 @@ public class SemanticEvaluateConsole extends AbstractDevConsole { } catch (Exception e) { return JsonRecordSupport.toJsonObject(new Response( name, "failed", null, null, null, null, null, - elapsed(start), e.getMessage())); + elapsed(start), e.getMessage() != null ? e.getMessage() : e.getClass().getSimpleName())); } } 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 3a77e9e2843b..1f494a8a83e3 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 @@ -18,14 +18,17 @@ package org.apache.camel.semantic; import java.util.List; import java.util.Map; +import java.util.concurrent.ExecutorService; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; import org.apache.camel.console.DevConsole; import org.apache.camel.impl.DefaultCamelContext; +import org.apache.camel.impl.engine.DefaultExecutorServiceManager; 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.PluginHelper; import org.apache.camel.util.json.JsonObject; import org.junit.jupiter.api.Test; @@ -177,6 +180,64 @@ class SemanticConsoleTest { } } + @Test + void directEvaluationReportsExceptionsWithoutMessages() throws Exception { + try (var context = new DefaultCamelContext()) { + context.getRegistry().bind("risk", new Scorer() { + @Override + public SemanticResult evaluate(SemanticEvaluation evaluation, Object state) { + throw new IllegalStateException(); + } + }); + context.start(); + assertThat(call(console(context, "semantic-evaluate"), + Map.of("expert", "risk", "operation", "risk", "input", "test"))) + .containsEntry("status", "failed").containsEntry("error", "IllegalStateException"); + } + } + + @Test + void executorFailureReportsExceptionsWithoutMessages() throws Exception { + try (var context = new DefaultCamelContext()) { + context.setExecutorServiceManager(new DefaultExecutorServiceManager(context) { + @Override + public ExecutorService newThreadPool(Object source, String name, ThreadPoolProfile profile) { + if ("SemanticEvaluation".equals(name)) { + throw new IllegalStateException(); + } + return super.newThreadPool(source, name, profile); + } + }); + context.start(); + assertThat(call(console(context, "semantic-evaluate"), Map.of())) + .containsEntry("status", "failed").containsEntry("error", "IllegalStateException"); + } + } + + @Test + void stoppedConsoleRejectsCallsUntilRestarted() throws Exception { + try (var context = new DefaultCamelContext()) { + var expert = new Scorer(); + context.getRegistry().bind("risk", expert); + context.start(); + var console = (SemanticEvaluateConsole) console(context, "semantic-evaluate"); + Map<String, Object> options = Map.of("expert", "risk", "operation", "risk", "input", "test"); + // Resolver-created consoles can be called before explicit lifecycle startup. + assertThat(call(console, options)).containsEntry("status", "success"); + console.stop(); + assertThat(call(console, options)).containsEntry("status", "failed") + .containsEntry("error", "Semantic evaluation console is stopping or stopped"); + assertThat(expert.calls).hasValue(1); + console.start(); + try { + assertThat(call(console, options)).containsEntry("status", "success"); + assertThat(expert.calls).hasValue(2); + } finally { + console.stop(); + } + } + } + private static SemanticEvaluation definition(String expert, String state) { return new SemanticEvaluation("risk", expert, state, Map.of()); } diff --git a/dsl/camel-cli-connector/src/main/java/org/apache/camel/cli/connector/FileCliConnectorTransport.java b/dsl/camel-cli-connector/src/main/java/org/apache/camel/cli/connector/FileCliConnectorTransport.java index a355f189a5db..d49e2cea803a 100644 --- a/dsl/camel-cli-connector/src/main/java/org/apache/camel/cli/connector/FileCliConnectorTransport.java +++ b/dsl/camel-cli-connector/src/main/java/org/apache/camel/cli/connector/FileCliConnectorTransport.java @@ -222,6 +222,7 @@ public class FileCliConnectorTransport extends ServiceSupport implements CliConn result = action.result().get(); } catch (Exception failure) { Throwable cause = failure.getCause() != null ? failure.getCause() : failure; + LOG.warn("Error executing action: {} due to: {}", request, cause.getMessage(), cause); result = new JsonObject(); result.put("status", "failed"); result.put("error", cause.getMessage() != null ? cause.getMessage() : cause.getClass().getSimpleName()); 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 1d0a6d6bf8b3..8318887b1594 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 @@ -45,6 +45,7 @@ import java.util.concurrent.Executors; import java.util.concurrent.FutureTask; import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; import org.apache.camel.CamelContext; import org.apache.camel.CamelContextAware; @@ -117,6 +118,7 @@ public class LocalCliConnector extends ServiceSupport private static final Logger LOG = LoggerFactory.getLogger(LocalCliConnector.class); private static final int BODY_MAX_CHARS = 128 * 1024; + private static final long SEMANTIC_CONSOLE_RETRY_NANOS = TimeUnit.SECONDS.toNanos(5); private final CliConnectorFactory cliConnectorFactory; private CamelContext camelContext; @@ -126,6 +128,7 @@ public class LocalCliConnector extends ServiceSupport private String mainClass; private volatile ExecutorService terminateExecutor; private ExecutorService semanticExecutor; + private final AtomicLong semanticConsoleLookup = new AtomicLong(System.nanoTime() - SEMANTIC_CONSOLE_RETRY_NANOS); private ProducerTemplate producer; private ConsumerTemplate consumer; // where the running action writes its result (actions run on one thread at a time) @@ -260,12 +263,17 @@ public class LocalCliConnector extends ServiceSupport } }); try { - if (semanticExecutor == null) { - semanticExecutor = new ThreadPoolBuilder(camelContext).poolSize(2).maxPoolSize(2).maxQueueSize(0) - .rejectedPolicy(ThreadPoolRejectedPolicy.Abort) - .build("cli-semantic-evaluation", "CliSemanticEvaluation"); + synchronized (this) { + if (isStoppingOrStopped()) { + throw new IllegalStateException("CLI connector is stopping or stopped"); + } + if (semanticExecutor == null) { + semanticExecutor = new ThreadPoolBuilder(camelContext).poolSize(2).maxPoolSize(2).maxQueueSize(0) + .rejectedPolicy(ThreadPoolRejectedPolicy.Abort) + .build("cli-semantic-evaluation", "CliSemanticEvaluation"); + } + semanticExecutor.execute(task); } - semanticExecutor.execute(task); } catch (RejectedExecutionException busy) { result.completeExceptionally(new IllegalStateException("Busy: semantic evaluations are already running")); } catch (Exception e) { @@ -1658,8 +1666,16 @@ public class LocalCliConnector extends ServiceSupport root.put("sqlTrace", json); } } - // Advertise the optional Semantic screen before its first metadata request. - dcr.resolveById("semantic-metadata"); + // Discover the optional console immediately, then retry misses without scanning on every status poll. + // Keep retrying so consoles registered after startup are still advertised. + if (dcr.getConsole("semantic-metadata").isEmpty()) { + long previous = semanticConsoleLookup.get(); + long now = System.nanoTime(); + if (now - previous >= SEMANTIC_CONSOLE_RETRY_NANOS + && semanticConsoleLookup.compareAndSet(previous, now)) { + dcr.resolveById("semantic-metadata"); + } + } JsonArray consoleIds = new JsonArray(); consoleIds.addAll(dcr.getConsoleIDs()); root.put("devConsoles", consoleIds); @@ -1909,9 +1925,12 @@ public class LocalCliConnector extends ServiceSupport @Override protected void doStop() throws Exception { ServiceHelper.stopService(transport); - if (semanticExecutor != null) { - semanticExecutor.shutdownNow(); - semanticExecutor = null; + // Stop the transport before taking this monitor: file dispatch may hold it while the transport stops. + synchronized (this) { + if (semanticExecutor != null) { + semanticExecutor.shutdownNow(); + semanticExecutor = null; + } } ServiceHelper.stopService(producer, consumer); } 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 new file mode 100644 index 000000000000..95f145ab412f --- /dev/null +++ b/dsl/camel-cli-connector/src/test/java/org/apache/camel/cli/connector/LocalCliConnectorSemanticLifecycleTest.java @@ -0,0 +1,144 @@ +/* + * 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.cli.connector; + +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.FutureTask; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import org.apache.camel.CamelContext; +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.util.json.JsonObject; +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.awaitility.Awaitility.await; + +class LocalCliConnectorSemanticLifecycleTest { + @Test + void stopClosesAnExecutorWhoseCreationIsAlreadyInProgress() throws Exception { + try (var context = new DefaultCamelContext()) { + var created = new CountDownLatch(1); + var release = new CountDownLatch(1); + var pool = new AtomicReference<ExecutorService>(); + context.setExecutorServiceManager(new DefaultExecutorServiceManager(context) { + @Override + public ExecutorService newThreadPool(Object source, String name, ThreadPoolProfile profile) { + ExecutorService executor = super.newThreadPool(source, name, profile); + if ("CliSemanticEvaluation".equals(name)) { + pool.set(executor); + created.countDown(); + try { + assertThat(release.await(10, TimeUnit.SECONDS)).isTrue(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new AssertionError(e); + } + } + return executor; + } + }); + var connector = startConnector(context); + FutureTask<CompletableFuture<Boolean>> dispatch = new FutureTask<>( + () -> connector.dispatchAsync(action(), ignored -> { + })); + FutureTask<Void> stop = new FutureTask<>(() -> { + connector.stop(); + return null; + }); + Thread caller = new Thread(dispatch); + Thread stopper = new Thread(stop); + caller.start(); + try { + assertThat(created.await(5, TimeUnit.SECONDS)).isTrue(); + stopper.start(); + // Stop must either finish or wait for the first executor to finish initializing. + await().atMost(5, TimeUnit.SECONDS) + .until(() -> stop.isDone() || stopper.getState() == Thread.State.BLOCKED); + release.countDown(); + stop.get(5, TimeUnit.SECONDS); + dispatch.get(5, TimeUnit.SECONDS).cancel(true); + assertThat(connector.isStopped()).isTrue(); + assertThat(pool.get().isShutdown()).isTrue(); + } finally { + release.countDown(); + caller.join(5000); + stopper.join(5000); + connector.stop(); + if (pool.get() != null) { + pool.get().shutdownNow(); + } + } + } + } + + @Test + void evaluationCannotCreateAnExecutorAfterStop() throws Exception { + try (var context = new DefaultCamelContext()) { + var connector = startConnector(context); + connector.stop(); + CompletableFuture<Boolean> result = connector.dispatchAsync(action(), ignored -> { + }); + assertThatThrownBy(() -> result.get(5, TimeUnit.SECONDS)) + .hasRootCauseInstanceOf(IllegalStateException.class) + .hasRootCauseMessage("CLI connector is stopping or stopped"); + } + } + + private static JsonObject action() { + return new JsonObject(Map.of("action", "semantic-evaluate")); + } + + private static LocalCliConnector startConnector(DefaultCamelContext context) { + var disabled = new DefaultCliConnectorFactory(); + disabled.setEnabled(false); + context.getCamelContextExtension().addContextPlugin(CliConnectorFactory.class, disabled); + context.start(); + var connector = new LocalCliConnector(new DefaultCliConnectorFactory()) { + @Override + protected CliConnectorTransport createTransport(String name) { + // As with WebSocket shutdown, stopping this transport does not join an in-flight dispatch. + return new CliConnectorTransport() { + @Override + public void configure( + CamelContext camelContext, CliActionDispatcher dispatcher, + CliSnapshotProducer snapshots, Runnable shutdown) { + } + + @Override + public void start() { + } + + @Override + public void stop() { + } + }; + } + }; + connector.setCamelContext(context); + connector.start(); + return connector; + } +} diff --git a/dsl/camel-cli-connector/src/test/java/org/apache/camel/cli/connector/LocalCliConnectorSemanticMetadataTest.java b/dsl/camel-cli-connector/src/test/java/org/apache/camel/cli/connector/LocalCliConnectorSemanticMetadataTest.java index 700c9427afb9..f5ea2f333135 100644 --- a/dsl/camel-cli-connector/src/test/java/org/apache/camel/cli/connector/LocalCliConnectorSemanticMetadataTest.java +++ b/dsl/camel-cli-connector/src/test/java/org/apache/camel/cli/connector/LocalCliConnectorSemanticMetadataTest.java @@ -18,16 +18,21 @@ package org.apache.camel.cli.connector; import java.util.List; import java.util.Map; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; +import org.apache.camel.console.DevConsole; import org.apache.camel.console.DevConsoleRegistry; import org.apache.camel.impl.DefaultCamelContext; +import org.apache.camel.impl.console.DefaultDevConsoleRegistry; import org.apache.camel.spi.CliConnectorFactory; import org.apache.camel.support.console.AbstractDevConsole; import org.apache.camel.util.json.JsonObject; import org.junit.jupiter.api.Test; import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; class LocalCliConnectorSemanticMetadataTest { @Test @@ -37,11 +42,25 @@ class LocalCliConnectorSemanticMetadataTest { var disabled = new DefaultCliConnectorFactory(); disabled.setEnabled(false); context.getCamelContextExtension().addContextPlugin(CliConnectorFactory.class, disabled); + var lookups = new AtomicInteger(); + var registry = new DefaultDevConsoleRegistry(context) { + @Override + public DevConsole resolveById(String id) { + if ("semantic-metadata".equals(id)) { + lookups.incrementAndGet(); + } + return super.resolveById(id); + } + }; + context.getCamelContextExtension().addContextPlugin(DevConsoleRegistry.class, registry); context.start(); var connector = new LocalCliConnector(new DefaultCliConnectorFactory()); connector.setCamelContext(context); assertThat(connector.status().<List<String>> getCollection("devConsoles")) .doesNotContain("semantic-metadata"); + int firstLookup = lookups.get(); + connector.status(); + assertThat(lookups).hasValue(firstLookup); context.getRegistry().bind("semantic-metadata-dev-console", new AbstractDevConsole("camel", "semantic-metadata", "Test", "Test") { @Override @@ -55,8 +74,11 @@ class LocalCliConnectorSemanticMetadataTest { } }); assertThat(DevConsoleRegistry.get(context).getConsoleIDs()).doesNotContain("semantic-metadata"); - assertThat(connector.status().<List<String>> getCollection("devConsoles")) - .contains("semantic-metadata"); + await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> assertThat( + connector.status().<List<String>> getCollection("devConsoles")).contains("semantic-metadata")); + int foundLookup = lookups.get(); + connector.status(); + assertThat(lookups).hasValue(foundLookup); } }
