This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 5dd7712722c8 CAMEL-26929: camel-console - the send console bounds the
send by its timeout, and camel cmd send --timeout is taken as given (#29935)
5dd7712722c8 is described below
commit 5dd7712722c83341ae1150d82508910b942c2500
Author: Claus Ibsen <[email protected]>
AuthorDate: Sat Nov 14 20:50:59 2026 +0100
CAMEL-26929: camel-console - the send console bounds the send by its
timeout, and camel cmd send --timeout is taken as given (#29935)
The timeout of the send console was for a poll only, so a send to
direct:nope waited the 30
seconds of direct for a consumer and the CLI gave up first (exit 124). The
send is now bounded
by the timeout and answers with status timeout, a direct endpoint that no
route consumes fails
at once, and camel cmd send no longer raises a timeout below 5 seconds to 5
seconds.
CAMEL-26929: camel-console - roomier timing bounds in SendDevConsoleTest,
and a negative --timeout is a usage error too
CAMEL-26929: camel-console - an interrupted send keeps the interrupt flag
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../apache/camel/catalog/dev-consoles-openapi.json | 2 +-
.../apache/camel/catalog/dev-consoles/send.json | 2 +-
.../org/apache/camel/dev-console/send.json | 2 +-
.../apache/camel/impl/console/SendDevConsole.java | 87 +++++++++++++++++-----
.../camel/impl/console/SendDevConsoleTest.java | 82 ++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 4 +
.../core/commands/action/CamelSendAction.java | 7 +-
.../commands/action/CamelSendActionResultTest.java | 21 +++++-
8 files changed, 182 insertions(+), 25 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/dev-consoles-openapi.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/dev-consoles-openapi.json
index 71c40d466b4a..56f159f76514 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/dev-consoles-openapi.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/dev-consoles-openapi.json
@@ -8223,7 +8223,7 @@
"pollTimeout": {
"type": "integer",
"default": 20000,
-
"description": "Timeout when using poll mode"
+
"description": "Timeout of the send, or of the poll in poll mode, in millis"
}
}
}
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/dev-consoles/send.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/dev-consoles/send.json
index cb8dd24d3c59..39513b204c46 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/dev-consoles/send.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/dev-consoles/send.json
@@ -118,7 +118,7 @@
"autowired": false,
"secret": false,
"defaultValue": 20000,
- "description": "Timeout when using poll mode"
+ "description": "Timeout of the send, or of the poll in poll mode, in
millis"
}
},
"response": {
diff --git
a/core/camel-console/src/generated/resources/META-INF/org/apache/camel/dev-console/send.json
b/core/camel-console/src/generated/resources/META-INF/org/apache/camel/dev-console/send.json
index cb8dd24d3c59..39513b204c46 100644
---
a/core/camel-console/src/generated/resources/META-INF/org/apache/camel/dev-console/send.json
+++
b/core/camel-console/src/generated/resources/META-INF/org/apache/camel/dev-console/send.json
@@ -118,7 +118,7 @@
"autowired": false,
"secret": false,
"defaultValue": 20000,
- "description": "Timeout when using poll mode"
+ "description": "Timeout of the send, or of the poll in poll mode, in
millis"
}
},
"response": {
diff --git
a/core/camel-console/src/main/java/org/apache/camel/impl/console/SendDevConsole.java
b/core/camel-console/src/main/java/org/apache/camel/impl/console/SendDevConsole.java
index 0ae6c8e82ef5..0729a0a4ffb8 100644
---
a/core/camel-console/src/main/java/org/apache/camel/impl/console/SendDevConsole.java
+++
b/core/camel-console/src/main/java/org/apache/camel/impl/console/SendDevConsole.java
@@ -22,7 +22,12 @@ import java.util.Base64;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import org.apache.camel.CamelExchangeException;
import org.apache.camel.ConsumerTemplate;
import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
@@ -39,6 +44,7 @@ import org.apache.camel.support.MessageHelper;
import org.apache.camel.support.console.AbstractDevConsole;
import org.apache.camel.support.service.ServiceHelper;
import org.apache.camel.util.StopWatch;
+import org.apache.camel.util.StringHelper;
import org.apache.camel.util.TimeUtils;
import org.apache.camel.util.json.JsonRecordSupport;
@@ -64,7 +70,7 @@ public class SendDevConsole extends AbstractDevConsole {
private int bodyMaxChars = 32 * 1024;
@Metadata(defaultValue = "20000", label = "advanced",
- description = "Timeout when using poll mode")
+ description = "Timeout of the send, or of the poll in poll mode,
in millis")
private int pollTimeout = 20000;
@Metadata(label = "query", description = "Maximum size of the message body
to include in the dump",
@@ -84,7 +90,7 @@ public class SendDevConsole extends AbstractDevConsole {
javaType = "java.lang.Boolean")
public static final String POLL = "poll";
- @Metadata(label = "query", description = "Timeout when using poll mode",
+ @Metadata(label = "query", description = "Timeout of the send, or of the
poll in poll mode, in millis",
javaType = "java.lang.Integer", defaultValue = "20000")
public static final String POLL_TIMEOUT = "pollTimeout";
@@ -141,9 +147,9 @@ public class SendDevConsole extends AbstractDevConsole {
}
String exchangePattern = optionString(options, EXCHANGE_PATTERN);
boolean poll = optionBoolean(options, POLL, false);
+ // bounds the send as well as the poll: a send to direct:nope waited
for the 30 seconds of direct, longer than
+ // the CLI waits for the answer (CAMEL-26929)
int timeout = optionInt(options, POLL_TIMEOUT, pollTimeout);
- // give extra time as CLI needs to process reply also
- timeout += 5000;
Endpoint target = null;
Exchange out = null;
@@ -162,7 +168,9 @@ public class SendDevConsole extends AbstractDevConsole {
}
long taken = watch.taken();
String status = "success";
- if (cause != null) {
+ if (cause instanceof SendTimeoutException) {
+ status = "timeout";
+ } else if (cause != null) {
status = "error";
} else if (poll && out == null) {
status = "timeout";
@@ -206,9 +214,9 @@ public class SendDevConsole extends AbstractDevConsole {
}
String exchangePattern = optionString(options, EXCHANGE_PATTERN);
boolean poll = optionBoolean(options, POLL, false);
+ // bounds the send as well as the poll: a send to direct:nope waited
for the 30 seconds of direct, longer than
+ // the CLI waits for the answer (CAMEL-26929)
int timeout = optionInt(options, POLL_TIMEOUT, pollTimeout);
- // give extra time as CLI needs to process reply also
- timeout += 5000;
Endpoint target = null;
Exchange out = null;
@@ -227,7 +235,9 @@ public class SendDevConsole extends AbstractDevConsole {
}
long taken = watch.taken();
String status = "success";
- if (cause != null) {
+ if (cause instanceof SendTimeoutException) {
+ status = "timeout";
+ } else if (cause != null) {
status = "error";
} else if (poll && out == null) {
status = "timeout";
@@ -270,20 +280,63 @@ public class SendDevConsole extends AbstractDevConsole {
if (poll) {
out = consumer.receive(target, timeout);
} else {
- final String mep = exchangePattern;
- out = producer.send(target, exchange -> {
- exchange.getMessage().setBody(inputBody);
- if (!inputHeaders.isEmpty()) {
- exchange.getMessage().setHeaders(inputHeaders);
- }
- exchange.setPattern(
- "InOut".equals(mep) ? ExchangePattern.InOut :
ExchangePattern.InOnly);
- });
+ Exchange exchange = target.createExchange(
+ "InOut".equals(exchangePattern) ?
ExchangePattern.InOut : ExchangePattern.InOnly);
+ exchange.getMessage().setBody(inputBody);
+ if (!inputHeaders.isEmpty()) {
+ exchange.getMessage().setHeaders(inputHeaders);
+ }
+ if (isDirectWithoutConsumer(target)) {
+ // direct waits 30 seconds for a consumer by default: no
route will start to consume it now
+ throw new CamelExchangeException("No consumers available
on endpoint: " + target, exchange);
+ }
+ CompletableFuture<Exchange> future =
producer.asyncSend(target, exchange);
+ try {
+ out = future.get(timeout, TimeUnit.MILLISECONDS);
+ } catch (TimeoutException e) {
+ // the exchange goes on in the integration, but the caller
has its answer
+ throw new SendTimeoutException(exchange, timeout);
+ } catch (ExecutionException e) {
+ throw e.getCause() instanceof Exception ex ? ex : e;
+ } catch (InterruptedException e) {
+ // keep the interrupt for the caller, such as a console
call that is stopped
+ Thread.currentThread().interrupt();
+ throw new CamelExchangeException("The send was
interrupted", exchange, e);
+ }
}
}
return out;
}
+ /**
+ * Whether the endpoint is a direct endpoint that no route consumes from.
+ */
+ private boolean isDirectWithoutConsumer(Endpoint target) {
+ String key = target.getEndpointKey();
+ if (key == null || !key.startsWith("direct:")) {
+ return false;
+ }
+ String path = StringHelper.before(key, "?", key);
+ for (Route route : getCamelContext().getRoutes()) {
+ String consumer = route.getEndpoint().getEndpointKey();
+ if (consumer != null && path.equals(StringHelper.before(consumer,
"?", consumer))) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ /**
+ * The send did not complete within the timeout.
+ */
+ private static final class SendTimeoutException extends
CamelExchangeException {
+
+ SendTimeoutException(Exchange exchange, long timeout) {
+ super("The send did not complete within " + timeout + " millis
(the exchange goes on in the integration)",
+ exchange);
+ }
+ }
+
/**
* Why nothing was sent: no endpoint matches, or, without an endpoint,
there is no route to send to (before, that
* was reported as sent).
diff --git
a/core/camel-console/src/test/java/org/apache/camel/impl/console/SendDevConsoleTest.java
b/core/camel-console/src/test/java/org/apache/camel/impl/console/SendDevConsoleTest.java
index 331046d4bbda..a640add52233 100644
---
a/core/camel-console/src/test/java/org/apache/camel/impl/console/SendDevConsoleTest.java
+++
b/core/camel-console/src/test/java/org/apache/camel/impl/console/SendDevConsoleTest.java
@@ -54,6 +54,9 @@ public class SendDevConsoleTest extends
AbstractDevConsoleTest {
public void configure() {
from("direct:start").routeId("myRoute")
.to("mock:result");
+ from("direct:slow").routeId("slowRoute")
+ .delay(5000)
+ .to("mock:slow");
}
};
}
@@ -72,6 +75,85 @@ public class SendDevConsoleTest extends
AbstractDevConsoleTest {
assertNotNull(jsonOut.getString("endpoint"));
}
+ @Test
+ public void testSendIsBoundedByTheTimeout() {
+ // the timeout was for --poll only: the send waited as long as the
route took (CAMEL-26929)
+ DevConsole console = startedSendConsole();
+
+ Map<String, Object> options = createOptions("direct:slow", "Hello");
+ options.put(SendDevConsole.POLL_TIMEOUT, "200");
+ JsonObject out = callJson(console, options);
+
+ assertEquals("timeout", out.getString("status"), out.toJson());
+ // well under the 5 seconds of the route, with room for a slow CI
machine
+ assertTrue(out.getLong("elapsed") < 4000, "answered at the timeout: "
+ out.toJson());
+ JsonObject exception = map(out, "exception");
+ assertNotNull(exception, out.toJson());
+ assertTrue(exception.getString("message").startsWith("The send did not
complete within 200 millis"),
+ exception.toJson());
+ }
+
+ @Test
+ public void testAnInterruptedSendKeepsTheInterrupt() {
+ DevConsole console = startedSendConsole();
+
+ Map<String, Object> options = createOptions("direct:slow", "Hello");
+ options.put(SendDevConsole.POLL_TIMEOUT, "20000");
+ Thread.currentThread().interrupt();
+ JsonObject out;
+ boolean interrupted;
+ try {
+ out = callJson(console, options);
+ } finally {
+ // clears the flag for the tests after this one
+ interrupted = Thread.interrupted();
+ }
+
+ assertTrue(interrupted, "the interrupt is kept for the caller");
+ assertEquals("error", out.getString("status"), out.toJson());
+ assertTrue(map(out, "exception").getString("message").startsWith("The
send was interrupted"), out.toJson());
+ }
+
+ @Test
+ public void testSendCompletesWithinTheTimeout() {
+ DevConsole console = startedSendConsole();
+
+ Map<String, Object> options = createOptions("direct:start", "Hello");
+ options.put(SendDevConsole.EXCHANGE_PATTERN, "InOut");
+ options.put(SendDevConsole.POLL_TIMEOUT, "5000");
+ JsonObject out = callJson(console, options);
+
+ assertEquals("success", out.getString("status"), out.toJson());
+ assertEquals("Hello", map(map(out, "message"),
"body").getString("value"), out.toJson());
+ }
+
+ @Test
+ public void testSendToDirectWithoutConsumerFailsAtOnce() {
+ // direct waits 30 seconds for a consumer by default, longer than the
CLI waits for the answer (CAMEL-26929)
+ DevConsole console = startedSendConsole();
+
+ Map<String, Object> options = createOptions("direct:nope", "Hello");
+ JsonObject out = callJson(console, options);
+
+ assertEquals("error", out.getString("status"), out.toJson());
+ // well under the 30 seconds direct waits
+ assertTrue(out.getLong("elapsed") < 10000, out.toJson());
+ assertTrue(map(out, "exception").getString("message").startsWith("No
consumers available on endpoint: direct://nope"),
+ out.toJson());
+
+ // a direct endpoint with options of its own is the endpoint of its
route
+ options = createOptions("direct:start?timeout=1000", "Hello");
+ assertEquals("success", callJson(console,
options).getString("status"));
+ }
+
+ private static JsonObject map(JsonObject jo, String key) {
+ Map<?, ?> map = (Map<?, ?>) jo.get(key);
+ assertNotNull(map, key + " in " + jo.toJson());
+ JsonObject answer = new JsonObject();
+ map.forEach((k, v) -> answer.put(String.valueOf(k), v));
+ return answer;
+ }
+
@Test
public void testSendConsoleInOutPattern() {
DevConsole console = assertConsoleExists("send");
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index 105b6ed66006..f8d465a3ff3c 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -2821,6 +2821,10 @@ A script that checks the exit code of a command must
take these changes into acc
cannot be generated and `70` for an unexpected error, instead of `4`, `2`,
`3` and `4`.
* `camel transform dataweave` exits with `2` without `--input` or
`--expression`, and with `66` when the input path does
not exist or holds no `.dwl` files, instead of `1`.
+* `camel cmd send --timeout` bounds the send in the integration: a send that
does not complete in time is `Timeout`
+ with exit `124`, and the exchange goes on in the integration. Before, it
only bounded how long the CLI waited, and a
+ timeout below 5 seconds was raised to 5 seconds. A send to a `direct`
endpoint that no route consumes fails at once
+ with "No consumers available" and exit `1`, instead of waiting 30 seconds
for a consumer.
* `camel cmd sql` exits with `124` instead of `1` when the query does not
answer in time.
* A source file of `camel cmd route-diagram`, `camel cmd route-structure` and
`camel cmd route-topology` that does not
exist exits with `66` instead of `1`. The parts of a Java route the parser
could not read are printed as a warning on
diff --git
a/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/action/CamelSendAction.java
b/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/action/CamelSendAction.java
index 031bc12acdbf..94e77bb6c816 100644
---
a/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/action/CamelSendAction.java
+++
b/dsl/camel-jbang/camel-jbang-core/src/main/java/org/apache/camel/dsl/jbang/core/commands/action/CamelSendAction.java
@@ -192,9 +192,10 @@ public class CamelSendAction extends ActionBaseCommand {
}
}
}
- // timeout cannot be too low
- if (timeout < 5000) {
- timeout = 5000;
+ // the integration bounds the send by the timeout as given: a timeout
below 5 seconds was raised to 5 seconds
+ // without a word, while the send itself was not bounded at all
(CAMEL-26929)
+ if (timeout <= 0) {
+ throw CliException.usage("--timeout must be more than 0, was: " +
timeout);
}
if (repeat < 0 || repeat > MAX_MESSAGES) {
throw CliException.usage("--repeat must be between 1 and " +
MAX_MESSAGES + ", was: " + repeat);
diff --git
a/dsl/camel-jbang/camel-jbang-core/src/test/java/org/apache/camel/dsl/jbang/core/commands/action/CamelSendActionResultTest.java
b/dsl/camel-jbang/camel-jbang-core/src/test/java/org/apache/camel/dsl/jbang/core/commands/action/CamelSendActionResultTest.java
index 3e2ca538658e..3f30a46e1dd2 100644
---
a/dsl/camel-jbang/camel-jbang-core/src/test/java/org/apache/camel/dsl/jbang/core/commands/action/CamelSendActionResultTest.java
+++
b/dsl/camel-jbang/camel-jbang-core/src/test/java/org/apache/camel/dsl/jbang/core/commands/action/CamelSendActionResultTest.java
@@ -58,14 +58,31 @@ class CamelSendActionResultTest extends
ActionCommandTestSupport {
assertThat(action.getString("endpoint")).isEqualTo("direct:start");
assertThat(action.getString("body")).isEqualTo("hello");
assertThat(action.getString("exchangePattern")).isEqualTo("InOnly");
- // the timeout is at least 5 seconds
- assertThat(action.getLong("pollTimeout")).isEqualTo(5000L);
+ // the timeout as given: it was raised to 5 seconds without a word,
and the integration bounds the send by it
+ // (CAMEL-26929)
+ assertThat(action.getLong("pollTimeout")).isEqualTo(100L);
JsonArray headers = action.getCollection("headers");
JsonObject header = headers.getMap(0);
assertThat(header.getString("key")).isEqualTo("id");
assertThat(header.getString("value")).isEqualTo("1");
}
+ @Test
+ void aTimeoutOfZeroIsAUsageError() throws Exception {
+ writeStatusFile(TEST_PID, "orders");
+
+ assertThatThrownBy(() -> callWithResponse(command("orders",
"--endpoint=direct:start", "--timeout=0"),
+ answer("success")))
+ .isInstanceOfSatisfying(CliException.class,
+ e ->
assertThat(e.getExitCode()).isEqualTo(CliExitCode.USAGE))
+ .hasMessage("--timeout must be more than 0, was: 0");
+ assertThatThrownBy(() -> callWithResponse(command("orders",
"--endpoint=direct:start", "--timeout=-1"),
+ answer("success")))
+ .isInstanceOfSatisfying(CliException.class,
+ e ->
assertThat(e.getExitCode()).isEqualTo(CliExitCode.USAGE))
+ .hasMessage("--timeout must be more than 0, was: -1");
+ }
+
@Test
void failedAndTimedOutExchanges() throws Exception {
writeStatusFile(TEST_PID, "orders");
