luigidemasi commented on code in PR #27602:
URL: https://github.com/apache/camel/pull/27602#discussion_r4229681999
##########
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:
Fixed in 68714d30c1c4. All accesses to `semanticExecutor` now use the same
monitor, including lazy construction, submission, and shutdown. Dispatch also
checks the connector lifecycle under that monitor and rejects calls once
stopping begins. This provides visibility and closes an additional race:
WebSocket shutdown does not join an in-flight dispatch, so first-use executor
creation could finish after connector shutdown. I reproduced that interleaving
with `volatile` alone as well. The transport is stopped before acquiring the
monitor to avoid deadlocking file dispatch. Two regression tests cover shutdown
during initialization and calls after stop; both fail on the previous
implementation and pass with the fix.
_OpenAI Codex on behalf of @luigidemasi, via /oss-address-review._
##########
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:
Fixed in 68714d30c1c4 using the suggested exception-class fallback. Applied
it to both catches: the submit/wait path and the inner provider-evaluation
path. A null message previously caused the JSON serializer to omit `error`
entirely, although `status` remained `failed`. Added separate regressions for a
message-less executor-factory exception and a message-less provider exception;
both fail on the previous implementation and pass with the fix.
_OpenAI Codex on behalf of @luigidemasi, via /oss-address-review._
--
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]
