gnodet-bot commented on code in PR #27602:
URL: https://github.com/apache/camel/pull/27602#discussion_r4229184911
##########
dsl/camel-cli-connector/src/main/java/org/apache/camel/cli/connector/LocalCliConnector.java:
##########
@@ -119,6 +125,7 @@ public class LocalCliConnector extends ServiceSupport
private String platformVersion;
private String mainClass;
private volatile ExecutorService terminateExecutor;
+ private ExecutorService semanticExecutor;
Review Comment:
⚠️ **Missing `volatile`:** `semanticExecutor` is written by the transport
thread (in `dispatchAsync`) and read by the stop thread (in `doStop`). Without
`volatile`, the JMM does not guarantee visibility between threads. The adjacent
`terminateExecutor` is already `volatile` — apply the same modifier here.
```suggestion
private volatile ExecutorService semanticExecutor;
```
##########
components/camel-ai/camel-semantic/src/main/java/org/apache/camel/semantic/SemanticEvaluateConsole.java:
##########
@@ -0,0 +1,202 @@
+/*
+ * 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.semantic;
+
+import java.util.Map;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import org.apache.camel.Expression;
+import org.apache.camel.builder.ThreadPoolBuilder;
+import org.apache.camel.language.semantic.SemanticLanguage;
+import org.apache.camel.spi.Metadata;
+import org.apache.camel.spi.annotations.DevConsole;
+import org.apache.camel.support.DefaultExchange;
+import org.apache.camel.support.console.AbstractDevConsole;
+import org.apache.camel.util.concurrent.ThreadPoolRejectedPolicy;
+import org.apache.camel.util.json.JsonRecordSupport;
+
+@DevConsole(name = "semantic-evaluate", displayName = "Semantic Evaluation",
+ description = "Evaluate sample input through a semantic definition
or expert operation", readOnly = false)
+public class SemanticEvaluateConsole extends AbstractDevConsole {
+ @Metadata(label = "query", description = "Name of the published evaluation
to execute", javaType = "java.lang.String")
+ public static final String EVALUATION = "evaluation";
+ @Metadata(label = "query", description = "Sample message body", javaType =
"java.lang.Object")
+ public static final String BODY = "body";
+ @Metadata(label = "query", description = "Sample message headers",
javaType = "java.util.Map")
+ public static final String HEADERS = "headers";
+ @Metadata(label = "query", description = "Sample exchange variables",
javaType = "java.util.Map")
+ public static final String VARIABLES = "variables";
+ @Metadata(label = "query", description = "Expert reference for direct
evaluation", javaType = "java.lang.String")
+ public static final String EXPERT = "expert";
+ @Metadata(label = "query", description = "Expert operation for direct
evaluation", javaType = "java.lang.String")
+ public static final String OPERATION = "operation";
+ @Metadata(label = "query", description = "Already-selected input for
direct evaluation", javaType = "java.lang.Object")
+ public static final String INPUT = "input";
+ @Metadata(label = "query", description = "Expert operation parameters",
javaType = "java.util.Map")
+ public static final String PARAMETERS = "parameters";
+
+ @Metadata(label = "query", description = "Maximum evaluation time in
milliseconds (1 to 50000, default 50000)",
+ javaType = "java.lang.Long")
+ public static final String TIMEOUT = "timeout";
+
+ private ExecutorService executor;
+
+ private synchronized ExecutorService executor() throws Exception {
+ if (executor == null) {
+ executor = new
ThreadPoolBuilder(getCamelContext()).poolSize(2).maxPoolSize(2).maxQueueSize(0)
+
.rejectedPolicy(ThreadPoolRejectedPolicy.Abort).build(this,
"SemanticEvaluation");
+ }
+ return executor;
+ }
+
+ @Override
+ protected synchronized void doStop() throws Exception {
+ if (executor != null) {
+ executor.shutdownNow();
+ executor = null;
+ }
+ super.doStop();
+ }
+
+ @Override
+ public Object call(MediaType mediaType, Map<String, Object> options) {
+ // This console bounds concurrent requests itself; do not queue
callers on AbstractDevConsole's lock.
+ return mediaType == MediaType.JSON ? doCallJson(options) :
doCallText(options);
+ }
+
+ public record Response(
+ @Metadata(description = "Evaluation name") String evaluation,
+ @Metadata(description = "Success or failed") String status,
+ @Metadata(description = "Validated result value") Object value,
+ @Metadata(description = "Provider probability") Double probability,
+ @Metadata(description = "Probabilities by label") Map<String,
Double> probabilities,
+ @Metadata(description = "Provider confidence") Double confidence,
+ @Metadata(description = "Provider result metadata") Map<String,
Object> metadata,
+ @Metadata(description = "Elapsed time in milliseconds") long
elapsedMillis,
+ @Metadata(description = "Evaluation error, if any") String error) {
+ }
+
+ public SemanticEvaluateConsole() {
+ super("camel", "semantic-evaluate", "Semantic Evaluation",
+ "Evaluate sample input through a semantic definition or expert
operation");
+ }
+
+ @Override
+ protected String doCallText(Map<String, Object> options) {
+ return doCallJson(options).toString();
+ }
+
+ @Override
+ protected Map<String, Object> doCallJson(Map<String, Object> options) {
+ long start = System.nanoTime();
+ Future<Map<String, Object>> task = null;
+ String failure;
+ try {
+ long timeout = optionLong(options, TIMEOUT, 50000);
+ if (timeout < 1 || timeout > 50000) {
+ throw new IllegalArgumentException("Evaluation timeout must be
between 1 and 50000 milliseconds");
+ }
+ task = executor().submit(() -> evaluate(options));
+ return task.get(timeout, TimeUnit.MILLISECONDS);
+ } catch (TimeoutException e) {
+ failure = "Evaluation timed out; cancellation requested";
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ failure = "Evaluation cancelled";
+ } catch (RejectedExecutionException e) {
+ failure = "Busy: semantic evaluations are already running";
+ } catch (Exception e) {
+ failure = e.getMessage();
Review Comment:
⚠️ **`e.getMessage()` can return `null`:** If the exception has no message,
`failure` will be `null`, and the `Response` will be built with `error=null` —
silently swallowing the failure cause. The same PR handles this correctly in
`WebSocketCliConnectorTransport`. Apply the same guard here:
```suggestion
failure = e.getMessage() != null ? e.getMessage() :
e.getClass().getSimpleName();
```
--
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]
