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 b2eb665ea77f CAMEL-26450: camel-telemetry - end the span of an
asynchronous step before the next step starts (#28806)
b2eb665ea77f is described below
commit b2eb665ea77f8fdc2aff7d4a202cb2f5894d48c3
Author: allthingssecurity <[email protected]>
AuthorDate: Mon Nov 2 16:22:40 2026 +0530
CAMEL-26450: camel-telemetry - end the span of an asynchronous step before
the next step starts (#28806)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../observability/AsyncProcessorSpanTest.java | 89 +++++++
.../opentelemetry2/AsyncProcessorSpanTest.java | 119 +++++++++
.../camel/telemetrydev/AsyncProcessorSpanTest.java | 92 +++++++
.../TraceProcessorsInterceptStrategy.java | 55 +++-
.../camel/telemetry/AsyncProcessorSpanTest.java | 284 +++++++++++++++++++++
.../telemetry/ProcessorSpanBeginFailureTest.java | 123 +++++++++
.../telemetry/ProcessorSpanEndFailureTest.java | 106 ++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 21 ++
8 files changed, 879 insertions(+), 10 deletions(-)
diff --git
a/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/AsyncProcessorSpanTest.java
b/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/AsyncProcessorSpanTest.java
new file mode 100644
index 000000000000..3c9b2f296f42
--- /dev/null
+++
b/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/AsyncProcessorSpanTest.java
@@ -0,0 +1,89 @@
+/*
+ * 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.micrometer.observability;
+
+import java.util.List;
+import java.util.Map;
+
+import io.opentelemetry.sdk.trace.data.SpanData;
+import org.apache.camel.CamelContext;
+import org.apache.camel.RoutesBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import
org.apache.camel.micrometer.observability.CamelOpenTelemetryExtension.OtelTrace;
+import org.apache.camel.telemetry.Op;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * With Micrometer Observation on OpenTelemetry: the steps after an
asynchronous delay are children of the route span,
+ * and the delay span is ended before them.
+ */
+class AsyncProcessorSpanTest extends
MicrometerObservabilityTracerPropagationTestSupport {
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+ // the spans of the core processors (the support class disables them)
+ tst.setDisableCoreProcessors(false);
+ return context;
+ }
+
+ @Test
+ void stepsAfterAsyncDelayAreChildrenOfTheRoute() {
+ template.sendBody("direct:start", "my-body");
+ Map<String, OtelTrace> traces = otelExtension.getTraces();
+ assertEquals(1, traces.size());
+ List<SpanData> spans = traces.values().iterator().next().getSpans();
+ SpanData route = getSpan(spans, "direct://start", Op.EVENT_RECEIVED);
+ SpanData delay = byName(spans, "delay-delay");
+ SpanData step2 = byName(spans, "step2-setHeader");
+ SpanData step3 = byName(spans, "step3-setHeader");
+ for (SpanData span : spans) {
+ assertTrue(span.hasEnded(), "Span not ended: " + span.getName());
+ }
+ assertEquals(route.getSpanId(), delay.getParentSpanId());
+ assertEquals(route.getSpanId(), step2.getParentSpanId());
+ assertEquals(route.getSpanId(), step3.getParentSpanId());
+ assertTrue(delay.getEndEpochNanos() <= step2.getStartEpochNanos(),
"The delay span ends after the next step starts");
+ }
+
+ private static SpanData byName(List<SpanData> spans, String name) {
+ for (SpanData span : spans) {
+ if (name.equals(span.getName())) {
+ return span;
+ }
+ }
+ throw new IllegalArgumentException("No span " + name + " in " + spans);
+ }
+
+ @Override
+ protected RoutesBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ // a delay is asynchronous by default
+ from("direct:start").routeId("start")
+ .delay(10).id("delay")
+ .setHeader("step2", constant("2")).id("step2")
+ .setHeader("step3", constant("3")).id("step3");
+ }
+ };
+ }
+
+}
diff --git
a/components/camel-opentelemetry2/src/test/java/org/apache/camel/opentelemetry2/AsyncProcessorSpanTest.java
b/components/camel-opentelemetry2/src/test/java/org/apache/camel/opentelemetry2/AsyncProcessorSpanTest.java
new file mode 100644
index 000000000000..48a8754b4731
--- /dev/null
+++
b/components/camel-opentelemetry2/src/test/java/org/apache/camel/opentelemetry2/AsyncProcessorSpanTest.java
@@ -0,0 +1,119 @@
+/*
+ * 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.opentelemetry2;
+
+import java.util.List;
+import java.util.Map;
+
+import io.opentelemetry.sdk.trace.data.SpanData;
+import org.apache.camel.CamelContext;
+import org.apache.camel.CamelContextAware;
+import org.apache.camel.RoutesBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.opentelemetry2.CamelOpenTelemetryExtension.OtelTrace;
+import org.apache.camel.telemetry.Op;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * With the OpenTelemetry SDK: the span of a step that completes on another
thread (an asynchronous delay, the threads
+ * EIP) ends before the next step of the route starts, and the next steps are
children of the route span.
+ */
+class AsyncProcessorSpanTest extends OpenTelemetryTracerTestSupport {
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ OpenTelemetryTracer tst = new OpenTelemetryTracer();
+ tst.setTracer(otelExtension.getOpenTelemetry().getTracer("traceTest"));
+
tst.setContextPropagators(otelExtension.getOpenTelemetry().getPropagators());
+ CamelContext context = super.createCamelContext();
+ CamelContextAware.trySetCamelContext(tst, context);
+ tst.init(context);
+ return context;
+ }
+
+ @Test
+ void stepsAfterAsyncDelayAreChildrenOfTheRoute() {
+ check("direct:delay", "async-delay", "delay");
+ }
+
+ @Test
+ void stepsAfterThreadsAreChildrenOfTheRoute() {
+ check("direct:threads", "threads-threads", "threads");
+ }
+
+ @Test
+ void stepsAfterSyncDelayAreChildrenOfTheRoute() {
+ // control: a delay that completes on the same thread
+ check("direct:syncdelay", "sync-delay", "sync");
+ }
+
+ private void check(String uri, String asyncStep, String prefix) {
+ template.sendBody(uri, "my-body");
+ Map<String, OtelTrace> traces = otelExtension.getTraces();
+ assertEquals(1, traces.size());
+ List<SpanData> spans = traces.values().iterator().next().getSpans();
+ SpanData route = getSpan(spans, uri.replace(":", "://"),
Op.EVENT_RECEIVED);
+ SpanData async = byName(spans, asyncStep);
+ SpanData step2 = byName(spans, prefix + "Step2-setHeader");
+ SpanData step3 = byName(spans, prefix + "Step3-setHeader");
+ for (SpanData span : spans) {
+ assertTrue(span.hasEnded(), "Span not ended: " + span.getName());
+ }
+ assertEquals(route.getSpanId(), async.getParentSpanId());
+ assertEquals(route.getSpanId(), step2.getParentSpanId());
+ assertEquals(route.getSpanId(), step3.getParentSpanId());
+ assertTrue(async.getEndEpochNanos() <= step2.getStartEpochNanos(),
+ "The span " + asyncStep + " ends after the next step starts");
+ }
+
+ private static SpanData byName(List<SpanData> spans, String name) {
+ for (SpanData span : spans) {
+ if (name.equals(span.getName())) {
+ return span;
+ }
+ }
+ throw new IllegalArgumentException("No span " + name + " in " + spans);
+ }
+
+ @Override
+ protected RoutesBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ // a delay is asynchronous by default
+ from("direct:delay").routeId("delayRoute")
+ .delay(10).id("async")
+ .setHeader("step2", constant("2")).id("delayStep2")
+ .setHeader("step3", constant("3")).id("delayStep3");
+
+ from("direct:threads").routeId("threadsRoute")
+ .threads(1).id("threads")
+ .setHeader("step2", constant("2")).id("threadsStep2")
+ .setHeader("step3", constant("3")).id("threadsStep3");
+
+ from("direct:syncdelay").routeId("syncDelayRoute")
+ .delay(10).syncDelayed().id("sync")
+ .setHeader("step2", constant("2")).id("syncStep2")
+ .setHeader("step3", constant("3")).id("syncStep3");
+ }
+ };
+ }
+
+}
diff --git
a/components/camel-telemetry-dev/src/test/java/org/apache/camel/telemetrydev/AsyncProcessorSpanTest.java
b/components/camel-telemetry-dev/src/test/java/org/apache/camel/telemetrydev/AsyncProcessorSpanTest.java
new file mode 100644
index 000000000000..7b6a7effc704
--- /dev/null
+++
b/components/camel-telemetry-dev/src/test/java/org/apache/camel/telemetrydev/AsyncProcessorSpanTest.java
@@ -0,0 +1,92 @@
+/*
+ * 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.telemetrydev;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.CamelContextAware;
+import org.apache.camel.RoutesBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.telemetry.Op;
+import org.apache.camel.telemetry.TagConstants;
+import org.junit.jupiter.api.Test;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * The steps after an asynchronous delay are children of the route span (and
not of the delay span).
+ */
+class AsyncProcessorSpanTest extends TelemetryDevTracerTestSupport {
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ TelemetryDevTracer tst = new TelemetryDevTracer();
+ tst.setTraceFormat("json");
+ CamelContext context = super.createCamelContext();
+ CamelContextAware.trySetCamelContext(tst, context);
+ tst.init(context);
+ return context;
+ }
+
+ @Test
+ void stepsAfterAsyncDelayAreChildrenOfTheRoute() throws IOException {
+ template.sendBody("direct:start", "my-body");
+ // the trace is written to the log when all its spans are ended
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
tracesFromLog().size() == 1);
+ Map<String, DevTrace> traces = tracesFromLog();
+ List<DevSpanAdapter> spans =
traces.values().iterator().next().getSpans();
+ // the template's send, the route, the delay and two steps
+ assertEquals(5, spans.size());
+ DevSpanAdapter route = getSpan(spans, "direct://start",
Op.EVENT_RECEIVED);
+ DevSpanAdapter delay = processorSpan(spans, "delay-delay");
+ DevSpanAdapter step2 = processorSpan(spans, "step2-setHeader");
+ DevSpanAdapter step3 = processorSpan(spans, "step3-setHeader");
+ assertEquals(route.getTag("spanid"), delay.getTag("parentSpan"));
+ assertEquals(route.getTag("spanid"), step2.getTag("parentSpan"));
+ assertEquals(route.getTag("spanid"), step3.getTag("parentSpan"));
+ }
+
+ private static DevSpanAdapter processorSpan(List<DevSpanAdapter> spans,
String component) {
+ for (DevSpanAdapter span : spans) {
+ if (component.equals(span.getTag(TagConstants.COMPONENT))
+ &&
Op.EVENT_PROCESS.toString().equals(span.getTag(TagConstants.OP))) {
+ return span;
+ }
+ }
+ throw new IllegalArgumentException("No processor span " + component);
+ }
+
+ @Override
+ protected RoutesBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ // a delay is asynchronous by default
+ from("direct:start").routeId("start")
+ .delay(10).id("delay")
+ .setHeader("step2", constant("2")).id("step2")
+ .setHeader("step3", constant("3")).id("step3");
+ }
+ };
+ }
+
+}
diff --git
a/components/camel-telemetry/src/main/java/org/apache/camel/telemetry/TraceProcessorsInterceptStrategy.java
b/components/camel-telemetry/src/main/java/org/apache/camel/telemetry/TraceProcessorsInterceptStrategy.java
index dc75cc4bd13c..420efa15a23f 100644
---
a/components/camel-telemetry/src/main/java/org/apache/camel/telemetry/TraceProcessorsInterceptStrategy.java
+++
b/components/camel-telemetry/src/main/java/org/apache/camel/telemetry/TraceProcessorsInterceptStrategy.java
@@ -28,12 +28,16 @@ import org.apache.camel.Processor;
import org.apache.camel.spi.InterceptStrategy;
import org.apache.camel.support.AsyncCallbackToCompletableFutureAdapter;
import org.apache.camel.support.processor.DelegateAsyncProcessor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* TraceProcessorsInterceptStrategy is used to wrap each processor calls and
generate a Span for each process execution.
*/
public class TraceProcessorsInterceptStrategy implements InterceptStrategy {
+ private static final Logger LOG =
LoggerFactory.getLogger(TraceProcessorsInterceptStrategy.class);
+
private Tracer tracer;
public TraceProcessorsInterceptStrategy(Tracer tracer) {
@@ -63,7 +67,14 @@ public class TraceProcessorsInterceptStrategy implements
InterceptStrategy {
tracer.beginProcessorSpan(exchange, processorName);
try {
processor.process(exchange);
- } finally {
+ } catch (Throwable e) {
+ // do not replace the exception of the processor with a
failure to end the span
+ endProcessorSpanKeepingFailure(exchange, processorName);
+ throw e;
+ }
+ if (exchange.getException() != null) {
+ endProcessorSpanKeepingFailure(exchange, processorName);
+ } else {
tracer.endProcessorSpan(exchange, processorName);
}
} else {
@@ -74,25 +85,36 @@ public class TraceProcessorsInterceptStrategy implements
InterceptStrategy {
@Override
public boolean process(Exchange exchange, AsyncCallback callback) {
String processorName = processorDefinition.getId() + "-" +
processorDefinition.getShortName();
- boolean trace = shouldTrace(processorName, exchange);
- if (trace) {
+ boolean started = false;
+ if (shouldTrace(processorName, exchange)) {
try {
tracer.beginProcessorSpan(exchange, processorName);
+ started = true;
} catch (Exception e) {
exchange.setException(e);
}
}
+ // only end the span when it was begun, as ending pulls the span
on top of the exchange's span stack
+ final boolean end = started;
return processor.process(exchange, doneSync -> {
try {
- callback.done(doneSync);
- } finally {
- if (trace) {
- try {
- tracer.endProcessorSpan(exchange, processorName);
- } catch (Exception e) {
- exchange.setException(e);
+ if (end) {
+ // end the span before the route continues: when the
processor completed on another thread (such
+ // as an asynchronous delay), callback.done runs the
next steps of the route inside this call,
+ // and they must not get this span as their parent
+ if (exchange.getException() != null) {
+ // do not replace an exception the processor has
already set on the exchange
+ endProcessorSpanKeepingFailure(exchange,
processorName);
+ } else {
+ try {
+ tracer.endProcessorSpan(exchange,
processorName);
+ } catch (Exception e) {
+ exchange.setException(e);
+ }
}
}
+ } finally {
+ callback.done(doneSync);
}
});
}
@@ -105,6 +127,19 @@ public class TraceProcessorsInterceptStrategy implements
InterceptStrategy {
return callback.getFuture();
}
+ /**
+ * Ends the span of a processor that failed: a failure to end the span
is logged, so that the exception of the
+ * processor is kept.
+ */
+ private void endProcessorSpanKeepingFailure(Exchange exchange, String
processorName) {
+ try {
+ tracer.endProcessorSpan(exchange, processorName);
+ } catch (Exception e) {
+ LOG.warn("Failed to end the processor span for {}. The
exception of the processor is kept.",
+ processorName, e);
+ }
+ }
+
private boolean shouldTrace(String processorName, Exchange exchange) {
// skip processors that send to an endpoint (to, toD, wireTap,
enrich)
// unwrap through any delegate chain (e.g. otel scope wrapper) to
find the real processor
diff --git
a/components/camel-telemetry/src/test/java/org/apache/camel/telemetry/AsyncProcessorSpanTest.java
b/components/camel-telemetry/src/test/java/org/apache/camel/telemetry/AsyncProcessorSpanTest.java
new file mode 100644
index 000000000000..e8b5ed6cd0b6
--- /dev/null
+++
b/components/camel-telemetry/src/test/java/org/apache/camel/telemetry/AsyncProcessorSpanTest.java
@@ -0,0 +1,284 @@
+/*
+ * 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.telemetry;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.CamelContextAware;
+import org.apache.camel.CamelExecutionException;
+import org.apache.camel.RoutesBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.model.RouteDefinition;
+import org.apache.camel.telemetry.mock.MockSpanAdapter;
+import org.apache.camel.telemetry.mock.MockTrace;
+import org.apache.camel.telemetry.mock.MockTracer;
+import org.apache.camel.test.junit6.ExchangeTestSupport;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The span of a step that calls back from another thread (an asynchronous
delay, the threads EIP, a bean that returns a
+ * CompletableFuture) ends before the next step of the route starts, so the
next steps are children of the span that
+ * encloses the step (the route span, or the span of an EIP such as a choice)
and not of the asynchronous step.
+ */
+class AsyncProcessorSpanTest extends ExchangeTestSupport {
+
+ MockTracer mockTracer;
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+ this.mockTracer = new MockTracer();
+ CamelContextAware.trySetCamelContext(mockTracer, context);
+ mockTracer.init(context);
+ context.getRegistry().bind("futureBean", new FutureBean());
+ return context;
+ }
+
+ @Test
+ void stepsAfterAsyncDelayAreChildrenOfTheRoute() {
+ checkStepsAfter("direct:delay", "async-delay", "delay");
+ }
+
+ @Test
+ void stepsAfterThreadsAreChildrenOfTheRoute() {
+ checkStepsAfter("direct:threads", "threads-threads", "threads");
+ }
+
+ @Test
+ void stepsAfterBeanReturningFutureAreChildrenOfTheRoute() {
+ checkStepsAfter("direct:future", "future-bean", "future");
+ }
+
+ @Test
+ void stepsAfterSyncDelayAreChildrenOfTheRoute() {
+ // control: a delay that completes on the same thread (passes without
the fix as well)
+ checkStepsAfter("direct:syncdelay", "sync-delay", "sync");
+ }
+
+ @Test
+ void stepsAfterAsyncDelayInsideChoiceAreChildrenOfTheChoice() {
+ List<Span> spans = send("direct:choice");
+ MockSpanAdapter route = SpanTestSupport.getSpan(spans,
"direct://choice", Op.EVENT_RECEIVED);
+ MockSpanAdapter choice = span(spans, "choice-choice");
+ MockSpanAdapter delay = span(spans, "innerDelay-delay");
+ MockSpanAdapter inner = span(spans, "inner-setHeader");
+ MockSpanAdapter after = span(spans, "after-setHeader");
+ allDone(spans);
+ assertEquals(route.getTag("spanid"), choice.getTag("parentSpan"));
+ assertEquals(choice.getTag("spanid"), delay.getTag("parentSpan"));
+ assertEquals(choice.getTag("spanid"), inner.getTag("parentSpan"));
+ assertEquals(route.getTag("spanid"), after.getTag("parentSpan"));
+ endsBeforeStart(delay, inner);
+ endsBeforeStart(choice, after);
+ }
+
+ @Test
+ void calledRouteWithAsyncDelayEndsItsOwnSpans() {
+ List<Span> spans = send("direct:caller");
+ MockSpanAdapter caller = SpanTestSupport.getSpan(spans,
"direct://caller", Op.EVENT_RECEIVED);
+ MockSpanAdapter send = SpanTestSupport.getSpan(spans,
"direct://called", Op.EVENT_SENT);
+ MockSpanAdapter called = SpanTestSupport.getSpan(spans,
"direct://called", Op.EVENT_RECEIVED);
+ MockSpanAdapter calledDelay = span(spans, "calledDelay-delay");
+ MockSpanAdapter calledStep = span(spans, "calledStep-setHeader");
+ MockSpanAdapter callerStep = span(spans, "callerStep-setHeader");
+ allDone(spans);
+ assertEquals(caller.getTag("spanid"), send.getTag("parentSpan"));
+ assertEquals(send.getTag("spanid"), called.getTag("parentSpan"));
+ assertEquals(called.getTag("spanid"),
calledDelay.getTag("parentSpan"));
+ assertEquals(called.getTag("spanid"), calledStep.getTag("parentSpan"));
+ assertEquals(caller.getTag("spanid"), callerStep.getTag("parentSpan"));
+ // each span is ended by its own step: the called route and the send
end after the steps they contain
+ endsBeforeStart(calledDelay, calledStep);
+ endsNotBefore(called, calledStep);
+ endsNotBefore(send, called);
+ endsBeforeStart(send, callerStep);
+ endsNotBefore(caller, callerStep);
+ }
+
+ @Test
+ void recipientListToRoutesWithAsyncDelayEndsAllSpans() {
+ List<Span> spans = send("direct:recipients");
+ // the template's send, the route, the recipient list, two sends, two
called routes, two delays and two steps
+ assertEquals(11, spans.size(), "Spans: " + spans);
+ allDone(spans);
+ MockSpanAdapter route = SpanTestSupport.getSpan(spans,
"direct://recipients", Op.EVENT_RECEIVED);
+ MockSpanAdapter recipientList = span(spans,
"recipients-recipientList");
+ assertEquals(route.getTag("spanid"),
recipientList.getTag("parentSpan"));
+ for (String step : List.of("rl1Step-setHeader", "rl2Step-setHeader")) {
+ MockSpanAdapter parent = byId(spans, span(spans,
step).getTag("parentSpan"));
+ assertEquals(Op.EVENT_RECEIVED.toString(), parent.getTag("op"),
+ "The step " + step + " is not a child of its route");
+ }
+ }
+
+ @Test
+ void failureAfterAsyncDelayIsOnlyOnTheFailingStep() {
+ assertThrows(CamelExecutionException.class, () ->
template.sendBody("direct:fail", "my-body"));
+ List<Span> spans = spans();
+ MockSpanAdapter route = SpanTestSupport.getSpan(spans,
"direct://fail", Op.EVENT_RECEIVED);
+ MockSpanAdapter delay = span(spans, "failDelay-delay");
+ MockSpanAdapter failing = span(spans, "failStep-throwException");
+ allDone(spans);
+ assertEquals(route.getTag("spanid"), failing.getTag("parentSpan"));
+ assertEquals("true", failing.getTag(TagConstants.ERROR));
+ assertNotEquals("true", delay.getTag(TagConstants.ERROR));
+ }
+
+ private void checkStepsAfter(String uri, String asyncStep, String prefix) {
+ List<Span> spans = send(uri);
+ MockSpanAdapter route = SpanTestSupport.getSpan(spans,
uri.replace(":", "://"), Op.EVENT_RECEIVED);
+ MockSpanAdapter async = span(spans, asyncStep);
+ MockSpanAdapter step2 = span(spans, prefix + "Step2-setHeader");
+ MockSpanAdapter step3 = span(spans, prefix + "Step3-setHeader");
+ allDone(spans);
+ assertEquals(route.getTag("spanid"), async.getTag("parentSpan"));
+ assertEquals(route.getTag("spanid"), step2.getTag("parentSpan"));
+ assertEquals(route.getTag("spanid"), step3.getTag("parentSpan"));
+ endsBeforeStart(async, step2);
+ }
+
+ private List<Span> send(String uri) {
+ template.sendBody(uri, "my-body");
+ return spans();
+ }
+
+ private List<Span> spans() {
+ Map<String, MockTrace> traces = mockTracer.traces();
+ assertEquals(1, traces.size());
+ return traces.values().iterator().next().spans();
+ }
+
+ private static MockSpanAdapter span(List<Span> spans, String component) {
+ MockSpanAdapter answer = null;
+ for (Span s : spans) {
+ MockSpanAdapter msa = (MockSpanAdapter) s;
+ if (component.equals(msa.getTag(TagConstants.COMPONENT))
+ && Op.EVENT_PROCESS.toString().equals(msa.getTag("op"))) {
+ answer = msa;
+ }
+ }
+ assertNotNull(answer, "No processor span " + component + " in " +
spans);
+ return answer;
+ }
+
+ private static MockSpanAdapter byId(List<Span> spans, String spanId) {
+ for (Span s : spans) {
+ MockSpanAdapter msa = (MockSpanAdapter) s;
+ if (spanId != null && spanId.equals(msa.getTag("spanid"))) {
+ return msa;
+ }
+ }
+ throw new IllegalArgumentException("No span with id " + spanId);
+ }
+
+ private static void allDone(List<Span> spans) {
+ for (Span s : spans) {
+ assertEquals("true", ((MockSpanAdapter) s).getTag("isDone"), "Span
not ended: " + s);
+ }
+ }
+
+ private static void endsBeforeStart(MockSpanAdapter first, MockSpanAdapter
next) {
+ long end = Long.parseLong(first.getTag("endTimestamp"));
+ long start = Long.parseLong(next.getTag("initTimestamp"));
+ assertTrue(end <= start, "The span " + first.tags() + " ends after the
span " + next.tags() + " starts");
+ }
+
+ private static void endsNotBefore(MockSpanAdapter outer, MockSpanAdapter
inner) {
+ long outerEnd = Long.parseLong(outer.getTag("endTimestamp"));
+ long innerEnd = Long.parseLong(inner.getTag("endTimestamp"));
+ assertTrue(outerEnd >= innerEnd, "The span " + outer.tags() + " ends
before the span " + inner.tags());
+ }
+
+ public static class FutureBean {
+ public CompletableFuture<String> call(String body) {
+ // completes on another thread, after the bean has returned
+ return CompletableFuture.supplyAsync(() -> body + "-reply",
+ CompletableFuture.delayedExecutor(10,
TimeUnit.MILLISECONDS));
+ }
+ }
+
+ @Override
+ protected RoutesBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ // a delay is asynchronous by default
+ from("direct:delay").routeId("delayRoute")
+ .delay(10).id("async")
+ .setHeader("step2", constant("2")).id("delayStep2")
+ .setHeader("step3", constant("3")).id("delayStep3");
+
+ from("direct:threads").routeId("threadsRoute")
+ .threads(1).id("threads")
+ .setHeader("step2", constant("2")).id("threadsStep2")
+ .setHeader("step3", constant("3")).id("threadsStep3");
+
+ from("direct:future").routeId("futureRoute")
+ .bean("futureBean", "call").id("future")
+ .setHeader("step2", constant("2")).id("futureStep2")
+ .setHeader("step3", constant("3")).id("futureStep3");
+
+ from("direct:syncdelay").routeId("syncDelayRoute")
+ .delay(10).syncDelayed().id("sync")
+ .setHeader("step2", constant("2")).id("syncStep2")
+ .setHeader("step3", constant("3")).id("syncStep3");
+
+ // the choice is built in two statements: in one chain, end()
after delay() does not end the choice
+ RouteDefinition choiceRoute =
from("direct:choice").routeId("choiceRoute");
+ choiceRoute.choice().id("choice")
+ .when(constant(true))
+ .delay(10).id("innerDelay")
+ .setHeader("inner", constant("x")).id("inner");
+ choiceRoute.setHeader("after", constant("y")).id("after");
+
+ from("direct:caller").routeId("callerRoute")
+ .to("direct:called")
+ .setHeader("caller", constant("x")).id("callerStep");
+
+ from("direct:called").routeId("calledRoute")
+ .delay(10).id("calledDelay")
+ .setHeader("called", constant("y")).id("calledStep");
+
+ from("direct:recipients").routeId("recipientsRoute")
+
.recipientList(constant("direct:rl1,direct:rl2")).id("recipients");
+
+ from("direct:rl1").routeId("rl1Route")
+ .delay(10).id("rl1Delay")
+ .setHeader("rl", constant("1")).id("rl1Step");
+
+ from("direct:rl2").routeId("rl2Route")
+ .delay(10).id("rl2Delay")
+ .setHeader("rl", constant("2")).id("rl2Step");
+
+ from("direct:fail").routeId("failRoute")
+ .delay(10).id("failDelay")
+ .throwException(new
IllegalStateException("fail")).id("failStep");
+ }
+ };
+ }
+
+}
diff --git
a/components/camel-telemetry/src/test/java/org/apache/camel/telemetry/ProcessorSpanBeginFailureTest.java
b/components/camel-telemetry/src/test/java/org/apache/camel/telemetry/ProcessorSpanBeginFailureTest.java
new file mode 100644
index 000000000000..45bac378a938
--- /dev/null
+++
b/components/camel-telemetry/src/test/java/org/apache/camel/telemetry/ProcessorSpanBeginFailureTest.java
@@ -0,0 +1,123 @@
+/*
+ * 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.telemetry;
+
+import java.util.List;
+import java.util.Map;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.CamelContextAware;
+import org.apache.camel.RoutesBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.telemetry.mock.MockSpanAdapter;
+import org.apache.camel.telemetry.mock.MockTrace;
+import org.apache.camel.telemetry.mock.MockTracer;
+import org.apache.camel.test.junit6.ExchangeTestSupport;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * When the span of a step cannot be begun, the end of that step does not end
another span (the span on top of the
+ * exchange's span stack is then the span of the route).
+ */
+class ProcessorSpanBeginFailureTest extends ExchangeTestSupport {
+
+ MockTracer mockTracer;
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+ this.mockTracer = new FailingBeginTracer();
+ CamelContextAware.trySetCamelContext(mockTracer, context);
+ mockTracer.init(context);
+ return context;
+ }
+
+ @Test
+ void stepAfterAFailedBeginIsAChildOfTheRoute() {
+ template.sendBody("direct:start", "my-body");
+ Map<String, MockTrace> traces = mockTracer.traces();
+ assertEquals(1, traces.size());
+ List<Span> spans = traces.values().iterator().next().spans();
+ // the template's send, the route, and the step after the failed one
(the failed one has no span)
+ assertEquals(3, spans.size(), "Spans: " + spans);
+ MockSpanAdapter route = SpanTestSupport.getSpan(spans,
"direct://start", Op.EVENT_RECEIVED);
+ MockSpanAdapter after = (MockSpanAdapter) spans.get(2);
+ assertEquals("after-setHeader", after.getTag(TagConstants.COMPONENT));
+ assertEquals(route.getTag("spanid"), after.getTag("parentSpan"));
+ for (Span s : spans) {
+ assertEquals("true", ((MockSpanAdapter) s).getTag("isDone"), "Span
not ended: " + s);
+ }
+ }
+
+ @Override
+ protected RoutesBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ // the failure to begin the span is set on the exchange, and
continued here
+ onException(IllegalStateException.class).continued(true);
+
+ from("direct:start").routeId("start")
+ .setHeader("a", constant("1")).id("boom")
+ .setHeader("b", constant("2")).id("after");
+ }
+ };
+ }
+
+ /**
+ * A tracer whose span creation fails for the step with the id boom.
+ */
+ private static class FailingBeginTracer extends MockTracer {
+ @Override
+ protected void initTracer() {
+ super.initTracer();
+ SpanLifecycleManager delegate = getSpanLifecycleManager();
+ setSpanLifecycleManager(new SpanLifecycleManager() {
+ @Override
+ public Span create(
+ String spanName, String spanKind, Span parent,
SpanContextPropagationExtractor extractor) {
+ if (spanName.startsWith("boom-")) {
+ throw new IllegalStateException("Cannot create the
span " + spanName);
+ }
+ return delegate.create(spanName, spanKind, parent,
extractor);
+ }
+
+ @Override
+ public void activate(Span span) {
+ delegate.activate(span);
+ }
+
+ @Override
+ public void deactivate(Span span) {
+ delegate.deactivate(span);
+ }
+
+ @Override
+ public void close(Span span) {
+ delegate.close(span);
+ }
+
+ @Override
+ public void inject(Span span, SpanContextPropagationInjector
injector, boolean includeTracing) {
+ delegate.inject(span, injector, includeTracing);
+ }
+ });
+ }
+ }
+}
diff --git
a/components/camel-telemetry/src/test/java/org/apache/camel/telemetry/ProcessorSpanEndFailureTest.java
b/components/camel-telemetry/src/test/java/org/apache/camel/telemetry/ProcessorSpanEndFailureTest.java
new file mode 100644
index 000000000000..a11d95dfa2f1
--- /dev/null
+++
b/components/camel-telemetry/src/test/java/org/apache/camel/telemetry/ProcessorSpanEndFailureTest.java
@@ -0,0 +1,106 @@
+/*
+ * 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.telemetry;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.CamelContextAware;
+import org.apache.camel.Exchange;
+import org.apache.camel.NamedNode;
+import org.apache.camel.Processor;
+import org.apache.camel.RoutesBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.model.ModelCamelContext;
+import org.apache.camel.telemetry.mock.MockTracer;
+import org.apache.camel.test.junit6.ExchangeTestSupport;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+/**
+ * A failure to end the span of a step does not replace an exception that the
step has already set on the exchange.
+ */
+class ProcessorSpanEndFailureTest extends ExchangeTestSupport {
+
+ MockTracer mockTracer;
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+ this.mockTracer = new FailingEndTracer();
+ CamelContextAware.trySetCamelContext(mockTracer, context);
+ mockTracer.init(context);
+ return context;
+ }
+
+ @Test
+ void endFailureKeepsTheExceptionOfTheStep() {
+ Exchange out = template.send("direct:fails", e ->
e.getMessage().setBody("my-body"));
+ Exception e = out.getException();
+ assertInstanceOf(IllegalArgumentException.class, e, "Exception: " + e);
+ assertEquals("business failure", e.getMessage());
+ }
+
+ @Test
+ void endFailureIsSetWhenTheStepDidNotFail() {
+ Exchange out = template.send("direct:ok", e ->
e.getMessage().setBody("my-body"));
+ Exception e = out.getException();
+ assertInstanceOf(IllegalStateException.class, e, "Exception: " + e);
+ assertEquals("Cannot end the span boomOk-setHeader", e.getMessage());
+ }
+
+ @Test
+ void endFailureKeepsTheExceptionOfTheBlockingStep() throws Exception {
+ // the blocking process(Exchange) of the processor that traces the step
+ NamedNode node = ((ModelCamelContext)
context).getRouteDefinition("fails").getOutputs().get(0);
+ Processor traced = new
TraceProcessorsInterceptStrategy(mockTracer).wrapProcessorInInterceptors(context,
node,
+ e -> {
+ throw new IllegalArgumentException("business failure");
+ }, null);
+ Exception e = assertThrows(Exception.class, () ->
traced.process(createExchangeWithBody("my-body")));
+ assertInstanceOf(IllegalArgumentException.class, e, "Exception: " + e);
+ assertEquals("business failure", e.getMessage());
+ }
+
+ @Override
+ protected RoutesBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:fails").routeId("fails")
+ .throwException(new IllegalArgumentException("business
failure")).id("boomFails");
+
+ from("direct:ok").routeId("ok")
+ .setHeader("a", constant("1")).id("boomOk");
+ }
+ };
+ }
+
+ /**
+ * A tracer that ends the span of a step with an id starting with boom,
then fails.
+ */
+ private static class FailingEndTracer extends MockTracer {
+ @Override
+ protected void endProcessorSpan(Exchange exchange, String
processorName) throws Exception {
+ super.endProcessorSpan(exchange, processorName);
+ if (processorName.startsWith("boom")) {
+ throw new IllegalStateException("Cannot end the span " +
processorName);
+ }
+ }
+ }
+}
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 170f1d8490c1..9becf9752ee5 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
@@ -597,6 +597,27 @@ closed that client for every other endpoint.
Camel now closes only the clients it created. A client set with
`amazonKinesisClient` or `amazonKinesisAsyncClient`, on
the component or on an endpoint, is no longer closed when Camel stops: close
it yourself when you no longer need it.
+=== camel-telemetry - the span of an asynchronous step ends when the step is
done
+
+This applies to `camel-opentelemetry2`, `camel-micrometer-observability` and
`camel-telemetry-dev`. The span of a step
+that calls back from another thread (such as `delay`, which is asynchronous by
default, `threads`, `throttle` with
+`asyncDelayed`, or a `bean` that returns a `CompletableFuture`) now ends when
the step is done, and the steps after it
+are children of the span that contains the step (the route span, or the span
of an EIP such as `choice`). Since 4.17
+(since 4.21 with the default options, which trace the core processors) the
following steps were children of the span
+of the asynchronous step, and that span lasted until the end of the route.
Around a called route the spans were ended
+by the wrong events, and with a `recipientList` or `enrich` to such a route
some spans were never ended. In particular:
+
+* The trace tree of such routes changes, and the span of the asynchronous step
is shorter.
+* A failure in a later step no longer marks the span of the asynchronous step
as failed: the error is on the span of
+ the failing step and on the spans that contain it.
+* Steps of an `onException` that run after the asynchronous step, and custom
spans created in a processor after it,
+ are no longer nested under its span.
+* An asynchronous step that fails and is then handled by an `onException` with
`handled(true)` now has its span marked
+ as an error, because the span ends before the error handler runs. A
synchronous step already behaved that way.
+
+This is the tree of Camel 4.16 and older, so there is no option to keep the
4.17 tree. A dashboard that read the time of
+the rest of the route from the span of the asynchronous step can use the route
span instead.
+
=== camel-oaipmh - the CamelOaimphOnlyFirst and CamelOaimphIgnoreSSLWarnings
headers apply to their message only
The `CamelOaimphOnlyFirst` and `CamelOaimphIgnoreSSLWarnings` headers now
apply only to the message that has them.
