This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch CAMEL-26911-send-thread in repository https://gitbox.apache.org/repos/asf/camel.git
commit 4c5464da211921e8f747204c2747ecd17193ec51 Author: Claus Ibsen <[email protected]> AuthorDate: Fri Nov 13 13:25:30 2026 +0100 CAMEL-26911: camel-cli-connector - a send runs on a thread of its own, so a poll or an InOut send that waits does not hold up camel stop and the status Co-Authored-By: Claude Opus 5.5 <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../camel/cli/connector/LocalCliConnector.java | 82 ++++++++++++++-------- .../LocalCliConnectorFileTransportTest.java | 45 ++++++++++++ 2 files changed, 99 insertions(+), 28 deletions(-) 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 ef4d34f672d2..78f185e0362b 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 @@ -39,6 +39,7 @@ import java.util.Locale; import java.util.Map; import java.util.Objects; import java.util.Optional; +import java.util.concurrent.Callable; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @@ -130,6 +131,7 @@ public class LocalCliConnector extends ServiceSupport private ExecutorService semanticExecutor; private ExecutorService auditExecutor; private ExecutorService sendExecutor; + private ExecutorService singleSendExecutor; private final AtomicLong semanticConsoleLookup = new AtomicLong(System.nanoTime() - SEMANTIC_CONSOLE_RETRY_NANOS); private ProducerTemplate producer; private ConsumerTemplate consumer; @@ -245,8 +247,8 @@ public class LocalCliConnector extends ServiceSupport @Override public CompletableFuture<Boolean> dispatchAsync(JsonObject root, CliActionOutput output) { - if ("send".equals(root.getString("action")) && RepeatedSend.isRepeated(root)) { - return repeatedSend(root, output); + if ("send".equals(root.getString("action"))) { + return RepeatedSend.isRepeated(root) ? repeatedSend(root, output) : singleSend(root, output); } boolean auditQuery = "semantic-audit".equals(root.getString("action")); if (!auditQuery && !"semantic-evaluate".equals(root.getString("action"))) { @@ -301,11 +303,38 @@ public class LocalCliConnector extends ServiceSupport * status and the other actions go on meanwhile. The transport cancels it when the client gives up the request. */ private CompletableFuture<Boolean> repeatedSend(JsonObject root, CliActionOutput output) { + return send(output, () -> repeatedSendResult(root), () -> { + if (sendExecutor == null) { + sendExecutor = new ThreadPoolBuilder(camelContext).poolSize(2).maxPoolSize(2).maxQueueSize(0) + .rejectedPolicy(ThreadPoolRejectedPolicy.Abort).build("cli-send", "CliRepeatedSend"); + } + return sendExecutor; + }, "Busy: two repeated sends are already running"); + } + + /** + * The send action of one message, on a thread of its own, as an InOut send or a poll waits for the reply or the + * message (up to the timeout of the client), and the connector thread must go on writing the status and looking for + * the lock file, which camel stop deletes (CAMEL-26911). One at a time and in order, as they were on the connector + * thread. + */ + private CompletableFuture<Boolean> singleSend(JsonObject root, CliActionOutput output) { + return send(output, () -> sendResult(root), () -> { + if (singleSendExecutor == null) { + singleSendExecutor = new ThreadPoolBuilder(camelContext).poolSize(1).maxPoolSize(1).maxQueueSize(100) + .rejectedPolicy(ThreadPoolRejectedPolicy.Abort).build("cli-send", "CliSend"); + } + return singleSendExecutor; + }, "Busy: 100 sends are already waiting"); + } + + private CompletableFuture<Boolean> send( + CliActionOutput output, Callable<JsonObject> send, Callable<ExecutorService> executor, String busy) { CompletableFuture<Boolean> result = new CompletableFuture<>(); FutureTask<Void> task = new FutureTask<>(() -> { try { // capture this request's output; never use the dispatcher's mutable actionOutput from a worker - output.write(repeatedSendResult(root)); + output.write(send.call()); result.complete(true); } catch (InterruptedException e) { Thread.currentThread().interrupt(); @@ -325,14 +354,10 @@ public class LocalCliConnector extends ServiceSupport if (isStoppingOrStopped()) { throw new IllegalStateException("CLI connector is stopping or stopped"); } - if (sendExecutor == null) { - sendExecutor = new ThreadPoolBuilder(camelContext).poolSize(2).maxPoolSize(2).maxQueueSize(0) - .rejectedPolicy(ThreadPoolRejectedPolicy.Abort).build("cli-send", "CliRepeatedSend"); - } - sendExecutor.execute(task); + executor.call().execute(task); } - } catch (RejectedExecutionException busy) { - result.completeExceptionally(new IllegalStateException("Busy: two repeated sends are already running")); + } catch (RejectedExecutionException e) { + result.completeExceptionally(new IllegalStateException(busy)); } catch (Exception e) { result.completeExceptionally(e); } @@ -725,28 +750,25 @@ public class LocalCliConnector extends ServiceSupport } private void doActionSendTask(JsonObject root) throws Exception { + // a transport that dispatches on its own thread, rather than with dispatchAsync + writeOutput(RepeatedSend.isRepeated(root) ? repeatedSendResult(root) : sendResult(root)); + } + + private JsonObject sendResult(JsonObject root) throws Exception { DevConsole dc = camelContext.getCamelContextExtension().getContextPlugin(DevConsoleRegistry.class) .resolveById("send"); - if (dc != null) { - - if (RepeatedSend.isRepeated(root)) { - // a transport that dispatches on its own thread, rather than with dispatchAsync - writeOutput(repeatedSendResult(root)); - return; - } - waitUntilStarted(); + if (dc == null) { + return new JsonObject(); + } + waitUntilStarted(); - final Map<String, Object> args = sendOptions(root); - String body = root.getString("body"); - if (body != null) { - args.put("body", body); - } - args.putAll(sendHeaders(root)); - JsonObject json = (JsonObject) dc.call(DevConsole.MediaType.JSON, args); - writeOutput(json); - } else { - writeOutput(new JsonObject()); + final Map<String, Object> args = sendOptions(root); + String body = root.getString("body"); + if (body != null) { + args.put("body", body); } + args.putAll(sendHeaders(root)); + return (JsonObject) dc.call(DevConsole.MediaType.JSON, args); } private void waitUntilStarted() throws InterruptedException { @@ -2088,6 +2110,10 @@ public class LocalCliConnector extends ServiceSupport camelContext.getExecutorServiceManager().shutdownNow(sendExecutor); sendExecutor = null; } + if (singleSendExecutor != null) { + camelContext.getExecutorServiceManager().shutdownNow(singleSendExecutor); + singleSendExecutor = null; + } } ServiceHelper.stopService(producer, consumer); } diff --git a/dsl/camel-cli-connector/src/test/java/org/apache/camel/cli/connector/LocalCliConnectorFileTransportTest.java b/dsl/camel-cli-connector/src/test/java/org/apache/camel/cli/connector/LocalCliConnectorFileTransportTest.java index bb273620bab4..a78918b38095 100644 --- a/dsl/camel-cli-connector/src/test/java/org/apache/camel/cli/connector/LocalCliConnectorFileTransportTest.java +++ b/dsl/camel-cli-connector/src/test/java/org/apache/camel/cli/connector/LocalCliConnectorFileTransportTest.java @@ -24,6 +24,7 @@ import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.UUID; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; @@ -63,6 +64,8 @@ class LocalCliConnectorFileTransportTest extends CamelTestSupport { private File dir; private LocalCliConnector connector; private final AtomicInteger sigterms = new AtomicInteger(); + private final CountDownLatch slowEntered = new CountDownLatch(1); + private final CountDownLatch slowRelease = new CountDownLatch(1); @Override protected boolean useJmx() { @@ -89,6 +92,12 @@ class LocalCliConnectorFileTransportTest extends CamelTestSupport { public void configure() { from("direct:hello").routeId("hello").setBody(simple("Hello ${body}")); from("direct:other").routeId("other").to("mock:other"); + from("direct:slow").routeId("slow") + .process(e -> { + slowEntered.countDown(); + slowRelease.await(20, TimeUnit.SECONDS); + }) + .setBody(simple("Slow ${body}")); } }; } @@ -217,6 +226,42 @@ class LocalCliConnectorFileTransportTest extends CamelTestSupport { assertThat(out.toJson()).contains("Hello World"); } + @Test + void slowSendDoesNotBlockTheStatusNorStop() throws Exception { + startConnector(); + try { + writeAction(action("send", "endpoint", "direct:slow", "body", "World", "exchangePattern", "InOut"), "slow"); + assertThat(slowEntered.await(5, TimeUnit.SECONDS)).isTrue(); + + // the send waits for its reply, and meanwhile the other actions and the status go on + writeAction(action("route", "command", "stop", "id", "other"), "fast"); + await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> assertThat( + routeState(readJson(file(PID + "-status.json")), "other")).isEqualTo("Stopped")); + + // as does camel stop, which deletes the lock file + Files.delete(file(PID).toPath()); + await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> assertThat(sigterms).hasValue(1)); + } finally { + slowRelease.countDown(); + } + } + + @Test + void sendsAreAnsweredInOrderOnceTheSlowOneIsDone() throws Exception { + startConnector(); + try { + writeAction(action("send", "endpoint", "direct:slow", "body", "first", "exchangePattern", "InOut"), "a"); + assertThat(slowEntered.await(5, TimeUnit.SECONDS)).isTrue(); + writeAction(action("send", "endpoint", "direct:hello", "body", "second", "exchangePattern", "InOut"), "b"); + await().during(500, TimeUnit.MILLISECONDS).atMost(2, TimeUnit.SECONDS) + .untilAsserted(() -> assertThat(file(PID + "-output-b.json")).doesNotExist()); + } finally { + slowRelease.countDown(); + } + assertThat(awaitJson(file(PID + "-output-a.json")).toJson()).contains("Slow first"); + assertThat(awaitJson(file(PID + "-output-b.json")).toJson()).contains("Hello second"); + } + @Test void actionsWithRequestIdWriteToTheirOwnOutputFile() { startConnector();
