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 5dfd6e6cc358 CAMEL-24036: Fix flaky AsyncWiretapTest across telemetry
modules
5dfd6e6cc358 is described below
commit 5dfd6e6cc3587b5c6292174944886e370dc10417
Author: Omar Atie <[email protected]>
AuthorDate: Wed Aug 5 02:14:20 2026 -0700
CAMEL-24036: Fix flaky AsyncWiretapTest across telemetry modules
Replace Thread.sleep, short mock timeouts, and missing trace waits with
Awaitility polling and MockEndpoint.assertIsSatisfied(30s) across
camel-telemetry, camel-telemetry-dev, camel-opentelemetry2, and
camel-micrometer-observability modules.
Closes #25331
Co-authored-by: Cursor Agent <[email protected]>
Co-authored-by: Claude Opus 4.6 <[email protected]>
---
.../micrometer/observability/AsyncWiretapTest.java | 37 +++++++++++++--------
.../camel/opentelemetry2/AsyncWiretapTest.java | 37 +++++++++++++--------
.../camel/telemetrydev/AsyncWiretapTest.java | 38 ++++++++++++----------
.../apache/camel/telemetry/AsyncWiretapTest.java | 38 ++++++++++++----------
4 files changed, 88 insertions(+), 62 deletions(-)
diff --git
a/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/AsyncWiretapTest.java
b/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/AsyncWiretapTest.java
index 225e9d109840..1d8918c893e0 100644
---
a/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/AsyncWiretapTest.java
+++
b/components/camel-micrometer-observability/src/test/java/org/apache/camel/micrometer/observability/AsyncWiretapTest.java
@@ -16,9 +16,9 @@
*/
package org.apache.camel.micrometer.observability;
-import java.io.IOException;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.TimeUnit;
import io.opentelemetry.api.common.AttributeKey;
import io.opentelemetry.api.trace.SpanId;
@@ -30,6 +30,7 @@ import
org.apache.camel.micrometer.observability.CamelOpenTelemetryExtension.Ote
import org.apache.camel.telemetry.Op;
import org.junit.jupiter.api.Test;
+import static org.awaitility.Awaitility.await;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -38,33 +39,41 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
* WiretappedRouteTest tests the execution of a new spin off component which
would create a new exchange,
* for example, using the wiretap component.
*/
-public class AsyncWiretapTest extends
MicrometerObservabilityTracerPropagationTestSupport {
+class AsyncWiretapTest extends
MicrometerObservabilityTracerPropagationTestSupport {
+
+ private static final int MESSAGE_COUNT = 10;
+ private static final int SPAN_COUNT = 7;
+ private static final long TIMEOUT_SECONDS = 30;
@Test
- void testRouteMultipleRequests() throws InterruptedException, IOException {
- int j = 10;
+ void testRouteMultipleRequests() throws Exception {
MockEndpoint mock = getMockEndpoint("mock:end");
- mock.expectedMessageCount(j);
- mock.setAssertPeriod(5000);
- for (int i = 0; i < j; i++) {
+ mock.expectedMessageCount(MESSAGE_COUNT);
+ for (int i = 0; i < MESSAGE_COUNT; i++) {
context.createProducerTemplate().sendBody("direct:start",
"Hello!");
}
- mock.assertIsSatisfied(1000);
+ MockEndpoint.assertIsSatisfied(context, TIMEOUT_SECONDS,
TimeUnit.SECONDS);
+ // Wait for async trace export to complete — spans are written
asynchronously
+ // after exchange processing, so we poll until all traces with
expected spans arrive.
+ await().atMost(TIMEOUT_SECONDS, TimeUnit.SECONDS)
+ .pollInterval(100, TimeUnit.MILLISECONDS)
+ .untilAsserted(() -> {
+ Map<String, OtelTrace> traces = otelExtension.getTraces();
+ assertEquals(MESSAGE_COUNT, traces.size());
+ for (OtelTrace trace : traces.values()) {
+ assertEquals(SPAN_COUNT, trace.getSpans().size());
+ }
+ });
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(7, spans.size());
+ assertEquals(SPAN_COUNT, spans.size());
SpanData testProducer = getSpan(spans, "direct://start",
Op.EVENT_SENT);
SpanData direct = getSpan(spans, "direct://start", Op.EVENT_RECEIVED);
SpanData wiretapDirectTo = getSpan(spans, "direct://tap",
Op.EVENT_SENT);
diff --git
a/components/camel-opentelemetry2/src/test/java/org/apache/camel/opentelemetry2/AsyncWiretapTest.java
b/components/camel-opentelemetry2/src/test/java/org/apache/camel/opentelemetry2/AsyncWiretapTest.java
index 81075b3b82d4..44220e59dc55 100644
---
a/components/camel-opentelemetry2/src/test/java/org/apache/camel/opentelemetry2/AsyncWiretapTest.java
+++
b/components/camel-opentelemetry2/src/test/java/org/apache/camel/opentelemetry2/AsyncWiretapTest.java
@@ -16,9 +16,9 @@
*/
package org.apache.camel.opentelemetry2;
-import java.io.IOException;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.TimeUnit;
import io.opentelemetry.api.common.AttributeKey;
import io.opentelemetry.sdk.trace.data.SpanData;
@@ -31,6 +31,7 @@ import
org.apache.camel.opentelemetry2.CamelOpenTelemetryExtension.OtelTrace;
import org.apache.camel.telemetry.Op;
import org.junit.jupiter.api.Test;
+import static org.awaitility.Awaitility.await;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
@@ -40,7 +41,11 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
* WiretappedRouteTest tests the execution of a new spin off component which
would create a new exchange, for example,
* using the wiretap component.
*/
-public class AsyncWiretapTest extends OpenTelemetryTracerTestSupport {
+class AsyncWiretapTest extends OpenTelemetryTracerTestSupport {
+
+ private static final int MESSAGE_COUNT = 10;
+ private static final int SPAN_COUNT = 7;
+ private static final long TIMEOUT_SECONDS = 30;
@Override
protected CamelContext createCamelContext() throws Exception {
@@ -55,30 +60,34 @@ public class AsyncWiretapTest extends
OpenTelemetryTracerTestSupport {
}
@Test
- void testRouteMultipleRequests() throws InterruptedException, IOException {
- int j = 10;
+ void testRouteMultipleRequests() throws Exception {
MockEndpoint mock = getMockEndpoint("mock:end");
- mock.expectedMessageCount(j);
- mock.setAssertPeriod(5000);
- for (int i = 0; i < j; i++) {
+ mock.expectedMessageCount(MESSAGE_COUNT);
+ for (int i = 0; i < MESSAGE_COUNT; i++) {
context.createProducerTemplate().sendBody("direct:start",
"Hello!");
}
- mock.assertIsSatisfied(1000);
+ MockEndpoint.assertIsSatisfied(context, TIMEOUT_SECONDS,
TimeUnit.SECONDS);
+ // Wait for async trace export to complete — spans are written
asynchronously
+ // after exchange processing, so we poll until all traces with
expected spans arrive.
+ await().atMost(TIMEOUT_SECONDS, TimeUnit.SECONDS)
+ .pollInterval(100, TimeUnit.MILLISECONDS)
+ .untilAsserted(() -> {
+ Map<String, OtelTrace> traces = otelExtension.getTraces();
+ assertEquals(MESSAGE_COUNT, traces.size());
+ for (OtelTrace trace : traces.values()) {
+ assertEquals(SPAN_COUNT, trace.getSpans().size());
+ }
+ });
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(7, spans.size());
+ assertEquals(SPAN_COUNT, spans.size());
SpanData testProducer = OpenTelemetryTracerTestSupport.getSpan(spans,
"direct://start", Op.EVENT_SENT);
SpanData direct = OpenTelemetryTracerTestSupport.getSpan(spans,
"direct://start", Op.EVENT_RECEIVED);
SpanData wiretapDirectTo =
OpenTelemetryTracerTestSupport.getSpan(spans, "direct://tap", Op.EVENT_SENT);
diff --git
a/components/camel-telemetry-dev/src/test/java/org/apache/camel/telemetrydev/AsyncWiretapTest.java
b/components/camel-telemetry-dev/src/test/java/org/apache/camel/telemetrydev/AsyncWiretapTest.java
index a7406f063d08..4f4b89777a1c 100644
---
a/components/camel-telemetry-dev/src/test/java/org/apache/camel/telemetrydev/AsyncWiretapTest.java
+++
b/components/camel-telemetry-dev/src/test/java/org/apache/camel/telemetrydev/AsyncWiretapTest.java
@@ -37,7 +37,11 @@ import static org.junit.jupiter.api.Assertions.assertNull;
* WiretappedRouteTest tests the execution of a new spin off component which
would create a new exchange, for example,
* using the wiretap component.
*/
-public class AsyncWiretapTest extends TelemetryDevTracerTestSupport {
+class AsyncWiretapTest extends TelemetryDevTracerTestSupport {
+
+ private static final int MESSAGE_COUNT = 10;
+ private static final int SPAN_COUNT = 7;
+ private static final long TIMEOUT_SECONDS = 30;
@Override
protected CamelContext createCamelContext() throws Exception {
@@ -52,27 +56,27 @@ public class AsyncWiretapTest extends
TelemetryDevTracerTestSupport {
@Test
void testRouteMultipleRequests() throws Exception {
- int j = 10;
MockEndpoint mock = getMockEndpoint("mock:end");
- mock.expectedMessageCount(j);
- for (int i = 0; i < j; i++) {
+ mock.expectedMessageCount(MESSAGE_COUNT);
+ for (int i = 0; i < MESSAGE_COUNT; i++) {
context.createProducerTemplate().sendBody("direct:start",
"Hello!");
}
- MockEndpoint.assertIsSatisfied(context, 30, TimeUnit.SECONDS);
+ MockEndpoint.assertIsSatisfied(context, TIMEOUT_SECONDS,
TimeUnit.SECONDS);
// Wait for async trace writing to complete — traces are written
asynchronously
// after exchange processing, so we poll until all traces with
expected spans arrive.
- await().atMost(30, TimeUnit.SECONDS).untilAsserted(() -> {
- Map<String, DevTrace> traces = tracesFromLog();
- assertEquals(j, traces.size());
- for (DevTrace trace : traces.values()) {
- assertEquals(7, trace.getSpans().size());
- }
- });
+ await().atMost(TIMEOUT_SECONDS, TimeUnit.SECONDS)
+ .pollInterval(100, TimeUnit.MILLISECONDS)
+ .untilAsserted(() -> {
+ Map<String, DevTrace> traces = tracesFromLog();
+ assertEquals(MESSAGE_COUNT, traces.size());
+ for (DevTrace trace : traces.values()) {
+ assertEquals(SPAN_COUNT, trace.getSpans().size());
+ for (DevSpanAdapter span : trace.getSpans()) {
+ assertEquals("true", span.getTag("isDone"));
+ }
+ }
+ });
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!");
@@ -81,7 +85,7 @@ public class AsyncWiretapTest extends
TelemetryDevTracerTestSupport {
private void checkTrace(DevTrace trace, String expectedBody) {
List<DevSpanAdapter> spans = trace.getSpans();
- assertEquals(7, spans.size());
+ assertEquals(SPAN_COUNT, spans.size());
DevSpanAdapter testProducer =
TelemetryDevTracerTestSupport.getSpan(spans, "direct://start", Op.EVENT_SENT);
DevSpanAdapter direct = TelemetryDevTracerTestSupport.getSpan(spans,
"direct://start", Op.EVENT_RECEIVED);
DevSpanAdapter wiretapDirectTo =
TelemetryDevTracerTestSupport.getSpan(spans, "direct://tap", Op.EVENT_SENT);
diff --git
a/components/camel-telemetry/src/test/java/org/apache/camel/telemetry/AsyncWiretapTest.java
b/components/camel-telemetry/src/test/java/org/apache/camel/telemetry/AsyncWiretapTest.java
index f7988573f4a3..291bc98dc293 100644
---
a/components/camel-telemetry/src/test/java/org/apache/camel/telemetry/AsyncWiretapTest.java
+++
b/components/camel-telemetry/src/test/java/org/apache/camel/telemetry/AsyncWiretapTest.java
@@ -40,7 +40,11 @@ import static org.junit.jupiter.api.Assertions.assertNull;
* WiretappedRouteTest tests the execution of a new spin off component which
would create a new exchange, for example,
* using the wiretap component.
*/
-public class AsyncWiretapTest extends ExchangeTestSupport {
+class AsyncWiretapTest extends ExchangeTestSupport {
+
+ private static final int MESSAGE_COUNT = 10;
+ private static final int SPAN_COUNT = 7;
+ private static final long TIMEOUT_SECONDS = 30;
MockTracer mockTracer;
@@ -56,27 +60,27 @@ public class AsyncWiretapTest extends ExchangeTestSupport {
@Test
void testRouteMultipleRequests() throws Exception {
- int j = 10;
MockEndpoint mock = getMockEndpoint("mock:end");
- mock.expectedMessageCount(j);
- for (int i = 0; i < j; i++) {
+ mock.expectedMessageCount(MESSAGE_COUNT);
+ for (int i = 0; i < MESSAGE_COUNT; i++) {
context.createProducerTemplate().sendBody("direct:start",
"Hello!");
}
- MockEndpoint.assertIsSatisfied(context, 30, TimeUnit.SECONDS);
+ MockEndpoint.assertIsSatisfied(context, TIMEOUT_SECONDS,
TimeUnit.SECONDS);
// Wait for async trace writing to complete — traces are written
asynchronously
// after exchange processing, so we poll until all traces with
expected spans arrive.
- await().atMost(30, TimeUnit.SECONDS).untilAsserted(() -> {
- Map<String, MockTrace> traces = mockTracer.traces();
- assertEquals(j, traces.size());
- for (MockTrace trace : traces.values()) {
- assertEquals(7, trace.spans().size());
- }
- });
+ await().atMost(TIMEOUT_SECONDS, TimeUnit.SECONDS)
+ .pollInterval(100, TimeUnit.MILLISECONDS)
+ .untilAsserted(() -> {
+ Map<String, MockTrace> traces = mockTracer.traces();
+ assertEquals(MESSAGE_COUNT, traces.size());
+ for (MockTrace trace : traces.values()) {
+ assertEquals(SPAN_COUNT, trace.spans().size());
+ for (Span span : trace.spans()) {
+ assertEquals("true", ((MockSpanAdapter)
span).getTag("isDone"));
+ }
+ }
+ });
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()) {
checkTrace(trace, "Hello!");
@@ -85,7 +89,7 @@ public class AsyncWiretapTest extends ExchangeTestSupport {
private void checkTrace(MockTrace trace, String expectedBody) {
List<Span> spans = trace.spans();
- assertEquals(7, spans.size());
+ assertEquals(SPAN_COUNT, 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);
