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 f304ad2cfc49 CAMEL-26451: camel-core - Routing Slip and Dynamic Router
fire the sent event of a step before the next step (#28807)
f304ad2cfc49 is described below
commit f304ad2cfc499e3739092f5d91f289704f7b19ee
Author: allthingssecurity <[email protected]>
AuthorDate: Tue Nov 3 13:25:26 2026 +0530
CAMEL-26451: camel-core - Routing Slip and Dynamic Router fire the sent
event of a step before the next step (#28807)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../org/apache/camel/processor/RoutingSlip.java | 176 +++++++++++----------
.../DynamicRouterEventNotifierTest.java | 15 ++
.../routingslip/RoutingSlipEventNotifierTest.java | 15 ++
.../routingslip/RoutingSlipSentEventOrderTest.java | 163 +++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 13 ++
5 files changed, 296 insertions(+), 86 deletions(-)
diff --git
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/RoutingSlip.java
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/RoutingSlip.java
index 304f4884fc81..0eb1abbce619 100644
---
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/RoutingSlip.java
+++
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/RoutingSlip.java
@@ -402,13 +402,11 @@ public class RoutingSlip extends BaseProcessorSupport
implements Traceable, IdAw
LOG.trace("Processing exchangeId: {} >>> {}",
exchange.getExchangeId(), exchange);
}
- // routing slip callback which are used when
- // - routing slip was routed asynchronously
- // - and we are completely done with the routing slip
- // so we need to signal done on the original callback so it can
continue
+ // routing slip callback which is used when the step was routed
asynchronously: the step has then been sent (its
+ // ExchangeSentEvent fired) and its producer released, so continue
with the remaining steps of the routing slip
AsyncCallback callback = doneSync -> {
if (!doneSync) {
- originalCallback.done(false);
+ continueRoutingSlip(exchange, endpoint, original,
originalCallback, iter);
}
};
return producerCache.doInAsyncProducer(endpoint, exchange, callback,
(p, ex, cb) -> {
@@ -443,94 +441,100 @@ public class RoutingSlip extends BaseProcessorSupport
implements Traceable, IdAw
ServiceHelper.stopAndShutdownService(endpoint);
}
- // we only have to handle async completion of the routing
slip
- if (doneSync) {
- cb.done(true);
- return;
- }
+ // the step is done: this fires its ExchangeSentEvent and
releases the producer before the next step
+ // (when the step completed asynchronously, the callback
above then continues the routing slip)
+ cb.done(doneSync);
+ }
+ });
+ });
+ }
- try {
- // continue processing the routing slip asynchronously
- Exchange current = prepareExchangeForRoutingSlip(ex,
endpoint);
-
- while (iter.hasNext(current)) {
-
- // we ignore some kind of exceptions and allow us
to continue
- if (isIgnoreInvalidEndpoints()) {
- FailedToCreateProducerException e =
current.getException(FailedToCreateProducerException.class);
- if (e != null) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Endpoint uri is invalid:
{}. This exception will be ignored.", endpoint,
- e);
- }
- current.setException(null);
- }
- }
-
- // Decide whether to continue with the recipients
or not; similar logic to the Pipeline
- // check for error if so we should break out
- if (!continueProcessing(current, "so breaking out
of the routing slip", LOG)) {
- break;
- }
-
- Endpoint nextEndpoint;
- boolean prototype = cacheSize < 0;
- try {
- Object recipient = iter.next(ex);
- recipient = prepareRecipient(exchange,
recipient);
- // enforce the optional allowed-schemes
allow-list on the resolved dynamic recipient (CAMEL-24298)
-
ProcessorHelper.checkAllowedSchemes(allowedSchemes, recipient);
- Endpoint existing =
getExistingEndpoint(exchange, recipient);
- if (existing == null) {
- nextEndpoint = resolveEndpoint(exchange,
recipient, prototype);
- } else {
- nextEndpoint = existing;
- // we have an existing endpoint then its
not a prototype scope
- prototype = false;
- }
- // if no endpoint was resolved then try the
next
- if (nextEndpoint == null) {
- continue;
- }
- } catch (Exception e) {
- // error resolving endpoint so we should break
out
- current.setException(e);
- break;
- }
-
- // prepare and process the routing slip
- final AsyncCallback cbNext =
getNextCallback(prototype, nextEndpoint, cb);
- boolean sync = processExchange(nextEndpoint,
current, original, cbNext, iter, prototype);
- current = prepareExchangeForRoutingSlip(current,
nextEndpoint);
-
- if (!sync) {
- if (LOG.isTraceEnabled()) {
- LOG.trace("Processing exchangeId: {} is
continued being processed asynchronously",
- original.getExchangeId());
- }
- return;
- }
- }
+ /**
+ * Continues the routing slip with its remaining steps after a step (on
stepExchange) that completed asynchronously
+ * and has been sent (its ExchangeSentEvent fired and its producer
released), and signals the original callback when
+ * the routing slip is completely done (unless a remaining step completes
asynchronously as well, which then
+ * continues in the same way).
+ */
+ private void continueRoutingSlip(
+ final Exchange stepExchange, final Endpoint endpoint, final
Exchange original, final AsyncCallback originalCallback,
+ final RoutingSlipIterator iter) {
+ try {
+ // continue processing the routing slip asynchronously
+ Exchange current = prepareExchangeForRoutingSlip(stepExchange,
endpoint);
- // logging nextExchange as it contains the exchange
that might have altered the payload and since
- // we are logging the completion if will be confusing
if we log the original instead
- // we could also consider logging the original and the
nextExchange then we have *before* and *after* snapshots
- if (LOG.isTraceEnabled()) {
- LOG.trace("Processing complete for exchangeId: {}
>>> {}", original.getExchangeId(), current);
+ while (iter.hasNext(current)) {
+
+ // we ignore some kind of exceptions and allow us to continue
+ if (isIgnoreInvalidEndpoints()) {
+ FailedToCreateProducerException e =
current.getException(FailedToCreateProducerException.class);
+ if (e != null) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Endpoint uri is invalid: {}. This
exception will be ignored.", endpoint, e);
}
+ current.setException(null);
+ }
+ }
+
+ // Decide whether to continue with the recipients or not;
similar logic to the Pipeline
+ // check for error if so we should break out
+ if (!continueProcessing(current, "so breaking out of the
routing slip", LOG)) {
+ break;
+ }
- // copy results back to the original exchange
- ExchangeHelper.copyResults(original, current);
- } catch (Exception e) {
- ex.setException(e);
+ Endpoint nextEndpoint;
+ boolean prototype = cacheSize < 0;
+ try {
+ Object recipient = iter.next(stepExchange);
+ recipient = prepareRecipient(stepExchange, recipient);
+ // enforce the optional allowed-schemes allow-list on the
resolved dynamic recipient (CAMEL-24298)
+ ProcessorHelper.checkAllowedSchemes(allowedSchemes,
recipient);
+ Endpoint existing = getExistingEndpoint(stepExchange,
recipient);
+ if (existing == null) {
+ nextEndpoint = resolveEndpoint(stepExchange,
recipient, prototype);
+ } else {
+ nextEndpoint = existing;
+ // we have an existing endpoint then its not a
prototype scope
+ prototype = false;
}
+ // if no endpoint was resolved then try the next
+ if (nextEndpoint == null) {
+ continue;
+ }
+ } catch (Exception e) {
+ // error resolving endpoint so we should break out
+ current.setException(e);
+ break;
+ }
- // okay we are completely done with the routing slip
- // so we need to signal done on the original callback so
it can continue
- cb.done(false);
+ // prepare and process the routing slip
+ final AsyncCallback cbNext = getNextCallback(prototype,
nextEndpoint, originalCallback);
+ boolean sync = processExchange(nextEndpoint, current,
original, cbNext, iter, prototype);
+ current = prepareExchangeForRoutingSlip(current, nextEndpoint);
+
+ if (!sync) {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Processing exchangeId: {} is continued
being processed asynchronously",
+ original.getExchangeId());
+ }
+ return;
}
- });
- });
+ }
+
+ // logging current as it contains the exchange that might have
altered the payload and since
+ // we are logging the completion if will be confusing if we log
the original instead
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Processing complete for exchangeId: {} >>> {}",
original.getExchangeId(), current);
+ }
+
+ // copy results back to the original exchange
+ ExchangeHelper.copyResults(original, current);
+ } catch (Exception e) {
+ stepExchange.setException(e);
+ }
+
+ // okay we are completely done with the routing slip
+ // so we need to signal done on the original callback so it can
continue
+ originalCallback.done(false);
}
private static AsyncCallback getNextCallback(boolean prototype, Endpoint
nextEndpoint, AsyncCallback cb) {
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/routingslip/DynamicRouterEventNotifierTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/routingslip/DynamicRouterEventNotifierTest.java
index db30614612a6..3105456a742c 100644
---
a/core/camel-core/src/test/java/org/apache/camel/processor/routingslip/DynamicRouterEventNotifierTest.java
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/routingslip/DynamicRouterEventNotifierTest.java
@@ -16,6 +16,9 @@
*/
package org.apache.camel.processor.routingslip;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+
import org.apache.camel.Body;
import org.apache.camel.CamelContext;
import org.apache.camel.ContextTestSupport;
@@ -54,6 +57,11 @@ public class DynamicRouterEventNotifierTest extends
ContextTestSupport {
assertEquals(5, notifier.getSending(), "Should have 5 sending events");
assertEquals(5, notifier.getSent(), "Should have 5 sent events");
+ // the sent event of each step comes before the next step is sent (the
steps complete asynchronously, as the
+ // error handler of each step schedules its work)
+ assertEquals(List.of("sending direct://start", "sending mock://x",
"sent mock://x", "sending mock://y",
+ "sent mock://y", "sending mock://z", "sent mock://z", "sending
mock://end", "sent mock://end",
+ "sent direct://start"), notifier.getEvents());
}
@Override
@@ -80,15 +88,18 @@ public class DynamicRouterEventNotifierTest extends
ContextTestSupport {
private final class MyEventNotifier extends EventNotifierSupport {
+ private final List<String> events = new CopyOnWriteArrayList<>();
private int sending;
private int sent;
@Override
public void notify(CamelEvent event) {
if (event instanceof ExchangeSendingEvent) {
+ events.add("sending " + ((ExchangeSendingEvent)
event).getEndpoint().getEndpointUri());
log.info("Sending: {}", event);
sending++;
} else {
+ events.add("sent " + ((ExchangeSentEvent)
event).getEndpoint().getEndpointUri());
sent++;
}
}
@@ -105,5 +116,9 @@ public class DynamicRouterEventNotifierTest extends
ContextTestSupport {
public int getSent() {
return sent;
}
+
+ public List<String> getEvents() {
+ return events;
+ }
}
}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/routingslip/RoutingSlipEventNotifierTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/routingslip/RoutingSlipEventNotifierTest.java
index 6761dcce9eb2..c76aa1e3e5dc 100644
---
a/core/camel-core/src/test/java/org/apache/camel/processor/routingslip/RoutingSlipEventNotifierTest.java
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/routingslip/RoutingSlipEventNotifierTest.java
@@ -16,6 +16,9 @@
*/
package org.apache.camel.processor.routingslip;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+
import org.apache.camel.CamelContext;
import org.apache.camel.ContextTestSupport;
import org.apache.camel.builder.RouteBuilder;
@@ -51,6 +54,11 @@ public class RoutingSlipEventNotifierTest extends
ContextTestSupport {
assertEquals(5, notifier.getSending(), "Should have 5 sending events");
assertEquals(5, notifier.getSent(), "Should have 5 sent events");
+ // the sent event of each step comes before the next step is sent (the
steps complete asynchronously, as the
+ // error handler of each step schedules its work)
+ assertEquals(List.of("sending direct://start", "sending mock://x",
"sent mock://x", "sending mock://y",
+ "sent mock://y", "sending mock://z", "sent mock://z", "sending
mock://end", "sent mock://end",
+ "sent direct://start"), notifier.getEvents());
}
@Override
@@ -64,15 +72,18 @@ public class RoutingSlipEventNotifierTest extends
ContextTestSupport {
private final class MyEventNotifier extends EventNotifierSupport {
+ private final List<String> events = new CopyOnWriteArrayList<>();
private int sending;
private int sent;
@Override
public void notify(CamelEvent event) {
if (event instanceof ExchangeSendingEvent) {
+ events.add("sending " + ((ExchangeSendingEvent)
event).getEndpoint().getEndpointUri());
log.info("Sending: {}", event);
sending++;
} else {
+ events.add("sent " + ((ExchangeSentEvent)
event).getEndpoint().getEndpointUri());
log.info("Sent: {}", event);
sent++;
}
@@ -90,5 +101,9 @@ public class RoutingSlipEventNotifierTest extends
ContextTestSupport {
public int getSent() {
return sent;
}
+
+ public List<String> getEvents() {
+ return events;
+ }
}
}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/routingslip/RoutingSlipSentEventOrderTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/routingslip/RoutingSlipSentEventOrderTest.java
new file mode 100644
index 000000000000..5ccdb5ce52ce
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/routingslip/RoutingSlipSentEventOrderTest.java
@@ -0,0 +1,163 @@
+/*
+ * 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.processor.routingslip;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.spi.CamelEvent;
+import org.apache.camel.support.EventNotifierSupport;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The ExchangeSentEvent of a step of a routing slip or dynamic router is
fired when that step is done, before the next
+ * step is sent, also when the step completes asynchronously, and its time
taken is the time of that step only.
+ */
+class RoutingSlipSentEventOrderTest extends ContextTestSupport {
+
+ private final List<String> events = Collections.synchronizedList(new
ArrayList<>());
+ private final Map<String, Long> timeTaken = new ConcurrentHashMap<>();
+ private final Map<String, String> visitedAtSent = new
ConcurrentHashMap<>();
+ private final Map<String, Boolean> failedAtSent = new
ConcurrentHashMap<>();
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+ context.getManagementStrategy().addEventNotifier(new
EventNotifierSupport() {
+ @Override
+ public void notify(CamelEvent event) {
+ if (event instanceof CamelEvent.ExchangeSendingEvent sending
+ && isStep(sending.getEndpoint().getEndpointUri())) {
+ events.add("sending " +
name(sending.getEndpoint().getEndpointUri()));
+ } else if (event instanceof CamelEvent.ExchangeSentEvent sent
&& isStep(sent.getEndpoint().getEndpointUri())) {
+ String step = name(sent.getEndpoint().getEndpointUri());
+ events.add("sent " + step);
+ timeTaken.put(step, sent.getTimeTaken());
+ String visited =
sent.getExchange().getMessage().getHeader("visited", String.class);
+ if (visited != null) {
+ visitedAtSent.put(step, visited);
+ }
+ failedAtSent.put(step, sent.getExchange().isFailed());
+ }
+ }
+ });
+ return context;
+ }
+
+ private static boolean isStep(String uri) {
+ return uri.startsWith("direct://step");
+ }
+
+ private static String name(String uri) {
+ // step1 and step1sync are both step1
+ return uri.substring("direct://".length()).replace("sync", "");
+ }
+
+ @Test
+ void routingSlipWithAsyncSteps() {
+ template.sendBody("direct:slipAsync", "Hello");
+ checkOrder();
+ }
+
+ @Test
+ void routingSlipWithStepsThatDelaySynchronously() {
+ // the steps still complete asynchronously, as the error handler of
each step schedules its work
+ template.sendBody("direct:slipSync", "Hello");
+ checkOrder();
+ }
+
+ @Test
+ void routingSlipWithPrototypeEndpoints() {
+ template.sendBody("direct:slipPrototype", "Hello");
+ checkOrder();
+ }
+
+ @Test
+ void dynamicRouterWithAsyncSteps() {
+ template.sendBody("direct:dynamicAsync", "Hello");
+ checkOrder();
+ }
+
+ @Test
+ void sentEventOfAStepSeesTheMessageOfThatStep() {
+ Exchange out = template.request("direct:slipAsync", e ->
e.getMessage().setBody("Hello"));
+ // the routing slip result is the message after its last step, and the
sent event of each step sees the message
+ // of that step (not yet the result of the steps after it)
+ assertEquals("step1,step2", out.getMessage().getHeader("visited"));
+ assertEquals("step1", visitedAtSent.get("step1"));
+ assertEquals("step1,step2", visitedAtSent.get("step2"));
+ }
+
+ @Test
+ void sentEventOfAStepIsNotFailedByALaterStep() {
+ Exchange out = template.request("direct:slipFail", e ->
e.getMessage().setBody("Hello"));
+ assertTrue(out.isFailed());
+ assertEquals(List.of("sending step1", "sent step1", "sending
stepFail", "sent stepFail"), events);
+ assertFalse(failedAtSent.get("step1"), "The sent event of step1 is
failed by the failure of the next step");
+ assertTrue(failedAtSent.get("stepFail"));
+ }
+
+ private void checkOrder() {
+ assertEquals(List.of("sending step1", "sent step1", "sending step2",
"sent step2"), events);
+ // step1 takes about 10 ms and step2 about 300 ms: the time of step1
does not include step2
+ assertNotNull(timeTaken.get("step2"));
+ assertTrue(timeTaken.get("step1") < timeTaken.get("step2"), "Time
taken of step1 includes step2: " + timeTaken);
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("direct:slipAsync").routingSlip(constant("direct:step1,direct:step2"));
+
from("direct:slipSync").routingSlip(constant("direct:step1sync,direct:step2sync"));
+
from("direct:slipPrototype").routingSlip(constant("direct:step1,direct:step2")).cacheSize(-1);
+
from("direct:slipFail").routingSlip(constant("direct:step1,direct:stepFail"));
+
from("direct:dynamicAsync").dynamicRouter(method(RoutingSlipSentEventOrderTest.class,
"next"));
+
+ // a delay is asynchronous by default
+ from("direct:step1").delay(10).setHeader("visited",
constant("step1"));
+ from("direct:step2").delay(300).setHeader("visited",
simple("${header.visited},step2"));
+ from("direct:stepFail").throwException(new
IllegalStateException("step failed"));
+
from("direct:step1sync").delay(10).syncDelayed().setHeader("visited",
constant("step1"));
+
from("direct:step2sync").delay(300).syncDelayed().setHeader("visited",
simple("${header.visited},step2"));
+ }
+ };
+ }
+
+ public static String next(Exchange exchange) {
+ String previous = exchange.getProperty(Exchange.SLIP_ENDPOINT,
String.class);
+ if (previous == null) {
+ return "direct:step1";
+ } else if (previous.endsWith("step1")) {
+ return "direct:step2";
+ }
+ return null;
+ }
+}
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 abce6acc36ae..adeba4ec966d 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
@@ -149,6 +149,19 @@ parallel split, no longer waits for a permit when none is
available. It runs the
expired, or forever with `Block`. A split with a `timeout` could lose items
that way. A submitter outside the pool waits
as before.
+=== camel-core - Routing Slip and Dynamic Router fire the sent event of a step
before the next step
+
+The `ExchangeSentEvent` of a step of a `routingSlip` or `dynamicRouter` is now
fired when that step is done, before the
+next step is sent, as with `recipientList`. Before, when a step completed
asynchronously (nearly every step, as the
+error handler of each step schedules its work), the sent events of the steps
were fired in reverse order after the last
+step: the time taken of a step included all the steps after it, and the sent
event of the first step carried the
+message and the failure after the last step.
+
+Users of the sent events see the change: the time of a step is lower (such as
in the exchange event notifier timers of
+`camel-micrometer` and `camel-opentelemetry-metrics`), the first step is no
longer reported as failed because a later
+step failed, and event-based tracing (`camel-jfr`, and `camel-telemetry`
together with its own fix for asynchronous
+steps) pairs the events of each step. There is no option to keep the old order.
+
=== camel-core - RedeliveryErrorHandler.defensiveCopyExchangeIfNeeded is
deprecated
When redelivery is enabled, the error handler (`DefaultErrorHandler`,
`DeadLetterChannel` and the transaction error
