You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by pc...@apache.org on 2025/09/30 12:09:35 UTC
(camel) branch main updated: chore(components): strengthen telemetry async tests
This is an automated email from the ASF dual-hosted git repository.
pcongiusti 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 11c9aacdaae chore(components): strengthen telemetry async tests
11c9aacdaae is described below
commit 11c9aacdaae871758f7e0b3c8a0de5d96f58f7f4
Author: Pasquale Congiusti <pa...@gmail.com>
AuthorDate: Tue Sep 30 12:43:06 2025 +0200
chore(components): strengthen telemetry async tests
Added some test to validate in case of any potential customer async component
---
.../camel/micrometer/observability/AsyncTest.java | 150 +++++++++++++++++
.../micrometer/observability/MyAsyncComponent.java | 46 ++++++
.../micrometer/observability/MyAsyncEndpoint.java | 98 +++++++++++
.../micrometer/observability/MyAsyncProducer.java | 75 +++++++++
.../org/apache/camel/opentelemetry2/AsyncTest.java | 179 +++++++++++++++++++++
.../camel/opentelemetry2/MyAsyncComponent.java | 46 ++++++
.../camel/opentelemetry2/MyAsyncEndpoint.java | 98 +++++++++++
.../camel/opentelemetry2/MyAsyncProducer.java | 75 +++++++++
.../org/apache/camel/telemetrydev/AsyncTest.java | 163 +++++++++++++++++++
.../camel/telemetrydev/MyAsyncComponent.java | 46 ++++++
.../apache/camel/telemetrydev/MyAsyncEndpoint.java | 98 +++++++++++
.../apache/camel/telemetrydev/MyAsyncProducer.java | 75 +++++++++
.../java/org/apache/camel/telemetry/AsyncTest.java | 170 +++++++++++++++++++
.../apache/camel/telemetry/MyAsyncComponent.java | 46 ++++++
.../apache/camel/telemetry/MyAsyncEndpoint.java | 98 +++++++++++
.../apache/camel/telemetry/MyAsyncProducer.java | 75 +++++++++
16 files changed, 1538 insertions(+)
diff --git a/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/AsyncTest.java b/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/AsyncTest.java
new file mode 100644
index 00000000000..b5ea8d1d843
--- /dev/null
+++ b/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/AsyncTest.java
@@ -0,0 +1,150 @@
+/*
+ * 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.io.IOException;
+import java.util.List;
+import java.util.Map;
+
+import io.micrometer.tracing.test.simple.SimpleSpan;
+import org.apache.camel.RoutesBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+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.assertNotEquals;
+
+/*
+ * AsyncTest tests the execution of a new spin off async components.
+ */
+public class AsyncTest extends MicrometerObservabilityTracerTestSupport {
+
+ @Test
+ void testRouteMultipleRequests() throws InterruptedException, IOException {
+ int j = 10;
+ MockEndpoint mock = getMockEndpoint("mock:end");
+ mock.expectedMessageCount(j);
+ mock.setAssertPeriod(5000);
+ for (int i = 0; i < j; i++) {
+ context.createProducerTemplate().sendBody("direct:start", "Hello!");
+ }
+ mock.assertIsSatisfied(1000);
+ Map<String, MicrometerObservabilityTrace> traces = traces();
+ // Each trace should have a unique trace id. It is enough to assert that
+ // the number of elements in the map is the same of the requests to prove
+ // all traces have been generated uniquely.
+ assertEquals(j, traces.size());
+ // Each trace should have the same structure
+ for (MicrometerObservabilityTrace trace : traces.values()) {
+ checkTrace(trace, "Hello!");
+ }
+
+ }
+
+ private void checkTrace(MicrometerObservabilityTrace trace, String expectedBody) {
+ List<SimpleSpan> spans = trace.getSpans();
+ assertEquals(8, spans.size());
+ SimpleSpan testProducer = MicrometerObservabilityTracerTestSupport.getSpan(spans, "direct://start", Op.EVENT_SENT);
+ SimpleSpan direct = MicrometerObservabilityTracerTestSupport.getSpan(spans, "direct://start", Op.EVENT_RECEIVED);
+ SimpleSpan asyncDirectTo = MicrometerObservabilityTracerTestSupport.getSpan(spans, "direct://async", Op.EVENT_SENT);
+ SimpleSpan asyncDirectFrom
+ = MicrometerObservabilityTracerTestSupport.getSpan(spans, "direct://async", Op.EVENT_RECEIVED);
+ SimpleSpan async = MicrometerObservabilityTracerTestSupport.getSpan(spans, "async://bye:camel", Op.EVENT_SENT);
+ SimpleSpan log = MicrometerObservabilityTracerTestSupport.getSpan(spans, "log://info", Op.EVENT_SENT);
+ SimpleSpan asyncLog = MicrometerObservabilityTracerTestSupport.getSpan(spans, "log://tapped", Op.EVENT_SENT);
+ SimpleSpan asyncMock = MicrometerObservabilityTracerTestSupport.getSpan(spans, "mock://end", Op.EVENT_SENT);
+
+ // Validate span completion
+ assertNotEquals("", testProducer.getEndTimestamp());
+ assertNotEquals("", direct.getEndTimestamp());
+ assertNotEquals("", asyncDirectTo.getEndTimestamp());
+ assertNotEquals("", log.getEndTimestamp());
+ assertNotEquals("", asyncDirectFrom.getEndTimestamp());
+ assertNotEquals("", async.getEndTimestamp());
+ assertNotEquals("", asyncLog.getEndTimestamp());
+ assertNotEquals("", asyncMock.getEndTimestamp());
+
+ // Validate same trace
+ assertEquals(testProducer.getTraceId(), direct.getTraceId());
+ assertEquals(testProducer.getTraceId(), asyncDirectTo.getTraceId());
+ assertEquals(testProducer.getTraceId(), log.getTraceId());
+ assertEquals(testProducer.getTraceId(), asyncDirectFrom.getTraceId());
+ assertEquals(testProducer.getTraceId(), async.getTraceId());
+ assertEquals(testProducer.getTraceId(), asyncLog.getTraceId());
+ assertEquals(testProducer.getTraceId(), asyncMock.getTraceId());
+
+ // Validate different Exchange ID
+ assertNotEquals(testProducer.getTags().get("exchangeId"), asyncDirectTo.getTags().get("exchangeId"));
+ assertEquals(testProducer.getTags().get("exchangeId"), direct.getTags().get("exchangeId"));
+ assertEquals(testProducer.getTags().get("exchangeId"), log.getTags().get("exchangeId"));
+ assertEquals(asyncDirectTo.getTags().get("exchangeId"), asyncDirectFrom.getTags().get("exchangeId"));
+ assertEquals(asyncDirectTo.getTags().get("exchangeId"), async.getTags().get("exchangeId"));
+ assertEquals(asyncDirectTo.getTags().get("exchangeId"), asyncLog.getTags().get("exchangeId"));
+ assertEquals(asyncDirectTo.getTags().get("exchangeId"), asyncMock.getTags().get("exchangeId"));
+
+ // Validate hierarchy
+ assertEquals("", testProducer.getParentId());
+ assertEquals(testProducer.getSpanId(), direct.getParentId());
+ assertEquals(direct.getSpanId(), asyncDirectTo.getParentId());
+ assertEquals(direct.getSpanId(), log.getParentId());
+ assertEquals(asyncDirectTo.getSpanId(), asyncDirectFrom.getParentId());
+ assertEquals(asyncDirectFrom.getSpanId(), async.getParentId());
+ assertEquals(asyncDirectFrom.getSpanId(), asyncLog.getParentId());
+ assertEquals(asyncDirectFrom.getSpanId(), asyncMock.getParentId());
+
+ // Validate message logging
+ assertEquals("message=A direct message", direct.getEvents().iterator().next().getValue());
+ assertEquals("message=An async message", asyncDirectFrom.getEvents().iterator().next().getValue());
+ String expectedBodyAsync = "Bye Camel";
+ if (expectedBody == null) {
+ assertEquals(
+ "message=Exchange[ExchangePattern: InOut, BodyType: null, Body: [Body is null]]",
+ log.getEvents().iterator().next().getValue());
+ } else {
+ assertEquals(
+ "message=Exchange[ExchangePattern: InOnly, BodyType: String, Body: " + expectedBody + "]",
+ log.getEvents().iterator().next().getValue());
+ }
+ assertEquals(
+ "message=Exchange[ExchangePattern: InOnly, BodyType: String, Body: " + expectedBodyAsync + "]",
+ asyncLog.getEvents().iterator().next().getValue());
+ }
+
+ @Override
+ protected RoutesBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ context.addComponent("async", new MyAsyncComponent());
+ from("direct:start")
+ .routeId("start")
+ .log("A direct message")
+ .to("log:info")
+ .recipientList(constant("direct:async"));
+
+ from("direct:async")
+ .to("async:bye:camel")
+ .routeId("async")
+ .log("An async message")
+ .to("log:tapped")
+ .to("mock:end");
+ }
+ };
+ }
+}
diff --git a/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/MyAsyncComponent.java b/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/MyAsyncComponent.java
new file mode 100644
index 00000000000..325381d21f4
--- /dev/null
+++ b/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/MyAsyncComponent.java
@@ -0,0 +1,46 @@
+/*
+ * 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.Map;
+
+import org.apache.camel.Endpoint;
+import org.apache.camel.support.DefaultComponent;
+import org.apache.camel.util.StringHelper;
+
+public class MyAsyncComponent extends DefaultComponent {
+
+ @Override
+ protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object> parameters) throws Exception {
+ MyAsyncEndpoint answer = new MyAsyncEndpoint(uri, this);
+ answer.setReply(prepareReply(remaining));
+ setProperties(answer, parameters);
+ return answer;
+ }
+
+ private String prepareReply(String value) {
+ // to make URIs valid we make the conventions of using ':' for ' ' and
+ // capitalize words
+ String[] words = value.split(":");
+ StringBuilder result = new StringBuilder();
+ for (String word : words) {
+ result.append(result.isEmpty() ? "" : " ");
+ result.append(StringHelper.capitalize(word));
+ }
+ return result.toString();
+ }
+}
diff --git a/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/MyAsyncEndpoint.java b/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/MyAsyncEndpoint.java
new file mode 100644
index 00000000000..fa0ce8f0eae
--- /dev/null
+++ b/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/MyAsyncEndpoint.java
@@ -0,0 +1,98 @@
+/*
+ * 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 org.apache.camel.Component;
+import org.apache.camel.Consumer;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.support.DefaultEndpoint;
+import org.apache.camel.support.SynchronousDelegateProducer;
+
+public class MyAsyncEndpoint extends DefaultEndpoint {
+
+ private boolean append;
+ private String reply;
+ private long delay = 25;
+ private int failFirstAttempts;
+ private boolean synchronous;
+
+ public MyAsyncEndpoint(String endpointUri, Component component) {
+ super(endpointUri, component);
+ }
+
+ @Override
+ public Producer createProducer() {
+ Producer answer = new MyAsyncProducer(this);
+ if (isSynchronous()) {
+ // force it to be synchronously
+ return new SynchronousDelegateProducer(answer);
+ } else {
+ return answer;
+ }
+ }
+
+ @Override
+ public Consumer createConsumer(Processor processor) {
+ throw new UnsupportedOperationException("Consumer not supported");
+ }
+
+ @Override
+ public boolean isSingleton() {
+ return true;
+ }
+
+ public boolean isSynchronous() {
+ return synchronous;
+ }
+
+ public void setSynchronous(boolean synchronous) {
+ this.synchronous = synchronous;
+ }
+
+ public String getReply() {
+ return reply;
+ }
+
+ public void setReply(String reply) {
+ this.reply = reply;
+ }
+
+ public long getDelay() {
+ return delay;
+ }
+
+ public void setDelay(long delay) {
+ this.delay = delay;
+ }
+
+ public int getFailFirstAttempts() {
+ return failFirstAttempts;
+ }
+
+ public void setFailFirstAttempts(int failFirstAttempts) {
+ this.failFirstAttempts = failFirstAttempts;
+ }
+
+ public boolean isAppend() {
+ return append;
+ }
+
+ public void setAppend(boolean append) {
+ this.append = append;
+ }
+}
diff --git a/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/MyAsyncProducer.java b/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/MyAsyncProducer.java
new file mode 100644
index 00000000000..9a16010e3c5
--- /dev/null
+++ b/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/MyAsyncProducer.java
@@ -0,0 +1,75 @@
+/*
+ * 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.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.CamelExchangeException;
+import org.apache.camel.Exchange;
+import org.apache.camel.support.DefaultAsyncProducer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class MyAsyncProducer extends DefaultAsyncProducer {
+
+ private static final Logger LOG = LoggerFactory.getLogger(MyAsyncProducer.class);
+ private final ExecutorService executor;
+ private final AtomicInteger counter = new AtomicInteger();
+
+ public MyAsyncProducer(MyAsyncEndpoint endpoint) {
+ super(endpoint);
+ this.executor = endpoint.getCamelContext().getExecutorServiceManager().newDefaultThreadPool(this, "MyProducer");
+ }
+
+ @Override
+ public MyAsyncEndpoint getEndpoint() {
+ return (MyAsyncEndpoint) super.getEndpoint();
+ }
+
+ @Override
+ public boolean process(final Exchange exchange, final AsyncCallback callback) {
+ executor.submit(new Callable<Object>() {
+ public Object call() throws Exception {
+ LOG.info("Simulating a task which takes {} millis to reply", getEndpoint().getDelay());
+ Thread.sleep(getEndpoint().getDelay());
+
+ int count = counter.incrementAndGet();
+ if (getEndpoint().getFailFirstAttempts() >= count) {
+ LOG.info("Simulating a failure at attempt {}", count);
+ exchange.setException(new CamelExchangeException("Simulated error at attempt " + count, exchange));
+ } else {
+ String reply = getEndpoint().getReply();
+ reply = getEndpoint().isAppend() ? exchange.getIn().getBody() + " " + reply : reply;
+ exchange.getMessage().setBody(reply);
+ LOG.info("Setting reply {}", reply);
+ }
+
+ LOG.info("Callback done(false)");
+ callback.done(false);
+ return null;
+ }
+ });
+
+ // indicate from this point forward its being routed asynchronously
+ LOG.info("Task submitted, now tell Camel routing engine to that this Exchange is being continued asynchronously");
+ return false;
+ }
+
+}
diff --git a/components/camel-opentelemetry2/src/test/java/org/apache/camel/opentelemetry2/AsyncTest.java b/components/camel-opentelemetry2/src/test/java/org/apache/camel/opentelemetry2/AsyncTest.java
new file mode 100644
index 00000000000..546ef24c226
--- /dev/null
+++ b/components/camel-opentelemetry2/src/test/java/org/apache/camel/opentelemetry2/AsyncTest.java
@@ -0,0 +1,179 @@
+/*
+ * 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.io.IOException;
+import java.util.List;
+import java.util.Map;
+
+import io.opentelemetry.api.common.AttributeKey;
+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.component.mock.MockEndpoint;
+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.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/*
+ * AsyncTest tests the execution of a new spin off async components.
+ */
+public class AsyncTest 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 testRouteMultipleRequests() throws InterruptedException, IOException {
+ int j = 10;
+ MockEndpoint mock = getMockEndpoint("mock:end");
+ mock.expectedMessageCount(j);
+ mock.setAssertPeriod(5000);
+ for (int i = 0; i < j; i++) {
+ context.createProducerTemplate().sendBody("direct:start", "Hello!");
+ }
+ mock.assertIsSatisfied(1000);
+ Map<String, OtelTrace> traces = otelExtension.getTraces();
+ // Each trace should have a unique trace id. It is enough to assert that
+ // the number of elements in the map is the same of the requests to prove
+ // all traces have been generated uniquely.
+ assertEquals(j, traces.size());
+ // Each trace should have the same structure
+ for (OtelTrace trace : traces.values()) {
+ checkTrace(trace, "Hello!");
+ }
+
+ }
+
+ private void checkTrace(OtelTrace trace, String expectedBody) {
+ List<SpanData> spans = trace.getSpans();
+ assertEquals(8, spans.size());
+ SpanData testProducer = OpenTelemetryTracerTestSupport.getSpan(spans, "direct://start", Op.EVENT_SENT);
+ SpanData direct = OpenTelemetryTracerTestSupport.getSpan(spans, "direct://start", Op.EVENT_RECEIVED);
+ SpanData asyncDirectTo = OpenTelemetryTracerTestSupport.getSpan(spans, "direct://async", Op.EVENT_SENT);
+ SpanData asyncDirectFrom = OpenTelemetryTracerTestSupport.getSpan(spans, "direct://async", Op.EVENT_RECEIVED);
+ SpanData async = OpenTelemetryTracerTestSupport.getSpan(spans, "async://bye:camel", Op.EVENT_SENT);
+ SpanData log = OpenTelemetryTracerTestSupport.getSpan(spans, "log://info", Op.EVENT_SENT);
+ SpanData asyncLog = OpenTelemetryTracerTestSupport.getSpan(spans, "log://tapped", Op.EVENT_SENT);
+ SpanData asyncMock = OpenTelemetryTracerTestSupport.getSpan(spans, "mock://end", Op.EVENT_SENT);
+
+ // Validate span completion
+ assertTrue(testProducer.hasEnded());
+ assertTrue(direct.hasEnded());
+ assertTrue(asyncDirectTo.hasEnded());
+ assertTrue(asyncDirectFrom.hasEnded());
+ assertTrue(async.hasEnded());
+ assertTrue(log.hasEnded());
+ assertTrue(asyncLog.hasEnded());
+ assertTrue(asyncMock.hasEnded());
+
+ // Validate same trace
+ assertEquals(testProducer.getSpanContext().getTraceId(), direct.getSpanContext().getTraceId());
+ assertEquals(testProducer.getSpanContext().getTraceId(), asyncDirectTo.getSpanContext().getTraceId());
+ assertEquals(testProducer.getSpanContext().getTraceId(), asyncDirectFrom.getSpanContext().getTraceId());
+ assertEquals(testProducer.getSpanContext().getTraceId(), async.getSpanContext().getTraceId());
+ assertEquals(testProducer.getSpanContext().getTraceId(), log.getSpanContext().getTraceId());
+ assertEquals(testProducer.getSpanContext().getTraceId(), asyncLog.getSpanContext().getTraceId());
+ assertEquals(testProducer.getSpanContext().getTraceId(), asyncMock.getSpanContext().getTraceId());
+
+ // Validate different Exchange ID
+ assertNotEquals(testProducer.getAttributes().get(AttributeKey.stringKey("exchangeId")),
+ asyncDirectTo.getAttributes().get(AttributeKey.stringKey("exchangeId")));
+ assertEquals(testProducer.getAttributes().get(AttributeKey.stringKey("exchangeId")),
+ direct.getAttributes().get(AttributeKey.stringKey("exchangeId")));
+ assertEquals(testProducer.getAttributes().get(AttributeKey.stringKey("exchangeId")),
+ log.getAttributes().get(AttributeKey.stringKey("exchangeId")));
+ assertEquals(asyncDirectTo.getAttributes().get(AttributeKey.stringKey("exchangeId")),
+ asyncDirectFrom.getAttributes().get(AttributeKey.stringKey("exchangeId")));
+ assertEquals(asyncDirectTo.getAttributes().get(AttributeKey.stringKey("exchangeId")),
+ async.getAttributes().get(AttributeKey.stringKey("exchangeId")));
+ assertEquals(asyncDirectTo.getAttributes().get(AttributeKey.stringKey("exchangeId")),
+ asyncLog.getAttributes().get(AttributeKey.stringKey("exchangeId")));
+ assertEquals(asyncDirectTo.getAttributes().get(AttributeKey.stringKey("exchangeId")),
+ asyncMock.getAttributes().get(AttributeKey.stringKey("exchangeId")));
+
+ // Validate hierarchy
+ assertFalse(testProducer.getParentSpanContext().isValid());
+ assertEquals(testProducer.getSpanContext().getSpanId(), direct.getParentSpanContext().getSpanId());
+ assertEquals(direct.getSpanContext().getSpanId(), log.getParentSpanContext().getSpanId());
+ assertEquals(direct.getSpanContext().getSpanId(), asyncDirectTo.getParentSpanContext().getSpanId());
+ assertEquals(asyncDirectTo.getSpanContext().getSpanId(), asyncDirectFrom.getParentSpanContext().getSpanId());
+ assertEquals(asyncDirectFrom.getSpanContext().getSpanId(), async.getParentSpanContext().getSpanId());
+ assertEquals(asyncDirectFrom.getSpanContext().getSpanId(), asyncLog.getParentSpanContext().getSpanId());
+ assertEquals(asyncDirectFrom.getSpanContext().getSpanId(), asyncMock.getParentSpanContext().getSpanId());
+
+ // Validate message logging
+ assertEquals("A direct message", direct.getEvents().get(0).getAttributes().get(
+ AttributeKey.stringKey("message")));
+ assertEquals("An async message", asyncDirectFrom.getEvents().get(0).getAttributes().get(
+ AttributeKey.stringKey("message")));
+ String expectedBodyAsync = "Bye Camel";
+ if (expectedBody == null) {
+ assertEquals(
+ "Exchange[ExchangePattern: InOut, BodyType: null, Body: [Body is null]]",
+ log.getEvents().get(0).getAttributes().get(
+ AttributeKey.stringKey("message")));
+ } else {
+ assertEquals(
+ "Exchange[ExchangePattern: InOnly, BodyType: String, Body: " + expectedBody + "]",
+ log.getEvents().get(0).getAttributes().get(
+ AttributeKey.stringKey("message")));
+ }
+
+ assertEquals(
+ "Exchange[ExchangePattern: InOnly, BodyType: String, Body: " + expectedBodyAsync + "]",
+ asyncLog.getEvents().get(0).getAttributes().get(
+ AttributeKey.stringKey("message")));
+ }
+
+ @Override
+ protected RoutesBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ context.addComponent("async", new MyAsyncComponent());
+ from("direct:start")
+ .routeId("start")
+ .log("A direct message")
+ .to("log:info")
+ .recipientList(constant("direct:async"));
+
+ from("direct:async")
+ .to("async:bye:camel")
+ .routeId("async")
+ .log("An async message")
+ .to("log:tapped")
+ .to("mock:end");
+ }
+ };
+ }
+}
diff --git a/components/camel-opentelemetry2/src/test/java/org/apache/camel/opentelemetry2/MyAsyncComponent.java b/components/camel-opentelemetry2/src/test/java/org/apache/camel/opentelemetry2/MyAsyncComponent.java
new file mode 100644
index 00000000000..bc5f57470c6
--- /dev/null
+++ b/components/camel-opentelemetry2/src/test/java/org/apache/camel/opentelemetry2/MyAsyncComponent.java
@@ -0,0 +1,46 @@
+/*
+ * 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.Map;
+
+import org.apache.camel.Endpoint;
+import org.apache.camel.support.DefaultComponent;
+import org.apache.camel.util.StringHelper;
+
+public class MyAsyncComponent extends DefaultComponent {
+
+ @Override
+ protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object> parameters) throws Exception {
+ MyAsyncEndpoint answer = new MyAsyncEndpoint(uri, this);
+ answer.setReply(prepareReply(remaining));
+ setProperties(answer, parameters);
+ return answer;
+ }
+
+ private String prepareReply(String value) {
+ // to make URIs valid we make the conventions of using ':' for ' ' and
+ // capitalize words
+ String[] words = value.split(":");
+ StringBuilder result = new StringBuilder();
+ for (String word : words) {
+ result.append(result.isEmpty() ? "" : " ");
+ result.append(StringHelper.capitalize(word));
+ }
+ return result.toString();
+ }
+}
diff --git a/components/camel-opentelemetry2/src/test/java/org/apache/camel/opentelemetry2/MyAsyncEndpoint.java b/components/camel-opentelemetry2/src/test/java/org/apache/camel/opentelemetry2/MyAsyncEndpoint.java
new file mode 100644
index 00000000000..e2862496880
--- /dev/null
+++ b/components/camel-opentelemetry2/src/test/java/org/apache/camel/opentelemetry2/MyAsyncEndpoint.java
@@ -0,0 +1,98 @@
+/*
+ * 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 org.apache.camel.Component;
+import org.apache.camel.Consumer;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.support.DefaultEndpoint;
+import org.apache.camel.support.SynchronousDelegateProducer;
+
+public class MyAsyncEndpoint extends DefaultEndpoint {
+
+ private boolean append;
+ private String reply;
+ private long delay = 25;
+ private int failFirstAttempts;
+ private boolean synchronous;
+
+ public MyAsyncEndpoint(String endpointUri, Component component) {
+ super(endpointUri, component);
+ }
+
+ @Override
+ public Producer createProducer() {
+ Producer answer = new MyAsyncProducer(this);
+ if (isSynchronous()) {
+ // force it to be synchronously
+ return new SynchronousDelegateProducer(answer);
+ } else {
+ return answer;
+ }
+ }
+
+ @Override
+ public Consumer createConsumer(Processor processor) {
+ throw new UnsupportedOperationException("Consumer not supported");
+ }
+
+ @Override
+ public boolean isSingleton() {
+ return true;
+ }
+
+ public boolean isSynchronous() {
+ return synchronous;
+ }
+
+ public void setSynchronous(boolean synchronous) {
+ this.synchronous = synchronous;
+ }
+
+ public String getReply() {
+ return reply;
+ }
+
+ public void setReply(String reply) {
+ this.reply = reply;
+ }
+
+ public long getDelay() {
+ return delay;
+ }
+
+ public void setDelay(long delay) {
+ this.delay = delay;
+ }
+
+ public int getFailFirstAttempts() {
+ return failFirstAttempts;
+ }
+
+ public void setFailFirstAttempts(int failFirstAttempts) {
+ this.failFirstAttempts = failFirstAttempts;
+ }
+
+ public boolean isAppend() {
+ return append;
+ }
+
+ public void setAppend(boolean append) {
+ this.append = append;
+ }
+}
diff --git a/components/camel-opentelemetry2/src/test/java/org/apache/camel/opentelemetry2/MyAsyncProducer.java b/components/camel-opentelemetry2/src/test/java/org/apache/camel/opentelemetry2/MyAsyncProducer.java
new file mode 100644
index 00000000000..15b1ef1215f
--- /dev/null
+++ b/components/camel-opentelemetry2/src/test/java/org/apache/camel/opentelemetry2/MyAsyncProducer.java
@@ -0,0 +1,75 @@
+/*
+ * 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.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.CamelExchangeException;
+import org.apache.camel.Exchange;
+import org.apache.camel.support.DefaultAsyncProducer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class MyAsyncProducer extends DefaultAsyncProducer {
+
+ private static final Logger LOG = LoggerFactory.getLogger(MyAsyncProducer.class);
+ private final ExecutorService executor;
+ private final AtomicInteger counter = new AtomicInteger();
+
+ public MyAsyncProducer(MyAsyncEndpoint endpoint) {
+ super(endpoint);
+ this.executor = endpoint.getCamelContext().getExecutorServiceManager().newDefaultThreadPool(this, "MyProducer");
+ }
+
+ @Override
+ public MyAsyncEndpoint getEndpoint() {
+ return (MyAsyncEndpoint) super.getEndpoint();
+ }
+
+ @Override
+ public boolean process(final Exchange exchange, final AsyncCallback callback) {
+ executor.submit(new Callable<Object>() {
+ public Object call() throws Exception {
+ LOG.info("Simulating a task which takes {} millis to reply", getEndpoint().getDelay());
+ Thread.sleep(getEndpoint().getDelay());
+
+ int count = counter.incrementAndGet();
+ if (getEndpoint().getFailFirstAttempts() >= count) {
+ LOG.info("Simulating a failure at attempt {}", count);
+ exchange.setException(new CamelExchangeException("Simulated error at attempt " + count, exchange));
+ } else {
+ String reply = getEndpoint().getReply();
+ reply = getEndpoint().isAppend() ? exchange.getIn().getBody() + " " + reply : reply;
+ exchange.getMessage().setBody(reply);
+ LOG.info("Setting reply {}", reply);
+ }
+
+ LOG.info("Callback done(false)");
+ callback.done(false);
+ return null;
+ }
+ });
+
+ // indicate from this point forward its being routed asynchronously
+ LOG.info("Task submitted, now tell Camel routing engine to that this Exchange is being continued asynchronously");
+ return false;
+ }
+
+}
diff --git a/components/camel-telemetry-dev/src/test/java/org/apache/camel/telemetrydev/AsyncTest.java b/components/camel-telemetry-dev/src/test/java/org/apache/camel/telemetrydev/AsyncTest.java
new file mode 100644
index 00000000000..ff794ef5a34
--- /dev/null
+++ b/components/camel-telemetry-dev/src/test/java/org/apache/camel/telemetrydev/AsyncTest.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.telemetrydev;
+
+import java.io.IOException;
+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.component.mock.MockEndpoint;
+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.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+/*
+ * AsyncTest tests the execution of a new spin off async components.
+ */
+public class AsyncTest 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 testRouteMultipleRequests() throws InterruptedException, IOException {
+ int j = 10;
+ MockEndpoint mock = getMockEndpoint("mock:end");
+ mock.expectedMessageCount(j);
+ mock.setAssertPeriod(5000);
+ for (int i = 0; i < j; i++) {
+ context.createProducerTemplate().sendBody("direct:start", "Hello!");
+ }
+ mock.assertIsSatisfied(1000);
+ Map<String, DevTrace> traces = tracesFromLog();
+ // Each trace should have a unique trace id. It is enough to assert that
+ // the number of elements in the map is the same of the requests to prove
+ // all traces have been generated uniquely.
+ assertEquals(j, traces.size());
+ // Each trace should have the same structure
+ for (DevTrace trace : traces.values()) {
+ checkTrace(trace, "Hello!");
+ }
+
+ }
+
+ private void checkTrace(DevTrace trace, String expectedBody) {
+ List<DevSpanAdapter> spans = trace.getSpans();
+ assertEquals(8, spans.size());
+ DevSpanAdapter testProducer = TelemetryDevTracerTestSupport.getSpan(spans, "direct://start", Op.EVENT_SENT);
+ DevSpanAdapter direct = TelemetryDevTracerTestSupport.getSpan(spans, "direct://start", Op.EVENT_RECEIVED);
+ DevSpanAdapter asyncDirectTo = TelemetryDevTracerTestSupport.getSpan(spans, "direct://async", Op.EVENT_SENT);
+ DevSpanAdapter asyncDirectFrom
+ = TelemetryDevTracerTestSupport.getSpan(spans, "direct://async", Op.EVENT_RECEIVED);
+ DevSpanAdapter async = TelemetryDevTracerTestSupport.getSpan(spans, "async://bye:camel", Op.EVENT_SENT);
+ DevSpanAdapter log = TelemetryDevTracerTestSupport.getSpan(spans, "log://info", Op.EVENT_SENT);
+ DevSpanAdapter asyncLog = TelemetryDevTracerTestSupport.getSpan(spans, "log://tapped", Op.EVENT_SENT);
+ DevSpanAdapter asyncMock = TelemetryDevTracerTestSupport.getSpan(spans, "mock://end", Op.EVENT_SENT);
+
+ // Validate span completion
+ assertEquals("true", testProducer.getTag("isDone"));
+ assertEquals("true", direct.getTag("isDone"));
+ assertEquals("true", asyncDirectTo.getTag("isDone"));
+ assertEquals("true", asyncDirectFrom.getTag("isDone"));
+ assertEquals("true", async.getTag("isDone"));
+ assertEquals("true", log.getTag("isDone"));
+ assertEquals("true", asyncLog.getTag("isDone"));
+ assertEquals("true", asyncMock.getTag("isDone"));
+
+ // Validate same trace
+ assertEquals(testProducer.getTag("traceid"), direct.getTag("traceid"));
+ assertEquals(testProducer.getTag("traceid"), asyncDirectTo.getTag("traceid"));
+ assertEquals(testProducer.getTag("traceid"), asyncDirectFrom.getTag("traceid"));
+ assertEquals(testProducer.getTag("traceid"), async.getTag("traceid"));
+ assertEquals(testProducer.getTag("traceid"), log.getTag("traceid"));
+ assertEquals(testProducer.getTag("traceid"), asyncLog.getTag("traceid"));
+ assertEquals(testProducer.getTag("traceid"), asyncMock.getTag("traceid"));
+
+ // Validate different Exchange ID
+ assertNotEquals(testProducer.getTag("exchangeId"), asyncDirectTo.getTag("exchangeId"));
+ assertEquals(testProducer.getTag("exchangeId"), direct.getTag("exchangeId"));
+ assertEquals(testProducer.getTag("exchangeId"), log.getTag("exchangeId"));
+ assertEquals(asyncDirectTo.getTag("exchangeId"), asyncDirectFrom.getTag("exchangeId"));
+ assertEquals(asyncDirectTo.getTag("exchangeId"), async.getTag("exchangeId"));
+ assertEquals(asyncDirectTo.getTag("exchangeId"), asyncLog.getTag("exchangeId"));
+ assertEquals(asyncDirectTo.getTag("exchangeId"), asyncMock.getTag("exchangeId"));
+
+ // Validate hierarchy
+ assertNull(testProducer.getTag("parentSpan"));
+ assertEquals(testProducer.getTag("spanid"), direct.getTag("parentSpan"));
+ assertEquals(direct.getTag("spanid"), log.getTag("parentSpan"));
+ assertEquals(direct.getTag("spanid"), asyncDirectTo.getTag("parentSpan"));
+ assertEquals(asyncDirectTo.getTag("spanid"), asyncDirectFrom.getTag("parentSpan"));
+ assertEquals(asyncDirectFrom.getTag("spanid"), async.getTag("parentSpan"));
+ assertEquals(asyncDirectFrom.getTag("spanid"), asyncLog.getTag("parentSpan"));
+ assertEquals(asyncDirectFrom.getTag("spanid"), asyncMock.getTag("parentSpan"));
+
+ // Validate message logging
+ assertEquals("A direct message", direct.getLogEntries().get(0).getFields().get("message"));
+ assertEquals("An async message", asyncDirectFrom.getLogEntries().get(0).getFields().get("message"));
+ String expectedAsyncBody = "Bye Camel";
+
+ if (expectedBody == null) {
+ assertEquals(
+ "Exchange[ExchangePattern: InOut, BodyType: null, Body: [Body is null]]",
+ log.getLogEntries().get(0).getFields().get("message"));
+ } else {
+ assertEquals(
+ "Exchange[ExchangePattern: InOnly, BodyType: String, Body: " + expectedBody + "]",
+ log.getLogEntries().get(0).getFields().get("message"));
+ }
+ assertEquals(
+ "Exchange[ExchangePattern: InOnly, BodyType: String, Body: " + expectedAsyncBody + "]",
+ asyncLog.getLogEntries().get(0).getFields().get("message"));
+ }
+
+ @Override
+ protected RoutesBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ context.addComponent("async", new MyAsyncComponent());
+ from("direct:start")
+ .routeId("start")
+ .log("A direct message")
+ .to("log:info")
+ .recipientList(constant("direct:async"));
+
+ from("direct:async")
+ .to("async:bye:camel")
+ .routeId("async")
+ .log("An async message")
+ .to("log:tapped")
+ .to("mock:end");
+ }
+ };
+ }
+}
diff --git a/components/camel-telemetry-dev/src/test/java/org/apache/camel/telemetrydev/MyAsyncComponent.java b/components/camel-telemetry-dev/src/test/java/org/apache/camel/telemetrydev/MyAsyncComponent.java
new file mode 100644
index 00000000000..d9fe2a4c230
--- /dev/null
+++ b/components/camel-telemetry-dev/src/test/java/org/apache/camel/telemetrydev/MyAsyncComponent.java
@@ -0,0 +1,46 @@
+/*
+ * 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.util.Map;
+
+import org.apache.camel.Endpoint;
+import org.apache.camel.support.DefaultComponent;
+import org.apache.camel.util.StringHelper;
+
+public class MyAsyncComponent extends DefaultComponent {
+
+ @Override
+ protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object> parameters) throws Exception {
+ MyAsyncEndpoint answer = new MyAsyncEndpoint(uri, this);
+ answer.setReply(prepareReply(remaining));
+ setProperties(answer, parameters);
+ return answer;
+ }
+
+ private String prepareReply(String value) {
+ // to make URIs valid we make the conventions of using ':' for ' ' and
+ // capitalize words
+ String[] words = value.split(":");
+ StringBuilder result = new StringBuilder();
+ for (String word : words) {
+ result.append(result.isEmpty() ? "" : " ");
+ result.append(StringHelper.capitalize(word));
+ }
+ return result.toString();
+ }
+}
diff --git a/components/camel-telemetry-dev/src/test/java/org/apache/camel/telemetrydev/MyAsyncEndpoint.java b/components/camel-telemetry-dev/src/test/java/org/apache/camel/telemetrydev/MyAsyncEndpoint.java
new file mode 100644
index 00000000000..48e2dd4c24b
--- /dev/null
+++ b/components/camel-telemetry-dev/src/test/java/org/apache/camel/telemetrydev/MyAsyncEndpoint.java
@@ -0,0 +1,98 @@
+/*
+ * 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 org.apache.camel.Component;
+import org.apache.camel.Consumer;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.support.DefaultEndpoint;
+import org.apache.camel.support.SynchronousDelegateProducer;
+
+public class MyAsyncEndpoint extends DefaultEndpoint {
+
+ private boolean append;
+ private String reply;
+ private long delay = 25;
+ private int failFirstAttempts;
+ private boolean synchronous;
+
+ public MyAsyncEndpoint(String endpointUri, Component component) {
+ super(endpointUri, component);
+ }
+
+ @Override
+ public Producer createProducer() {
+ Producer answer = new MyAsyncProducer(this);
+ if (isSynchronous()) {
+ // force it to be synchronously
+ return new SynchronousDelegateProducer(answer);
+ } else {
+ return answer;
+ }
+ }
+
+ @Override
+ public Consumer createConsumer(Processor processor) {
+ throw new UnsupportedOperationException("Consumer not supported");
+ }
+
+ @Override
+ public boolean isSingleton() {
+ return true;
+ }
+
+ public boolean isSynchronous() {
+ return synchronous;
+ }
+
+ public void setSynchronous(boolean synchronous) {
+ this.synchronous = synchronous;
+ }
+
+ public String getReply() {
+ return reply;
+ }
+
+ public void setReply(String reply) {
+ this.reply = reply;
+ }
+
+ public long getDelay() {
+ return delay;
+ }
+
+ public void setDelay(long delay) {
+ this.delay = delay;
+ }
+
+ public int getFailFirstAttempts() {
+ return failFirstAttempts;
+ }
+
+ public void setFailFirstAttempts(int failFirstAttempts) {
+ this.failFirstAttempts = failFirstAttempts;
+ }
+
+ public boolean isAppend() {
+ return append;
+ }
+
+ public void setAppend(boolean append) {
+ this.append = append;
+ }
+}
diff --git a/components/camel-telemetry-dev/src/test/java/org/apache/camel/telemetrydev/MyAsyncProducer.java b/components/camel-telemetry-dev/src/test/java/org/apache/camel/telemetrydev/MyAsyncProducer.java
new file mode 100644
index 00000000000..8a37657718e
--- /dev/null
+++ b/components/camel-telemetry-dev/src/test/java/org/apache/camel/telemetrydev/MyAsyncProducer.java
@@ -0,0 +1,75 @@
+/*
+ * 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.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.CamelExchangeException;
+import org.apache.camel.Exchange;
+import org.apache.camel.support.DefaultAsyncProducer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class MyAsyncProducer extends DefaultAsyncProducer {
+
+ private static final Logger LOG = LoggerFactory.getLogger(MyAsyncProducer.class);
+ private final ExecutorService executor;
+ private final AtomicInteger counter = new AtomicInteger();
+
+ public MyAsyncProducer(MyAsyncEndpoint endpoint) {
+ super(endpoint);
+ this.executor = endpoint.getCamelContext().getExecutorServiceManager().newDefaultThreadPool(this, "MyProducer");
+ }
+
+ @Override
+ public MyAsyncEndpoint getEndpoint() {
+ return (MyAsyncEndpoint) super.getEndpoint();
+ }
+
+ @Override
+ public boolean process(final Exchange exchange, final AsyncCallback callback) {
+ executor.submit(new Callable<Object>() {
+ public Object call() throws Exception {
+ LOG.info("Simulating a task which takes {} millis to reply", getEndpoint().getDelay());
+ Thread.sleep(getEndpoint().getDelay());
+
+ int count = counter.incrementAndGet();
+ if (getEndpoint().getFailFirstAttempts() >= count) {
+ LOG.info("Simulating a failure at attempt {}", count);
+ exchange.setException(new CamelExchangeException("Simulated error at attempt " + count, exchange));
+ } else {
+ String reply = getEndpoint().getReply();
+ reply = getEndpoint().isAppend() ? exchange.getIn().getBody() + " " + reply : reply;
+ exchange.getMessage().setBody(reply);
+ LOG.info("Setting reply {}", reply);
+ }
+
+ LOG.info("Callback done(false)");
+ callback.done(false);
+ return null;
+ }
+ });
+
+ // indicate from this point forward its being routed asynchronously
+ LOG.info("Task submitted, now tell Camel routing engine to that this Exchange is being continued asynchronously");
+ return false;
+ }
+
+}
diff --git a/components/camel-telemetry/src/test/java/org/apache/camel/telemetry/AsyncTest.java b/components/camel-telemetry/src/test/java/org/apache/camel/telemetry/AsyncTest.java
new file mode 100644
index 00000000000..862527b0979
--- /dev/null
+++ b/components/camel-telemetry/src/test/java/org/apache/camel/telemetry/AsyncTest.java
@@ -0,0 +1,170 @@
+/*
+ * 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.component.mock.MockEndpoint;
+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.junit5.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.assertNull;
+
+/*
+ * AsyncTest tests the execution of a new spin off async components.
+ */
+public class AsyncTest 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);
+ return context;
+ }
+
+ @Test
+ void testRouteMultipleRequests() throws InterruptedException {
+ int j = 10;
+ MockEndpoint mock = getMockEndpoint("mock:end");
+ mock.expectedMessageCount(j);
+ mock.setAssertPeriod(5000);
+ for (int i = 0; i < j; i++) {
+ context.createProducerTemplate().sendBody("direct:start", "Hello!");
+ }
+ mock.assertIsSatisfied(1000);
+ // We must wait a safe time to let the context completing the async writing to the log trace.
+ Thread.sleep(10000);
+ Map<String, MockTrace> traces = mockTracer.traces();
+ // Each trace should have a unique trace id. It is enough to assert that
+ // the number of elements in the map is the same of the requests to prove
+ // all traces have been generated uniquely.
+ assertEquals(j, traces.size());
+ // Each trace should have the same structure
+ for (MockTrace trace : traces.values()) {
+ // Body will be altered by the async component
+ checkTrace(trace, "Hello!");
+ }
+
+ }
+
+ private void checkTrace(MockTrace trace, String expectedBody) {
+ List<Span> spans = trace.spans();
+ assertEquals(8, spans.size());
+ // Cast to implementation object to be able to
+ // inspect the status of the Span.
+ MockSpanAdapter testProducer = SpanTestSupport.getSpan(spans, "direct://start", Op.EVENT_SENT);
+ MockSpanAdapter direct = SpanTestSupport.getSpan(spans, "direct://start", Op.EVENT_RECEIVED);
+ MockSpanAdapter asyncDirectTo = SpanTestSupport.getSpan(spans, "direct://async", Op.EVENT_SENT);
+ MockSpanAdapter asyncDirectFrom = SpanTestSupport.getSpan(spans, "direct://async", Op.EVENT_RECEIVED);
+ MockSpanAdapter async = SpanTestSupport.getSpan(spans, "async://bye:camel", Op.EVENT_SENT);
+ MockSpanAdapter log = SpanTestSupport.getSpan(spans, "log://info", Op.EVENT_SENT);
+ MockSpanAdapter asyncLog = SpanTestSupport.getSpan(spans, "log://tapped", Op.EVENT_SENT);
+ MockSpanAdapter asyncMock = SpanTestSupport.getSpan(spans, "mock://end", Op.EVENT_SENT);
+
+ // Validate span completion
+ assertEquals("true", testProducer.getTag("isDone"));
+ assertEquals("true", direct.getTag("isDone"));
+ assertEquals("true", asyncDirectTo.getTag("isDone"));
+ assertEquals("true", asyncDirectFrom.getTag("isDone"));
+ assertEquals("true", async.getTag("isDone"));
+ assertEquals("true", log.getTag("isDone"));
+ assertEquals("true", asyncLog.getTag("isDone"));
+ assertEquals("true", asyncMock.getTag("isDone"));
+
+ // Validate same trace
+ assertEquals(testProducer.getTag("traceid"), direct.getTag("traceid"));
+ assertEquals(testProducer.getTag("traceid"), asyncDirectTo.getTag("traceid"));
+ assertEquals(testProducer.getTag("traceid"), asyncDirectFrom.getTag("traceid"));
+ assertEquals(testProducer.getTag("traceid"), async.getTag("traceid"));
+ assertEquals(testProducer.getTag("traceid"), log.getTag("traceid"));
+ assertEquals(testProducer.getTag("traceid"), asyncLog.getTag("traceid"));
+ assertEquals(testProducer.getTag("traceid"), asyncMock.getTag("traceid"));
+
+ // Validate different Exchange ID
+ assertNotEquals(testProducer.getTag("exchangeId"), asyncDirectTo.getTag("exchangeId"));
+ assertEquals(testProducer.getTag("exchangeId"), direct.getTag("exchangeId"));
+ assertEquals(testProducer.getTag("exchangeId"), log.getTag("exchangeId"));
+ assertEquals(asyncDirectTo.getTag("exchangeId"), asyncDirectFrom.getTag("exchangeId"));
+ assertEquals(asyncDirectTo.getTag("exchangeId"), async.getTag("exchangeId"));
+ assertEquals(asyncDirectTo.getTag("exchangeId"), asyncLog.getTag("exchangeId"));
+ assertEquals(asyncDirectTo.getTag("exchangeId"), asyncMock.getTag("exchangeId"));
+
+ // Validate hierarchy
+ assertNull(testProducer.getTag("parentSpan"));
+ assertEquals(testProducer.getTag("spanid"), direct.getTag("parentSpan"));
+ assertEquals(direct.getTag("spanid"), log.getTag("parentSpan"));
+ assertEquals(direct.getTag("spanid"), asyncDirectTo.getTag("parentSpan"));
+ assertEquals(asyncDirectTo.getTag("spanid"), asyncDirectFrom.getTag("parentSpan"));
+ assertEquals(asyncDirectFrom.getTag("spanid"), async.getTag("parentSpan"));
+ assertEquals(asyncDirectFrom.getTag("spanid"), asyncLog.getTag("parentSpan"));
+ assertEquals(asyncDirectFrom.getTag("spanid"), asyncMock.getTag("parentSpan"));
+
+ // Validate message logging
+ assertEquals("A direct message", direct.logEntries().get(0).fields().get("message"));
+ assertEquals("An async message", asyncDirectFrom.logEntries().get(0).fields().get("message"));
+ String expectedAsyncBody = "Bye Camel";
+
+ if (expectedBody == null) {
+ assertEquals(
+ "Exchange[ExchangePattern: InOut, BodyType: null, Body: [Body is null]]",
+ log.logEntries().get(0).fields().get("message"));
+ } else {
+ assertEquals(
+ "Exchange[ExchangePattern: InOnly, BodyType: String, Body: " + expectedBody + "]",
+ log.logEntries().get(0).fields().get("message"));
+ }
+ assertEquals(
+ "Exchange[ExchangePattern: InOnly, BodyType: String, Body: " + expectedAsyncBody + "]",
+ asyncLog.logEntries().get(0).fields().get("message"));
+ }
+
+ @Override
+ protected RoutesBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ context.addComponent("async", new MyAsyncComponent());
+ from("direct:start")
+ .routeId("start")
+ .log("A direct message")
+ .to("log:info")
+ .recipientList(constant("direct:async"));
+
+ from("direct:async")
+ .to("async:bye:camel")
+ .routeId("async")
+ .log("An async message")
+ .to("log:tapped")
+ .to("mock:end");
+ }
+ };
+ }
+}
diff --git a/components/camel-telemetry/src/test/java/org/apache/camel/telemetry/MyAsyncComponent.java b/components/camel-telemetry/src/test/java/org/apache/camel/telemetry/MyAsyncComponent.java
new file mode 100644
index 00000000000..09d71196ccd
--- /dev/null
+++ b/components/camel-telemetry/src/test/java/org/apache/camel/telemetry/MyAsyncComponent.java
@@ -0,0 +1,46 @@
+/*
+ * 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.Map;
+
+import org.apache.camel.Endpoint;
+import org.apache.camel.support.DefaultComponent;
+import org.apache.camel.util.StringHelper;
+
+public class MyAsyncComponent extends DefaultComponent {
+
+ @Override
+ protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object> parameters) throws Exception {
+ MyAsyncEndpoint answer = new MyAsyncEndpoint(uri, this);
+ answer.setReply(prepareReply(remaining));
+ setProperties(answer, parameters);
+ return answer;
+ }
+
+ private String prepareReply(String value) {
+ // to make URIs valid we make the conventions of using ':' for ' ' and
+ // capitalize words
+ String[] words = value.split(":");
+ StringBuilder result = new StringBuilder();
+ for (String word : words) {
+ result.append(result.isEmpty() ? "" : " ");
+ result.append(StringHelper.capitalize(word));
+ }
+ return result.toString();
+ }
+}
diff --git a/components/camel-telemetry/src/test/java/org/apache/camel/telemetry/MyAsyncEndpoint.java b/components/camel-telemetry/src/test/java/org/apache/camel/telemetry/MyAsyncEndpoint.java
new file mode 100644
index 00000000000..48966538f83
--- /dev/null
+++ b/components/camel-telemetry/src/test/java/org/apache/camel/telemetry/MyAsyncEndpoint.java
@@ -0,0 +1,98 @@
+/*
+ * 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.Component;
+import org.apache.camel.Consumer;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.support.DefaultEndpoint;
+import org.apache.camel.support.SynchronousDelegateProducer;
+
+public class MyAsyncEndpoint extends DefaultEndpoint {
+
+ private boolean append;
+ private String reply;
+ private long delay = 25;
+ private int failFirstAttempts;
+ private boolean synchronous;
+
+ public MyAsyncEndpoint(String endpointUri, Component component) {
+ super(endpointUri, component);
+ }
+
+ @Override
+ public Producer createProducer() {
+ Producer answer = new MyAsyncProducer(this);
+ if (isSynchronous()) {
+ // force it to be synchronously
+ return new SynchronousDelegateProducer(answer);
+ } else {
+ return answer;
+ }
+ }
+
+ @Override
+ public Consumer createConsumer(Processor processor) {
+ throw new UnsupportedOperationException("Consumer not supported");
+ }
+
+ @Override
+ public boolean isSingleton() {
+ return true;
+ }
+
+ public boolean isSynchronous() {
+ return synchronous;
+ }
+
+ public void setSynchronous(boolean synchronous) {
+ this.synchronous = synchronous;
+ }
+
+ public String getReply() {
+ return reply;
+ }
+
+ public void setReply(String reply) {
+ this.reply = reply;
+ }
+
+ public long getDelay() {
+ return delay;
+ }
+
+ public void setDelay(long delay) {
+ this.delay = delay;
+ }
+
+ public int getFailFirstAttempts() {
+ return failFirstAttempts;
+ }
+
+ public void setFailFirstAttempts(int failFirstAttempts) {
+ this.failFirstAttempts = failFirstAttempts;
+ }
+
+ public boolean isAppend() {
+ return append;
+ }
+
+ public void setAppend(boolean append) {
+ this.append = append;
+ }
+}
diff --git a/components/camel-telemetry/src/test/java/org/apache/camel/telemetry/MyAsyncProducer.java b/components/camel-telemetry/src/test/java/org/apache/camel/telemetry/MyAsyncProducer.java
new file mode 100644
index 00000000000..07c520717cc
--- /dev/null
+++ b/components/camel-telemetry/src/test/java/org/apache/camel/telemetry/MyAsyncProducer.java
@@ -0,0 +1,75 @@
+/*
+ * 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.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.CamelExchangeException;
+import org.apache.camel.Exchange;
+import org.apache.camel.support.DefaultAsyncProducer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class MyAsyncProducer extends DefaultAsyncProducer {
+
+ private static final Logger LOG = LoggerFactory.getLogger(MyAsyncProducer.class);
+ private final ExecutorService executor;
+ private final AtomicInteger counter = new AtomicInteger();
+
+ public MyAsyncProducer(MyAsyncEndpoint endpoint) {
+ super(endpoint);
+ this.executor = endpoint.getCamelContext().getExecutorServiceManager().newDefaultThreadPool(this, "MyProducer");
+ }
+
+ @Override
+ public MyAsyncEndpoint getEndpoint() {
+ return (MyAsyncEndpoint) super.getEndpoint();
+ }
+
+ @Override
+ public boolean process(final Exchange exchange, final AsyncCallback callback) {
+ executor.submit(new Callable<Object>() {
+ public Object call() throws Exception {
+ LOG.info("Simulating a task which takes {} millis to reply", getEndpoint().getDelay());
+ Thread.sleep(getEndpoint().getDelay());
+
+ int count = counter.incrementAndGet();
+ if (getEndpoint().getFailFirstAttempts() >= count) {
+ LOG.info("Simulating a failure at attempt {}", count);
+ exchange.setException(new CamelExchangeException("Simulated error at attempt " + count, exchange));
+ } else {
+ String reply = getEndpoint().getReply();
+ reply = getEndpoint().isAppend() ? exchange.getIn().getBody() + " " + reply : reply;
+ exchange.getMessage().setBody(reply);
+ LOG.info("Setting reply {}", reply);
+ }
+
+ LOG.info("Callback done(false)");
+ callback.done(false);
+ return null;
+ }
+ });
+
+ // indicate from this point forward its being routed asynchronously
+ LOG.info("Task submitted, now tell Camel routing engine to that this Exchange is being continued asynchronously");
+ return false;
+ }
+
+}