You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2022/12/16 08:30:44 UTC
[camel] branch main updated: CAMEL-18661: Make span current and clear scope properly for async processing (#8713)
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 002c9ca5824 CAMEL-18661: Make span current and clear scope properly for async processing (#8713)
002c9ca5824 is described below
commit 002c9ca5824e06bd7d11d1ed0d70cd52ddfc0091
Author: Liudmila Molkova <li...@microsoft.com>
AuthorDate: Fri Dec 16 00:30:35 2022 -0800
CAMEL-18661: Make span current and clear scope properly for async processing (#8713)
* Prototype: make span current and clear scope
* cleanup
* tests and fixes
* cleanup and tests
* cleanup
* fix build
* Remove unnecessary notifications
* fix tests
* oops
* Fix tests on Java 11
* checkstyle
---
.../camel/cdi/test/CamelEventNotifierTest.java | 5 +
.../opentelemetry/OpenTelemetrySpanAdapter.java | 5 +
.../CamelOpenTelemetryTestSupport.java | 24 +-
.../camel/opentelemetry/CurrentSpanTest.java | 329 +++++++++++++++++++++
.../apache/camel/tracing/ActiveSpanManager.java | 36 ++-
.../java/org/apache/camel/tracing/SpanAdapter.java | 4 +
.../main/java/org/apache/camel/tracing/Tracer.java | 22 +-
.../camel/tracing/ActiveSpanManagerTest.java | 24 +-
.../org/apache/camel/tracing/MockSpanAdapter.java | 21 ++
.../main/java/org/apache/camel/spi/CamelEvent.java | 7 +
.../java/org/apache/camel/spi/EventFactory.java | 11 +-
.../java/org/apache/camel/spi/EventNotifier.java | 6 +
.../camel/impl/engine/CamelInternalProcessor.java | 5 +
.../camel/impl/event/DefaultEventFactory.java | 9 +
.../event/ExchangeAsyncProcessingStartedEvent.java | 42 +++
.../org/apache/camel/processor/SendProcessor.java | 6 +-
.../camel/impl/event/EventNotifierEventsTest.java | 66 +++--
.../impl/event/EventNotifierExchangeSentTest.java | 2 +
.../EventNotifierFailureHandledEventsTest.java | 5 +
.../event/EventNotifierRedeliveryEventsTest.java | 1 +
.../event/MultipleEventNotifierEventsTest.java | 92 +++---
.../camel/processor/interceptor/DebugTest.java | 2 +-
.../camel/management/AddEventNotifierTest.java | 4 +-
.../JmxNotificationEventNotifierTest.java | 4 +-
.../camel/management/PublishEventNotifierTest.java | 4 +-
.../camel/management/RemoveEventNotifierTest.java | 4 +-
.../java/org/apache/camel/support/EventHelper.java | 47 ++-
.../camel/support/SimpleEventNotifierSupport.java | 11 +
28 files changed, 698 insertions(+), 100 deletions(-)
diff --git a/components/camel-cdi/src/test/java/org/apache/camel/cdi/test/CamelEventNotifierTest.java b/components/camel-cdi/src/test/java/org/apache/camel/cdi/test/CamelEventNotifierTest.java
index 7b0a913a52c..ca889a10242 100644
--- a/components/camel-cdi/src/test/java/org/apache/camel/cdi/test/CamelEventNotifierTest.java
+++ b/components/camel-cdi/src/test/java/org/apache/camel/cdi/test/CamelEventNotifierTest.java
@@ -36,6 +36,7 @@ import org.apache.camel.spi.CamelEvent.CamelContextStartedEvent;
import org.apache.camel.spi.CamelEvent.CamelContextStartingEvent;
import org.apache.camel.spi.CamelEvent.CamelContextStoppedEvent;
import org.apache.camel.spi.CamelEvent.CamelContextStoppingEvent;
+import org.apache.camel.spi.CamelEvent.ExchangeAsyncProcessingStartedEvent;
import org.apache.camel.spi.CamelEvent.ExchangeCompletedEvent;
import org.apache.camel.spi.CamelEvent.ExchangeCreatedEvent;
import org.apache.camel.spi.CamelEvent.ExchangeEvent;
@@ -138,9 +139,11 @@ public class CamelEventNotifierTest {
CamelContextStartedEvent.class,
ExchangeSendingEvent.class,
ExchangeCreatedEvent.class,
+ ExchangeAsyncProcessingStartedEvent.class,
ExchangeSendingEvent.class,
ExchangeSentEvent.class,
ExchangeCompletedEvent.class,
+ ExchangeAsyncProcessingStartedEvent.class,
ExchangeSentEvent.class));
}
@@ -157,9 +160,11 @@ public class CamelEventNotifierTest {
CamelContextStartedEvent.class,
ExchangeSendingEvent.class,
ExchangeCreatedEvent.class,
+ ExchangeAsyncProcessingStartedEvent.class,
ExchangeSendingEvent.class,
ExchangeSentEvent.class,
ExchangeCompletedEvent.class,
+ ExchangeAsyncProcessingStartedEvent.class,
ExchangeSentEvent.class,
CamelContextStoppingEvent.class,
CamelContextStoppedEvent.class));
diff --git a/components/camel-opentelemetry/src/main/java/org/apache/camel/opentelemetry/OpenTelemetrySpanAdapter.java b/components/camel-opentelemetry/src/main/java/org/apache/camel/opentelemetry/OpenTelemetrySpanAdapter.java
index 780b9ffe383..8ea66f28a55 100644
--- a/components/camel-opentelemetry/src/main/java/org/apache/camel/opentelemetry/OpenTelemetrySpanAdapter.java
+++ b/components/camel-opentelemetry/src/main/java/org/apache/camel/opentelemetry/OpenTelemetrySpanAdapter.java
@@ -108,6 +108,11 @@ public class OpenTelemetrySpanAdapter implements SpanAdapter {
return span.getSpanContext().getSpanId();
}
+ @Override
+ public AutoCloseable makeCurrent() {
+ return span.makeCurrent();
+ }
+
String getEventNameFromFields(Map<String, ?> fields) {
Object eventValue = fields == null ? null : fields.get("event");
if (eventValue != null) {
diff --git a/components/camel-opentelemetry/src/test/java/org/apache/camel/opentelemetry/CamelOpenTelemetryTestSupport.java b/components/camel-opentelemetry/src/test/java/org/apache/camel/opentelemetry/CamelOpenTelemetryTestSupport.java
index f34a2382442..93fe09d23f1 100644
--- a/components/camel-opentelemetry/src/test/java/org/apache/camel/opentelemetry/CamelOpenTelemetryTestSupport.java
+++ b/components/camel-opentelemetry/src/test/java/org/apache/camel/opentelemetry/CamelOpenTelemetryTestSupport.java
@@ -48,13 +48,13 @@ class CamelOpenTelemetryTestSupport extends CamelTestSupport {
static final AttributeKey<String> MESSAGE_KEY = AttributeKey.stringKey("message");
private InMemorySpanExporter inMemorySpanExporter = InMemorySpanExporter.create();
- private SpanTestData[] testdata;
+ private SpanTestData[] expected;
private Tracer tracer;
private OpenTelemetryTracer ottracer;
private SdkTracerProvider tracerFactory;
- CamelOpenTelemetryTestSupport(SpanTestData[] testdata) {
- this.testdata = testdata;
+ CamelOpenTelemetryTestSupport(SpanTestData[] expected) {
+ this.expected = expected;
}
@Override
@@ -78,10 +78,14 @@ class CamelOpenTelemetryTestSupport extends CamelTestSupport {
}
protected void verify() {
- verify(false);
+ verify(expected, false);
}
protected void verify(boolean async) {
+ verify(expected, async);
+ }
+
+ protected List<SpanData> verify(SpanTestData[] expected, boolean async) {
List<SpanData> spans = inMemorySpanExporter.getFinishedSpanItems();
spans.forEach(mockSpan -> {
System.out.println("Span: " + mockSpan);
@@ -90,19 +94,21 @@ class CamelOpenTelemetryTestSupport extends CamelTestSupport {
System.out.println("\tLogs: ");
});
- assertEquals(testdata.length, spans.size(), "Incorrect number of spans");
+ assertEquals(expected.length, spans.size(), "Incorrect number of spans");
verifySameTrace();
if (async) {
final List<SpanData> unsortedSpans = spans;
- spans = Arrays.stream(testdata)
+ spans = Arrays.stream(expected)
.map(td -> findSpan(td, unsortedSpans)).distinct().collect(Collectors.toList());
- assertEquals(testdata.length, spans.size(), "Incorrect number of spans after sorting");
+ assertEquals(expected.length, spans.size(), "Incorrect number of spans after sorting");
}
- for (int i = 0; i < testdata.length; i++) {
- verifySpan(i, testdata, spans);
+ for (int i = 0; i < expected.length; i++) {
+ verifySpan(i, expected, spans);
}
+
+ return spans;
}
protected SpanData findSpan(SpanTestData testdata, List<SpanData> spans) {
diff --git a/components/camel-opentelemetry/src/test/java/org/apache/camel/opentelemetry/CurrentSpanTest.java b/components/camel-opentelemetry/src/test/java/org/apache/camel/opentelemetry/CurrentSpanTest.java
new file mode 100644
index 00000000000..9caf8792b16
--- /dev/null
+++ b/components/camel-opentelemetry/src/test/java/org/apache/camel/opentelemetry/CurrentSpanTest.java
@@ -0,0 +1,329 @@
+/*
+ * 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.opentelemetry;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ForkJoinPool;
+import java.util.concurrent.TimeUnit;
+
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.SpanKind;
+import io.opentelemetry.sdk.trace.ReadableSpan;
+import io.opentelemetry.sdk.trace.data.SpanData;
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.Consumer;
+import org.apache.camel.Endpoint;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.RoutesBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockComponent;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.support.DefaultAsyncProducer;
+import org.apache.camel.support.DefaultConsumer;
+import org.apache.camel.support.DefaultProducer;
+import org.apache.camel.tracing.ActiveSpanManager;
+import org.apache.camel.util.StopWatch;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+
+class CurrentSpanTest extends CamelOpenTelemetryTestSupport {
+ private static final Executor DELAYED = CompletableFuture.delayedExecutor(10L, TimeUnit.MILLISECONDS, new ForkJoinPool(3));
+
+ CurrentSpanTest() {
+ super(new SpanTestData[0]);
+ }
+
+ @Test
+ void testSync() {
+ SpanTestData[] expectedSpans = {
+ new SpanTestData().setLabel("syncmock:result").setUri("syncmock://result").setOperation("syncmock")
+ .setKind(SpanKind.CLIENT),
+ new SpanTestData().setLabel("direct:bar").setUri("direct://bar").setOperation("bar").setKind(SpanKind.INTERNAL),
+ };
+
+ // sync pipeline
+ template.sendBody("direct:bar", "Hello World");
+
+ List<SpanData> spans = verify(expectedSpans, false);
+ assertEquals(spans.get(0).getParentSpanId(), spans.get(1).getSpanId());
+
+ // validates that span was active in async producer's processor
+ assertFalse(Span.current().getSpanContext().isValid());
+ }
+
+ @Test
+ void testSyncToAsync() {
+ SpanTestData[] expectedSpans = {
+ new SpanTestData().setLabel("asyncmock1:result").setUri("asyncmock1://result").setOperation("asyncmock1")
+ .setKind(SpanKind.CLIENT),
+ new SpanTestData().setLabel("direct:foo").setUri("direct://foo").setOperation("foo").setKind(SpanKind.INTERNAL),
+ };
+
+ // sync to async pipeline
+ template.sendBody("direct:foo", "Hello World");
+
+ List<SpanData> spans = verify(expectedSpans, false);
+ assertEquals(spans.get(0).getParentSpanId(), spans.get(1).getSpanId());
+
+ // context is cleaned up
+ assertFalse(Span.current().getSpanContext().isValid());
+ }
+
+ @Test
+ void testAsyncToSync() {
+ // direct client spans (event spans) are not created, so we saw only two spans in previous tests
+ SpanTestData[] expectedSpans = {
+ new SpanTestData().setLabel("syncmock:result").setUri("syncmock://result").setOperation("syncmock")
+ .setKind(SpanKind.CLIENT),
+ new SpanTestData().setLabel("asyncmock1:start").setUri("asyncmock1://start").setOperation("asyncmock1")
+ .setKind(SpanKind.INTERNAL),
+ new SpanTestData().setLabel("asyncmock1:start").setUri("asyncmock1://start").setOperation("asyncmock1")
+ .setKind(SpanKind.CLIENT),
+ };
+
+ // sync pipeline
+ template.sendBody("asyncmock1:start", "Hello World");
+
+ List<SpanData> spans = verify(expectedSpans, false);
+ assertEquals(spans.get(0).getParentSpanId(), spans.get(1).getSpanId());
+ assertFalse(Span.current().getSpanContext().isValid());
+ }
+
+ @Test
+ void testAsyncToAsync() {
+ SpanTestData[] expectedSpans = {
+ new SpanTestData().setLabel("asyncmock2:result").setUri("asyncmock2://result").setOperation("asyncmock2")
+ .setKind(SpanKind.CLIENT),
+ new SpanTestData().setLabel("asyncmock2:start").setUri("asyncmock2://start").setOperation("asyncmock2")
+ .setKind(SpanKind.INTERNAL),
+ new SpanTestData().setLabel("asyncmock2:start").setUri("asyncmock2://start").setOperation("asyncmock2")
+ .setKind(SpanKind.CLIENT),
+ };
+
+ // sync pipeline
+ template.sendBody("asyncmock2:start", "Hello World");
+
+ List<SpanData> spans = verify(expectedSpans, false);
+ assertEquals(spans.get(0).getParentSpanId(), spans.get(1).getSpanId());
+ assertFalse(Span.current().getSpanContext().isValid());
+ }
+
+ @Test
+ void testMulticastAsync() {
+ SpanTestData[] expectedSpans = {
+ new SpanTestData().setLabel("asyncmock1:result").setUri("asyncmock1://result").setOperation("asyncmock1")
+ .setKind(SpanKind.CLIENT),
+ new SpanTestData().setLabel("asyncmock2:result").setUri("asyncmock2://result").setOperation("asyncmock2")
+ .setKind(SpanKind.CLIENT),
+ new SpanTestData().setLabel("syncmock:result").setUri("syncmock://result").setOperation("syncmock")
+ .setKind(SpanKind.CLIENT),
+ new SpanTestData().setLabel("direct:start").setUri("direct://start").setOperation("start")
+ .setKind(SpanKind.INTERNAL)
+ };
+
+ // sync pipeline
+ template.sendBody("direct:start", "Hello World");
+
+ List<SpanData> spans = verify(expectedSpans, false);
+ assertEquals(spans.get(0).getParentSpanId(), spans.get(3).getSpanId());
+ assertEquals(spans.get(1).getParentSpanId(), spans.get(3).getSpanId());
+ assertEquals(spans.get(2).getParentSpanId(), spans.get(3).getSpanId());
+ assertFalse(Span.current().getSpanContext().isValid());
+ }
+
+ @Test
+ void testContextDoesNotLeak() {
+ for (int i = 0; i < 30; i++) {
+ template.sendBody("asyncmock3:start", String.valueOf(i));
+ assertFalse(Span.current().getSpanContext().isValid());
+ }
+
+ verifyTraceSpanNumbers(30, 10);
+ }
+
+ @Override
+ protected RoutesBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ context.addComponent("asyncmock1", new AsyncMockComponent());
+ context.addComponent("asyncmock2", new AsyncMockComponent());
+ context.addComponent("asyncmock3", new AsyncMockComponent());
+ context.addComponent("syncmock", new SyncMockComponent());
+
+ // sync pipeline
+ from("direct:bar").to("syncmock:result");
+
+ // sync to async pipeline
+ from("direct:foo").to("asyncmock1:result");
+
+ // async to sync pipeline
+ from("asyncmock1:start").to("syncmock:result");
+
+ // async pipeline
+ from("asyncmock2:start").to("asyncmock2:result");
+
+ // multicast pipeline
+ from("direct:start").multicast()
+ .to("asyncmock1:result")
+ .to("asyncmock2:result")
+ .to("syncmock:result");
+
+ // stress pipeline
+ from("asyncmock3:start").multicast()
+ .aggregationStrategy((oldExchange, newExchange) -> {
+ // context should be cleaned up
+ // BUT
+ // we have a stack of spans for this pipeline:
+ // root is producer (asyncmock3:start) and a bunch of nested under each other successors, e.g:
+ // - consumer asyncmock3:start
+ // - producer asyncmock2:start
+ // - consumer asyncmock2:start
+ // -producer asyncmock2:result
+ // the root span is still current during aggregation on *some* thread. It's also still running.
+ //
+ // OTel instrumentation for executor service should take care of propagation of
+ // current asyncmock3:start span when possible, but it's not enabled here.
+ //
+ // So we can have either no context, or, accidentally have asyncmock3:start, which is also valid.
+ // hence the condition here:
+ if (Span.current().getSpanContext().isValid()) {
+ ReadableSpan readable = (ReadableSpan) Span.current();
+ if (readable.hasEnded()) {
+ System.out.printf("Detected current ended span: name - '%s', parent id - '%s'",
+ readable.getName(), readable.getParentSpanContext().getSpanId());
+ }
+ // we must never get current, but ended span.
+ assertFalse(readable.hasEnded());
+ assertEquals("asyncmock3", readable.getName());
+ }
+ return newExchange;
+ })
+ .executorService(Executors.newFixedThreadPool(10))
+ .parallelProcessing()
+ .streaming()
+ .delay(10)
+ .to("log:line", "asyncmock1:start")
+ .to("log:line", "asyncmock2:start")
+ .to("log:line", "direct:bar")
+ .process(ignored -> assertFalse(Span.current().getSpanContext().isValid()));
+ }
+ };
+ }
+
+ private class AsyncMockComponent extends MockComponent {
+
+ @Override
+ protected Endpoint createEndpoint(String uri, String key, Map<String, Object> parameters) {
+ return new AsyncMockEndpoint(this, uri, key);
+ }
+ }
+
+ private static class AsyncMockEndpoint extends MockEndpoint {
+ private Consumer consumer;
+ private final String key;
+
+ public AsyncMockEndpoint(AsyncMockComponent component, String uri, String key) {
+ super(uri, component);
+ this.key = key;
+ }
+
+ @Override
+ public Consumer createConsumer(Processor processor) {
+ consumer = new DefaultConsumer(this, exchange -> {
+ assertCurrentSpan(exchange);
+ processor.process(exchange);
+ });
+ try {
+ configureConsumer(consumer);
+ } catch (Exception e) {
+ // ignore
+ }
+ return consumer;
+ }
+
+ @Override
+ public Producer createProducer() {
+ return new DefaultAsyncProducer(this) {
+ @Override
+ public boolean process(Exchange exchange, AsyncCallback callback) {
+ assertCurrentSpan(exchange);
+ if (!key.equals("result")) {
+ try {
+ getConsumer(1000).getProcessor().process(exchange);
+ } catch (Exception e) {
+ fail(e);
+ }
+ }
+ CompletableFuture.runAsync(() -> {
+ }, DELAYED)
+ .thenRun(() -> callback.run());
+
+ return false;
+ }
+ };
+ }
+
+ private Consumer getConsumer(long timeout) throws InterruptedException {
+ StopWatch watch = new StopWatch();
+ while (consumer == null) {
+ long rem = timeout - watch.taken();
+ if (rem <= 0) {
+ break;
+ }
+ consumer.wait(rem);
+ }
+ return consumer;
+ }
+ }
+
+ private class SyncMockComponent extends MockComponent {
+
+ @Override
+ protected Endpoint createEndpoint(String uri, String key, Map<String, Object> parameters) {
+ return new SyncMockEndpoint(this, uri, key);
+ }
+ }
+
+ private class SyncMockEndpoint extends MockEndpoint {
+ public SyncMockEndpoint(SyncMockComponent component, String uri, String key) {
+ super(uri, component);
+ }
+
+ @Override
+ public Producer createProducer() {
+ return new DefaultProducer(this) {
+ @Override
+ public void process(Exchange exchange) {
+ assertCurrentSpan(exchange);
+ }
+ };
+ }
+ }
+
+ private static void assertCurrentSpan(Exchange exchange) {
+ assertEquals(Span.current().getSpanContext().getSpanId(), ActiveSpanManager.getSpan(exchange).spanId());
+ }
+}
diff --git a/components/camel-tracing/src/main/java/org/apache/camel/tracing/ActiveSpanManager.java b/components/camel-tracing/src/main/java/org/apache/camel/tracing/ActiveSpanManager.java
index 15e3b9a7fdf..5aedf64a038 100644
--- a/components/camel-tracing/src/main/java/org/apache/camel/tracing/ActiveSpanManager.java
+++ b/components/camel-tracing/src/main/java/org/apache/camel/tracing/ActiveSpanManager.java
@@ -17,6 +17,8 @@
package org.apache.camel.tracing;
import org.apache.camel.Exchange;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
/**
@@ -27,6 +29,7 @@ public final class ActiveSpanManager {
public static final String MDC_TRACE_ID = "trace_id";
public static final String MDC_SPAN_ID = "span_id";
private static final String ACTIVE_SPAN_PROPERTY = "OpenTracing.activeSpan";
+ private static final Logger LOG = LoggerFactory.getLogger(ActiveSpanManager.class);
private ActiveSpanManager() {
}
@@ -54,8 +57,7 @@ public final class ActiveSpanManager {
*/
public static void activate(Exchange exchange, SpanAdapter span) {
exchange.setProperty(ACTIVE_SPAN_PROPERTY,
- new Holder((Holder) exchange.getProperty(ACTIVE_SPAN_PROPERTY), span));
-
+ new Holder((Holder) exchange.getProperty(ACTIVE_SPAN_PROPERTY), span, span.makeCurrent()));
if (exchange.getContext().isUseMDCLogging()) {
MDC.put(MDC_TRACE_ID, "" + span.traceId());
MDC.put(MDC_SPAN_ID, "" + span.spanId());
@@ -74,6 +76,7 @@ public final class ActiveSpanManager {
if (holder != null) {
exchange.setProperty(ACTIVE_SPAN_PROPERTY, holder.getParent());
+ holder.closeScope();
if (exchange.getContext().isUseMDCLogging()) {
Holder parent = holder.getParent();
if (parent != null) {
@@ -88,6 +91,20 @@ public final class ActiveSpanManager {
}
}
+ /**
+ * If underlying span is active, closes its scope without ending the span. This methods should be called after async
+ * execution is started on the same thread on which span was activated. ExchangeAsyncStartedEvent is used to notify
+ * about it.
+ *
+ * @param exchange The exchange
+ */
+ public static void endScope(Exchange exchange) {
+ Holder holder = (Holder) exchange.getProperty(ACTIVE_SPAN_PROPERTY);
+ if (holder != null) {
+ holder.closeScope();
+ }
+ }
+
/**
* Simple holder for the currently active span and an optional reference to the parent holder. This will be used to
* maintain a stack for spans, built up during the execution of a series of chained camel exchanges, and then
@@ -97,10 +114,12 @@ public final class ActiveSpanManager {
public static class Holder {
private Holder parent;
private SpanAdapter span;
+ private AutoCloseable scope;
- public Holder(Holder parent, SpanAdapter span) {
+ public Holder(Holder parent, SpanAdapter span, AutoCloseable scope) {
this.parent = parent;
this.span = span;
+ this.scope = scope;
}
public Holder getParent() {
@@ -110,5 +129,16 @@ public final class ActiveSpanManager {
public SpanAdapter getSpan() {
return span;
}
+
+ private void closeScope() {
+ if (scope != null) {
+ try {
+ scope.close();
+ } catch (Exception e) {
+ LOG.debug("Failed to close span scope", e);
+ }
+ this.scope = null;
+ }
+ }
}
}
diff --git a/components/camel-tracing/src/main/java/org/apache/camel/tracing/SpanAdapter.java b/components/camel-tracing/src/main/java/org/apache/camel/tracing/SpanAdapter.java
index ce52c38df65..d256f4528be 100644
--- a/components/camel-tracing/src/main/java/org/apache/camel/tracing/SpanAdapter.java
+++ b/components/camel-tracing/src/main/java/org/apache/camel/tracing/SpanAdapter.java
@@ -38,4 +38,8 @@ public interface SpanAdapter {
String traceId();
String spanId();
+
+ default AutoCloseable makeCurrent() {
+ return Tracer.NOOP_CLOSEABLE;
+ }
}
diff --git a/components/camel-tracing/src/main/java/org/apache/camel/tracing/Tracer.java b/components/camel-tracing/src/main/java/org/apache/camel/tracing/Tracer.java
index 7b6590b04d7..bf1662c06f7 100644
--- a/components/camel-tracing/src/main/java/org/apache/camel/tracing/Tracer.java
+++ b/components/camel-tracing/src/main/java/org/apache/camel/tracing/Tracer.java
@@ -52,6 +52,8 @@ import org.slf4j.LoggerFactory;
public abstract class Tracer extends ServiceSupport implements RoutePolicyFactory, StaticService, CamelContextAware {
protected static final Map<String, SpanDecorator> DECORATORS = new HashMap<>();
+ static final AutoCloseable NOOP_CLOSEABLE = () -> {
+ };
private static final Logger LOG = LoggerFactory.getLogger(Tracer.class);
static {
@@ -236,10 +238,10 @@ public abstract class Tracer extends ServiceSupport implements RoutePolicyFactor
if (event instanceof CamelEvent.ExchangeSendingEvent) {
CamelEvent.ExchangeSendingEvent ese = (CamelEvent.ExchangeSendingEvent) event;
SpanDecorator sd = getSpanDecorator(ese.getEndpoint());
- if (sd instanceof AbstractInternalSpanDecorator || !sd.newSpan()
- || isExcluded(ese.getExchange(), ese.getEndpoint())) {
+ if (exclude(sd, ese.getExchange(), ese.getEndpoint())) {
return;
}
+
SpanAdapter parent = ActiveSpanManager.getSpan(ese.getExchange());
SpanAdapter span = startSendingEventSpan(sd.getOperationName(ese.getExchange(), ese.getEndpoint()),
sd.getInitiatorSpanKind(), parent);
@@ -252,10 +254,10 @@ public abstract class Tracer extends ServiceSupport implements RoutePolicyFactor
} else if (event instanceof CamelEvent.ExchangeSentEvent) {
CamelEvent.ExchangeSentEvent ese = (CamelEvent.ExchangeSentEvent) event;
SpanDecorator sd = getSpanDecorator(ese.getEndpoint());
- if (sd instanceof AbstractInternalSpanDecorator || !sd.newSpan()
- || isExcluded(ese.getExchange(), ese.getEndpoint())) {
+ if (exclude(sd, ese.getExchange(), ese.getEndpoint())) {
return;
}
+
SpanAdapter span = ActiveSpanManager.getSpan(ese.getExchange());
if (span != null) {
if (LOG.isTraceEnabled()) {
@@ -267,12 +269,24 @@ public abstract class Tracer extends ServiceSupport implements RoutePolicyFactor
} else {
LOG.warn("Tracing: could not find managed span for exchange={}", ese.getExchange());
}
+ } else if (event instanceof CamelEvent.ExchangeAsyncProcessingStartedEvent) {
+ CamelEvent.ExchangeAsyncProcessingStartedEvent eap = (CamelEvent.ExchangeAsyncProcessingStartedEvent) event;
+
+ // no need to filter scopes here. It's ok to close a scope multiple times and
+ // implementations check if scope being disposed is current
+ // and should not do anything if scopes don't match.
+ ActiveSpanManager.endScope(eap.getExchange());
}
} catch (Exception t) {
// This exception is ignored
LOG.warn("Tracing: Failed to capture tracing data", t);
}
}
+
+ private boolean exclude(SpanDecorator sd, Exchange exchange, Endpoint endpoint) {
+ return sd instanceof AbstractInternalSpanDecorator || !sd.newSpan()
+ || isExcluded(exchange, endpoint);
+ }
}
private final class TracingRoutePolicy extends RoutePolicySupport {
diff --git a/components/camel-tracing/src/test/java/org/apache/camel/tracing/ActiveSpanManagerTest.java b/components/camel-tracing/src/test/java/org/apache/camel/tracing/ActiveSpanManagerTest.java
index 3da5f1143e8..b6669d4a3bb 100644
--- a/components/camel-tracing/src/test/java/org/apache/camel/tracing/ActiveSpanManagerTest.java
+++ b/components/camel-tracing/src/test/java/org/apache/camel/tracing/ActiveSpanManagerTest.java
@@ -22,7 +22,9 @@ import org.junit.jupiter.api.Test;
import org.slf4j.MDC;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
public class ActiveSpanManagerTest extends ExchangeTestSupport {
@@ -30,19 +32,38 @@ public class ActiveSpanManagerTest extends ExchangeTestSupport {
public void testNoSpan() {
Exchange exchange = createExchange();
assertNull(ActiveSpanManager.getSpan(exchange));
+
+ // don't throw
+ ActiveSpanManager.endScope(exchange);
}
@Test
public void testCurrentSpan() {
Exchange exchange = createExchange();
- SpanAdapter span = MockSpanAdapter.buildSpan("test");
+ MockSpanAdapter span = MockSpanAdapter.buildSpan("test");
ActiveSpanManager.activate(exchange, span);
+ assertTrue(span.isCurrent());
assertEquals(span, ActiveSpanManager.getSpan(exchange));
ActiveSpanManager.deactivate(exchange);
+ assertFalse(span.isCurrent());
assertNull(ActiveSpanManager.getSpan(exchange));
}
+ @Test
+ public void testSEndScope() {
+ Exchange exchange = createExchange();
+ MockSpanAdapter span = MockSpanAdapter.buildSpan("test");
+ ActiveSpanManager.activate(exchange, span);
+ assertTrue(span.isCurrent());
+
+ ActiveSpanManager.endScope(exchange);
+ assertFalse(span.isCurrent());
+
+ // scope has ended, but span is still on the exchange and can become parent to other spans.
+ assertEquals(span, ActiveSpanManager.getSpan(exchange));
+ }
+
@Test
public void testCreateChild() {
Exchange exchange = createExchange();
@@ -50,7 +71,6 @@ public class ActiveSpanManagerTest extends ExchangeTestSupport {
ActiveSpanManager.activate(exchange, parent);
SpanAdapter child = MockSpanAdapter.buildSpan("child");
ActiveSpanManager.activate(exchange, child);
-
assertEquals(child, ActiveSpanManager.getSpan(exchange));
ActiveSpanManager.deactivate(exchange);
diff --git a/components/camel-tracing/src/test/java/org/apache/camel/tracing/MockSpanAdapter.java b/components/camel-tracing/src/test/java/org/apache/camel/tracing/MockSpanAdapter.java
index 157900c0dba..e8a46f4dae9 100644
--- a/components/camel-tracing/src/test/java/org/apache/camel/tracing/MockSpanAdapter.java
+++ b/components/camel-tracing/src/test/java/org/apache/camel/tracing/MockSpanAdapter.java
@@ -27,6 +27,7 @@ public class MockSpanAdapter implements SpanAdapter {
private Map<String, Object> tags = new HashMap<>();
private String traceId;
private String spanId;
+ private boolean isCurrent;
static long nowMicros() {
return System.currentTimeMillis() * 1000;
@@ -102,6 +103,15 @@ public class MockSpanAdapter implements SpanAdapter {
return new ArrayList<>(this.logEntries);
}
+ @Override
+ public AutoCloseable makeCurrent() {
+ return new Scope();
+ }
+
+ public boolean isCurrent() {
+ return this.isCurrent;
+ }
+
public static final class LogEntry {
private final long timestampMicros;
private final Map<String, ?> fields;
@@ -123,4 +133,15 @@ public class MockSpanAdapter implements SpanAdapter {
public MockSpanAdapter setOperation(String operation) {
return this;
}
+
+ private final class Scope implements AutoCloseable {
+ public Scope() {
+ isCurrent = true;
+ }
+
+ @Override
+ public void close() {
+ isCurrent = false;
+ }
+ }
}
diff --git a/core/camel-api/src/main/java/org/apache/camel/spi/CamelEvent.java b/core/camel-api/src/main/java/org/apache/camel/spi/CamelEvent.java
index fb65ea9046a..2bb7ce234a9 100644
--- a/core/camel-api/src/main/java/org/apache/camel/spi/CamelEvent.java
+++ b/core/camel-api/src/main/java/org/apache/camel/spi/CamelEvent.java
@@ -52,6 +52,7 @@ public interface CamelEvent {
ExchangeRedelivery,
ExchangeSending,
ExchangeSent,
+ ExchangeAsyncProcessingStarted,
RoutesStarting,
RoutesStarted,
RoutesStopping,
@@ -449,4 +450,10 @@ public interface CamelEvent {
}
}
+ interface ExchangeAsyncProcessingStartedEvent extends ExchangeEvent {
+ @Override
+ default Type getType() {
+ return Type.ExchangeAsyncProcessingStarted;
+ }
+ }
}
diff --git a/core/camel-api/src/main/java/org/apache/camel/spi/EventFactory.java b/core/camel-api/src/main/java/org/apache/camel/spi/EventFactory.java
index bca9024b6dd..ef4313b28f3 100644
--- a/core/camel-api/src/main/java/org/apache/camel/spi/EventFactory.java
+++ b/core/camel-api/src/main/java/org/apache/camel/spi/EventFactory.java
@@ -320,6 +320,16 @@ public interface EventFactory {
*/
CamelEvent createExchangeSendingEvent(Exchange exchange, Endpoint endpoint);
+ /**
+ * Creates an {@link CamelEvent} when an {@link org.apache.camel.Exchange} asynchronous processing has been started.
+ * This is guaranteed to run on the same thread on which {@code RoutePolicySupport.onExchangeBegin} was called
+ * and/or {@code ExchangeSendingEvent} was fired.
+ *
+ * @param exchange the exchange
+ * @return the created event
+ */
+ CamelEvent createCamelExchangeAsyncProcessingStartedEvent(Exchange exchange);
+
/**
* Creates an {@link CamelEvent} when an {@link org.apache.camel.Exchange} has completely been sent to the endpoint
* (eg after).
@@ -398,5 +408,4 @@ public interface EventFactory {
* @return the created event
*/
CamelEvent createCamelContextResumeFailureEvent(CamelContext context, Throwable cause);
-
}
diff --git a/core/camel-api/src/main/java/org/apache/camel/spi/EventNotifier.java b/core/camel-api/src/main/java/org/apache/camel/spi/EventNotifier.java
index 4312d011aff..9d1f8a688ec 100644
--- a/core/camel-api/src/main/java/org/apache/camel/spi/EventNotifier.java
+++ b/core/camel-api/src/main/java/org/apache/camel/spi/EventNotifier.java
@@ -96,4 +96,10 @@ public interface EventNotifier {
void setIgnoreStepEvents(boolean ignoreStepEvents);
+ default void setIgnoreExchangeAsyncProcessingStartedEvents(boolean ignoreExchangeAsyncProcessingStartedEvents) {
+ }
+
+ default boolean isIgnoreExchangeAsyncProcessingStartedEvents() {
+ return false;
+ }
}
diff --git a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/CamelInternalProcessor.java b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/CamelInternalProcessor.java
index 20068a4d892..d16d1fab5d4 100644
--- a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/CamelInternalProcessor.java
+++ b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/CamelInternalProcessor.java
@@ -57,6 +57,7 @@ import org.apache.camel.spi.Transformer;
import org.apache.camel.spi.UnitOfWork;
import org.apache.camel.spi.UnitOfWorkFactory;
import org.apache.camel.support.CamelContextHelper;
+import org.apache.camel.support.EventHelper;
import org.apache.camel.support.ExchangeHelper;
import org.apache.camel.support.LoggerHelper;
import org.apache.camel.support.MessageHelper;
@@ -389,6 +390,10 @@ public class CamelInternalProcessor extends DelegateAsyncProcessor implements In
LOG.trace("Processing exchange for exchangeId: {} -> {}", exchange.getExchangeId(), exchange);
}
boolean sync = processor.process(exchange, async);
+ if (!sync) {
+ EventHelper.notifyExchangeAsyncProcessingStartedEvent(exchange.getContext(), exchange);
+ }
+
// ----------------------------------------------------------
// CAMEL END USER - DEBUG ME HERE +++ END +++
// ----------------------------------------------------------
diff --git a/core/camel-base/src/main/java/org/apache/camel/impl/event/DefaultEventFactory.java b/core/camel-base/src/main/java/org/apache/camel/impl/event/DefaultEventFactory.java
index 98cb27af552..5ce424cce97 100644
--- a/core/camel-base/src/main/java/org/apache/camel/impl/event/DefaultEventFactory.java
+++ b/core/camel-base/src/main/java/org/apache/camel/impl/event/DefaultEventFactory.java
@@ -417,4 +417,13 @@ public class DefaultEventFactory implements EventFactory {
}
return answer;
}
+
+ @Override
+ public CamelEvent createCamelExchangeAsyncProcessingStartedEvent(Exchange exchange) {
+ CamelEvent answer = new ExchangeAsyncProcessingStartedEvent(exchange);
+ if (timestampEnabled) {
+ answer.setTimestamp(System.currentTimeMillis());
+ }
+ return answer;
+ }
}
diff --git a/core/camel-base/src/main/java/org/apache/camel/impl/event/ExchangeAsyncProcessingStartedEvent.java b/core/camel-base/src/main/java/org/apache/camel/impl/event/ExchangeAsyncProcessingStartedEvent.java
new file mode 100644
index 00000000000..2f38d412aa5
--- /dev/null
+++ b/core/camel-base/src/main/java/org/apache/camel/impl/event/ExchangeAsyncProcessingStartedEvent.java
@@ -0,0 +1,42 @@
+/*
+ * 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.impl.event;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.Route;
+import org.apache.camel.spi.CamelEvent;
+import org.apache.camel.support.RoutePolicySupport;
+
+/**
+ * Notifies that async processing has started. It's guaranteed to run on the same thread on which
+ * {@link RoutePolicySupport#onExchangeBegin(Route, Exchange)} was called and/or {@link ExchangeSendingEvent} was fired.
+ *
+ * @see ExchangeAsyncProcessingStartedEvent
+ */
+public class ExchangeAsyncProcessingStartedEvent extends AbstractExchangeEvent
+ implements CamelEvent.ExchangeAsyncProcessingStartedEvent {
+ private static final long serialVersionUID = -19248832613958122L;
+
+ public ExchangeAsyncProcessingStartedEvent(Exchange source) {
+ super(source);
+ }
+
+ @Override
+ public String toString() {
+ return getExchange().getExchangeId();
+ }
+}
diff --git a/core/camel-core-processor/src/main/java/org/apache/camel/processor/SendProcessor.java b/core/camel-core-processor/src/main/java/org/apache/camel/processor/SendProcessor.java
index 9e8fec41217..f4138b74b30 100644
--- a/core/camel-core-processor/src/main/java/org/apache/camel/processor/SendProcessor.java
+++ b/core/camel-core-processor/src/main/java/org/apache/camel/processor/SendProcessor.java
@@ -169,7 +169,11 @@ public class SendProcessor extends AsyncProcessorSupport implements Traceable, E
}
try {
LOG.debug(">>>> {} {}", destination, exchange);
- return producer.process(exchange, ac);
+ boolean sync = producer.process(exchange, ac);
+ if (!sync) {
+ EventHelper.notifyExchangeAsyncProcessingStartedEvent(exchange.getContext(), exchange);
+ }
+ return sync;
} catch (Throwable throwable) {
exchange.setException(throwable);
callback.done(true);
diff --git a/core/camel-core/src/test/java/org/apache/camel/impl/event/EventNotifierEventsTest.java b/core/camel-core/src/test/java/org/apache/camel/impl/event/EventNotifierEventsTest.java
index 98bd0f054fb..65feefa2917 100644
--- a/core/camel-core/src/test/java/org/apache/camel/impl/event/EventNotifierEventsTest.java
+++ b/core/camel-core/src/test/java/org/apache/camel/impl/event/EventNotifierEventsTest.java
@@ -79,7 +79,7 @@ public class EventNotifierEventsTest {
mock.assertIsSatisfied();
- assertEquals(20, events.size());
+ assertEquals(23, events.size());
assertIsInstanceOf(CamelEvent.CamelContextInitializingEvent.class, events.get(0));
assertIsInstanceOf(CamelEvent.CamelContextInitializedEvent.class, events.get(1));
assertIsInstanceOf(CamelContextStartingEvent.class, events.get(2));
@@ -93,28 +93,32 @@ public class EventNotifierEventsTest {
assertIsInstanceOf(CamelContextRoutesStartedEvent.class, events.get(10));
assertIsInstanceOf(CamelContextStartedEvent.class, events.get(11));
assertIsInstanceOf(ExchangeCreatedEvent.class, events.get(13));
- assertIsInstanceOf(ExchangeSendingEvent.class, events.get(14));
- assertIsInstanceOf(ExchangeSentEvent.class, events.get(15));
- assertIsInstanceOf(ExchangeSendingEvent.class, events.get(16));
- assertIsInstanceOf(ExchangeSentEvent.class, events.get(17));
- assertIsInstanceOf(ExchangeCompletedEvent.class, events.get(18));
+ assertIsInstanceOf(ExchangeAsyncProcessingStartedEvent.class, events.get(14));
+ assertIsInstanceOf(ExchangeSendingEvent.class, events.get(15));
+
+ assertIsInstanceOf(ExchangeSentEvent.class, events.get(16));
+ assertIsInstanceOf(ExchangeAsyncProcessingStartedEvent.class, events.get(17));
+ assertIsInstanceOf(ExchangeSendingEvent.class, events.get(18));
+ assertIsInstanceOf(ExchangeSentEvent.class, events.get(19));
+ assertIsInstanceOf(ExchangeCompletedEvent.class, events.get(20));
+ assertIsInstanceOf(ExchangeAsyncProcessingStartedEvent.class, events.get(21));
// this is the sent using the produce template to start the test
- assertIsInstanceOf(ExchangeSentEvent.class, events.get(17));
+ assertIsInstanceOf(ExchangeSentEvent.class, events.get(22));
context.stop();
- assertEquals(30, events.size());
- assertIsInstanceOf(CamelContextStoppingEvent.class, events.get(20));
- assertIsInstanceOf(CamelContextRoutesStoppingEvent.class, events.get(21));
- assertIsInstanceOf(RouteStoppingEvent.class, events.get(22));
- assertIsInstanceOf(RouteStoppedEvent.class, events.get(23));
- assertIsInstanceOf(RouteRemovedEvent.class, events.get(24));
+ assertEquals(33, events.size());
+ assertIsInstanceOf(CamelContextStoppingEvent.class, events.get(23));
+ assertIsInstanceOf(CamelContextRoutesStoppingEvent.class, events.get(24));
assertIsInstanceOf(RouteStoppingEvent.class, events.get(25));
assertIsInstanceOf(RouteStoppedEvent.class, events.get(26));
assertIsInstanceOf(RouteRemovedEvent.class, events.get(27));
- assertIsInstanceOf(CamelContextRoutesStoppedEvent.class, events.get(28));
- assertIsInstanceOf(CamelContextStoppedEvent.class, events.get(29));
+ assertIsInstanceOf(RouteStoppingEvent.class, events.get(28));
+ assertIsInstanceOf(RouteStoppedEvent.class, events.get(29));
+ assertIsInstanceOf(RouteRemovedEvent.class, events.get(30));
+ assertIsInstanceOf(CamelContextRoutesStoppedEvent.class, events.get(31));
+ assertIsInstanceOf(CamelContextStoppedEvent.class, events.get(32));
}
@Test
@@ -130,7 +134,7 @@ public class EventNotifierEventsTest {
assertIsInstanceOf(IllegalArgumentException.class, e.getCause());
}
- assertEquals(16, events.size());
+ assertEquals(18, events.size());
assertIsInstanceOf(CamelEvent.CamelContextInitializingEvent.class, events.get(0));
assertIsInstanceOf(CamelEvent.CamelContextInitializedEvent.class, events.get(1));
assertIsInstanceOf(CamelContextStartingEvent.class, events.get(2));
@@ -145,23 +149,27 @@ public class EventNotifierEventsTest {
assertIsInstanceOf(CamelContextStartedEvent.class, events.get(11));
assertIsInstanceOf(ExchangeSendingEvent.class, events.get(12));
assertIsInstanceOf(ExchangeCreatedEvent.class, events.get(13));
- assertIsInstanceOf(ExchangeFailedEvent.class, events.get(14));
+
+ assertIsInstanceOf(ExchangeAsyncProcessingStartedEvent.class, events.get(14));
+ assertIsInstanceOf(ExchangeFailedEvent.class, events.get(15));
+ assertIsInstanceOf(ExchangeAsyncProcessingStartedEvent.class, events.get(16));
+
// this is the sent using the produce template to start the test
- assertIsInstanceOf(ExchangeSentEvent.class, events.get(15));
+ assertIsInstanceOf(ExchangeSentEvent.class, events.get(17));
context.stop();
- assertEquals(26, events.size());
- assertIsInstanceOf(CamelContextStoppingEvent.class, events.get(16));
- assertIsInstanceOf(CamelContextRoutesStoppingEvent.class, events.get(17));
- assertIsInstanceOf(RouteStoppingEvent.class, events.get(18));
- assertIsInstanceOf(RouteStoppedEvent.class, events.get(19));
- assertIsInstanceOf(RouteRemovedEvent.class, events.get(20));
- assertIsInstanceOf(RouteStoppingEvent.class, events.get(21));
- assertIsInstanceOf(RouteStoppedEvent.class, events.get(22));
- assertIsInstanceOf(RouteRemovedEvent.class, events.get(23));
- assertIsInstanceOf(CamelContextRoutesStoppedEvent.class, events.get(24));
- assertIsInstanceOf(CamelContextStoppedEvent.class, events.get(25));
+ assertEquals(28, events.size());
+ assertIsInstanceOf(CamelContextStoppingEvent.class, events.get(18));
+ assertIsInstanceOf(CamelContextRoutesStoppingEvent.class, events.get(19));
+ assertIsInstanceOf(RouteStoppingEvent.class, events.get(20));
+ assertIsInstanceOf(RouteStoppedEvent.class, events.get(21));
+ assertIsInstanceOf(RouteRemovedEvent.class, events.get(22));
+ assertIsInstanceOf(RouteStoppingEvent.class, events.get(23));
+ assertIsInstanceOf(RouteStoppedEvent.class, events.get(24));
+ assertIsInstanceOf(RouteRemovedEvent.class, events.get(25));
+ assertIsInstanceOf(CamelContextRoutesStoppedEvent.class, events.get(26));
+ assertIsInstanceOf(CamelContextStoppedEvent.class, events.get(27));
}
@Test
diff --git a/core/camel-core/src/test/java/org/apache/camel/impl/event/EventNotifierExchangeSentTest.java b/core/camel-core/src/test/java/org/apache/camel/impl/event/EventNotifierExchangeSentTest.java
index 68970c51710..3091692e740 100644
--- a/core/camel-core/src/test/java/org/apache/camel/impl/event/EventNotifierExchangeSentTest.java
+++ b/core/camel-core/src/test/java/org/apache/camel/impl/event/EventNotifierExchangeSentTest.java
@@ -60,6 +60,7 @@ public class EventNotifierExchangeSentTest extends ContextTestSupport {
setIgnoreExchangeCompletedEvent(true);
setIgnoreExchangeFailedEvents(true);
setIgnoreExchangeRedeliveryEvents(true);
+ setIgnoreExchangeAsyncProcessingStartedEvents(true);
}
});
return context;
@@ -74,6 +75,7 @@ public class EventNotifierExchangeSentTest extends ContextTestSupport {
assertMockEndpointsSatisfied();
assertEquals(8, events.size());
+
ExchangeSendingEvent e0 = assertIsInstanceOf(ExchangeSendingEvent.class, events.get(0));
ExchangeSendingEvent e1 = assertIsInstanceOf(ExchangeSendingEvent.class, events.get(1));
ExchangeSentEvent e2 = assertIsInstanceOf(ExchangeSentEvent.class, events.get(2));
diff --git a/core/camel-core/src/test/java/org/apache/camel/impl/event/EventNotifierFailureHandledEventsTest.java b/core/camel-core/src/test/java/org/apache/camel/impl/event/EventNotifierFailureHandledEventsTest.java
index 979ce7d5337..b5a58700d45 100644
--- a/core/camel-core/src/test/java/org/apache/camel/impl/event/EventNotifierFailureHandledEventsTest.java
+++ b/core/camel-core/src/test/java/org/apache/camel/impl/event/EventNotifierFailureHandledEventsTest.java
@@ -56,6 +56,11 @@ public class EventNotifierFailureHandledEventsTest extends ContextTestSupport {
public void notify(CamelEvent event) throws Exception {
events.add(event);
}
+
+ @Override
+ protected void doBuild() throws Exception {
+ setIgnoreExchangeAsyncProcessingStartedEvents(true);
+ }
});
return context;
}
diff --git a/core/camel-core/src/test/java/org/apache/camel/impl/event/EventNotifierRedeliveryEventsTest.java b/core/camel-core/src/test/java/org/apache/camel/impl/event/EventNotifierRedeliveryEventsTest.java
index f331281d60b..877664bdfd2 100644
--- a/core/camel-core/src/test/java/org/apache/camel/impl/event/EventNotifierRedeliveryEventsTest.java
+++ b/core/camel-core/src/test/java/org/apache/camel/impl/event/EventNotifierRedeliveryEventsTest.java
@@ -60,6 +60,7 @@ public class EventNotifierRedeliveryEventsTest extends ContextTestSupport {
setIgnoreCamelContextEvents(true);
setIgnoreRouteEvents(true);
setIgnoreServiceEvents(true);
+ setIgnoreExchangeAsyncProcessingStartedEvents(true);
}
});
return context;
diff --git a/core/camel-core/src/test/java/org/apache/camel/impl/event/MultipleEventNotifierEventsTest.java b/core/camel-core/src/test/java/org/apache/camel/impl/event/MultipleEventNotifierEventsTest.java
index b53eaf830d1..967793635eb 100644
--- a/core/camel-core/src/test/java/org/apache/camel/impl/event/MultipleEventNotifierEventsTest.java
+++ b/core/camel-core/src/test/java/org/apache/camel/impl/event/MultipleEventNotifierEventsTest.java
@@ -80,7 +80,7 @@ public class MultipleEventNotifierEventsTest extends ContextTestSupport {
assertMockEndpointsSatisfied();
- assertEquals(20, events.size());
+ assertEquals(23, events.size());
assertIsInstanceOf(CamelEvent.CamelContextInitializingEvent.class, events.get(0));
assertIsInstanceOf(CamelEvent.CamelContextInitializedEvent.class, events.get(1));
assertIsInstanceOf(CamelContextStartingEvent.class, events.get(2));
@@ -95,38 +95,44 @@ public class MultipleEventNotifierEventsTest extends ContextTestSupport {
assertIsInstanceOf(CamelContextStartedEvent.class, events.get(11));
assertIsInstanceOf(ExchangeSendingEvent.class, events.get(12));
assertIsInstanceOf(ExchangeCreatedEvent.class, events.get(13));
- assertIsInstanceOf(ExchangeSendingEvent.class, events.get(14));
- assertIsInstanceOf(ExchangeSentEvent.class, events.get(15));
- assertIsInstanceOf(ExchangeSendingEvent.class, events.get(16));
- assertIsInstanceOf(ExchangeSentEvent.class, events.get(17));
- assertIsInstanceOf(ExchangeCompletedEvent.class, events.get(18));
+ assertIsInstanceOf(ExchangeAsyncProcessingStartedEvent.class, events.get(14));
+ assertIsInstanceOf(ExchangeSendingEvent.class, events.get(15));
+ assertIsInstanceOf(ExchangeSentEvent.class, events.get(16));
+ assertIsInstanceOf(ExchangeAsyncProcessingStartedEvent.class, events.get(17));
+ assertIsInstanceOf(ExchangeSendingEvent.class, events.get(18));
assertIsInstanceOf(ExchangeSentEvent.class, events.get(19));
+ assertIsInstanceOf(ExchangeCompletedEvent.class, events.get(20));
+ assertIsInstanceOf(ExchangeAsyncProcessingStartedEvent.class, events.get(21));
+ assertIsInstanceOf(ExchangeSentEvent.class, events.get(22));
- assertEquals(8, events2.size());
+ assertEquals(11, events2.size());
assertIsInstanceOf(ExchangeSendingEvent.class, events2.get(0));
assertIsInstanceOf(ExchangeCreatedEvent.class, events2.get(1));
- assertIsInstanceOf(ExchangeSendingEvent.class, events2.get(2));
- assertIsInstanceOf(ExchangeSentEvent.class, events2.get(3));
- assertIsInstanceOf(ExchangeSendingEvent.class, events2.get(4));
- assertIsInstanceOf(ExchangeSentEvent.class, events2.get(5));
- assertIsInstanceOf(ExchangeCompletedEvent.class, events2.get(6));
+ assertIsInstanceOf(ExchangeAsyncProcessingStartedEvent.class, events2.get(2));
+ assertIsInstanceOf(ExchangeSendingEvent.class, events2.get(3));
+ assertIsInstanceOf(ExchangeSentEvent.class, events2.get(4));
+ assertIsInstanceOf(ExchangeAsyncProcessingStartedEvent.class, events2.get(5));
+ assertIsInstanceOf(ExchangeSendingEvent.class, events2.get(6));
assertIsInstanceOf(ExchangeSentEvent.class, events2.get(7));
+ assertIsInstanceOf(ExchangeCompletedEvent.class, events2.get(8));
+ assertIsInstanceOf(ExchangeAsyncProcessingStartedEvent.class, events2.get(9));
+ assertIsInstanceOf(ExchangeSentEvent.class, events2.get(10));
context.stop();
- assertEquals(30, events.size());
- assertIsInstanceOf(CamelContextStoppingEvent.class, events.get(20));
- assertIsInstanceOf(CamelContextRoutesStoppingEvent.class, events.get(21));
- assertIsInstanceOf(RouteStoppingEvent.class, events.get(22));
- assertIsInstanceOf(RouteStoppedEvent.class, events.get(23));
- assertIsInstanceOf(RouteRemovedEvent.class, events.get(24));
+ assertEquals(33, events.size());
+ assertIsInstanceOf(CamelContextStoppingEvent.class, events.get(23));
+ assertIsInstanceOf(CamelContextRoutesStoppingEvent.class, events.get(24));
assertIsInstanceOf(RouteStoppingEvent.class, events.get(25));
assertIsInstanceOf(RouteStoppedEvent.class, events.get(26));
assertIsInstanceOf(RouteRemovedEvent.class, events.get(27));
- assertIsInstanceOf(CamelContextRoutesStoppedEvent.class, events.get(28));
- assertIsInstanceOf(CamelContextStoppedEvent.class, events.get(29));
+ assertIsInstanceOf(RouteStoppingEvent.class, events.get(28));
+ assertIsInstanceOf(RouteStoppedEvent.class, events.get(29));
+ assertIsInstanceOf(RouteRemovedEvent.class, events.get(30));
+ assertIsInstanceOf(CamelContextRoutesStoppedEvent.class, events.get(31));
+ assertIsInstanceOf(CamelContextStoppedEvent.class, events.get(32));
- assertEquals(8, events2.size());
+ assertEquals(11, events2.size());
}
@Test
@@ -139,7 +145,7 @@ public class MultipleEventNotifierEventsTest extends ContextTestSupport {
assertIsInstanceOf(IllegalArgumentException.class, e.getCause());
}
- assertEquals(16, events.size());
+ assertEquals(18, events.size());
assertIsInstanceOf(CamelEvent.CamelContextInitializingEvent.class, events.get(0));
assertIsInstanceOf(CamelEvent.CamelContextInitializedEvent.class, events.get(1));
assertIsInstanceOf(CamelContextStartingEvent.class, events.get(2));
@@ -154,30 +160,34 @@ public class MultipleEventNotifierEventsTest extends ContextTestSupport {
assertIsInstanceOf(CamelContextStartedEvent.class, events.get(11));
assertIsInstanceOf(ExchangeSendingEvent.class, events.get(12));
assertIsInstanceOf(ExchangeCreatedEvent.class, events.get(13));
- assertIsInstanceOf(ExchangeFailedEvent.class, events.get(14));
- assertIsInstanceOf(ExchangeSentEvent.class, events.get(15));
+ assertIsInstanceOf(ExchangeAsyncProcessingStartedEvent.class, events.get(14));
+ assertIsInstanceOf(ExchangeFailedEvent.class, events.get(15));
+ assertIsInstanceOf(ExchangeAsyncProcessingStartedEvent.class, events.get(16));
+ assertIsInstanceOf(ExchangeSentEvent.class, events.get(17));
- assertEquals(4, events2.size());
+ assertEquals(6, events2.size());
context.stop();
assertIsInstanceOf(ExchangeSendingEvent.class, events2.get(0));
assertIsInstanceOf(ExchangeCreatedEvent.class, events2.get(1));
- assertIsInstanceOf(ExchangeFailedEvent.class, events2.get(2));
- assertIsInstanceOf(ExchangeSentEvent.class, events2.get(3));
-
- assertEquals(26, events.size());
- assertIsInstanceOf(CamelContextStoppingEvent.class, events.get(16));
- assertIsInstanceOf(CamelContextRoutesStoppingEvent.class, events.get(17));
- assertIsInstanceOf(RouteStoppingEvent.class, events.get(18));
- assertIsInstanceOf(RouteStoppedEvent.class, events.get(19));
- assertIsInstanceOf(RouteRemovedEvent.class, events.get(20));
- assertIsInstanceOf(RouteStoppingEvent.class, events.get(21));
- assertIsInstanceOf(RouteStoppedEvent.class, events.get(22));
- assertIsInstanceOf(RouteRemovedEvent.class, events.get(23));
- assertIsInstanceOf(CamelContextRoutesStoppedEvent.class, events.get(24));
- assertIsInstanceOf(CamelContextStoppedEvent.class, events.get(25));
-
- assertEquals(4, events2.size());
+ assertIsInstanceOf(ExchangeAsyncProcessingStartedEvent.class, events2.get(2));
+ assertIsInstanceOf(ExchangeFailedEvent.class, events2.get(3));
+ assertIsInstanceOf(ExchangeAsyncProcessingStartedEvent.class, events2.get(4));
+ assertIsInstanceOf(ExchangeSentEvent.class, events2.get(5));
+
+ assertEquals(28, events.size());
+ assertIsInstanceOf(CamelContextStoppingEvent.class, events.get(18));
+ assertIsInstanceOf(CamelContextRoutesStoppingEvent.class, events.get(19));
+ assertIsInstanceOf(RouteStoppingEvent.class, events.get(20));
+ assertIsInstanceOf(RouteStoppedEvent.class, events.get(21));
+ assertIsInstanceOf(RouteRemovedEvent.class, events.get(22));
+ assertIsInstanceOf(RouteStoppingEvent.class, events.get(23));
+ assertIsInstanceOf(RouteStoppedEvent.class, events.get(24));
+ assertIsInstanceOf(RouteRemovedEvent.class, events.get(25));
+ assertIsInstanceOf(CamelContextRoutesStoppedEvent.class, events.get(26));
+ assertIsInstanceOf(CamelContextStoppedEvent.class, events.get(27));
+
+ assertEquals(6, events2.size());
}
@Override
diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/interceptor/DebugTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/interceptor/DebugTest.java
index 7df10cab404..3aa7c400fdf 100644
--- a/core/camel-core/src/test/java/org/apache/camel/processor/interceptor/DebugTest.java
+++ b/core/camel-core/src/test/java/org/apache/camel/processor/interceptor/DebugTest.java
@@ -161,7 +161,7 @@ public class DebugTest extends ContextTestSupport {
assertMockEndpointsSatisfied();
- assertEquals(10, logs.size());
+ assertEquals(13, logs.size());
// remove the breakpoint
context.getDebugger().removeBreakpoint(breakpoint);
diff --git a/core/camel-management/src/test/java/org/apache/camel/management/AddEventNotifierTest.java b/core/camel-management/src/test/java/org/apache/camel/management/AddEventNotifierTest.java
index 68fd88d66d2..83bcb502641 100644
--- a/core/camel-management/src/test/java/org/apache/camel/management/AddEventNotifierTest.java
+++ b/core/camel-management/src/test/java/org/apache/camel/management/AddEventNotifierTest.java
@@ -67,7 +67,7 @@ public class AddEventNotifierTest extends ContextTestSupport {
template.sendBody("direct:start", "Bye World");
assertMockEndpointsSatisfied();
- assertEquals(8, events.size());
+ assertEquals(11, events.size());
// remove and we should not get new events
context.getManagementStrategy().removeEventNotifier(notifier);
@@ -77,7 +77,7 @@ public class AddEventNotifierTest extends ContextTestSupport {
template.sendBody("direct:start", "Hi World");
assertMockEndpointsSatisfied();
- assertEquals(8, events.size());
+ assertEquals(11, events.size());
}
@Override
diff --git a/core/camel-management/src/test/java/org/apache/camel/management/JmxNotificationEventNotifierTest.java b/core/camel-management/src/test/java/org/apache/camel/management/JmxNotificationEventNotifierTest.java
index 5d46399d52a..44a47ed69b0 100644
--- a/core/camel-management/src/test/java/org/apache/camel/management/JmxNotificationEventNotifierTest.java
+++ b/core/camel-management/src/test/java/org/apache/camel/management/JmxNotificationEventNotifierTest.java
@@ -77,7 +77,7 @@ public class JmxNotificationEventNotifierTest extends ManagementTestSupport {
assertMockEndpointsSatisfied();
- assertEquals(8, listener.getEventCounter(), "Get a wrong number of events");
+ assertEquals(11, listener.getEventCounter(), "Get a wrong number of events");
context.stop();
}
@@ -104,7 +104,7 @@ public class JmxNotificationEventNotifierTest extends ManagementTestSupport {
assertIsInstanceOf(IllegalArgumentException.class, e.getCause());
}
- assertEquals(4, listener.getEventCounter(), "Get a wrong number of events");
+ assertEquals(6, listener.getEventCounter(), "Get a wrong number of events");
context.stop();
}
diff --git a/core/camel-management/src/test/java/org/apache/camel/management/PublishEventNotifierTest.java b/core/camel-management/src/test/java/org/apache/camel/management/PublishEventNotifierTest.java
index 3decdb7043d..c00e3683f51 100644
--- a/core/camel-management/src/test/java/org/apache/camel/management/PublishEventNotifierTest.java
+++ b/core/camel-management/src/test/java/org/apache/camel/management/PublishEventNotifierTest.java
@@ -50,7 +50,7 @@ public class PublishEventNotifierTest extends ContextTestSupport {
@Test
public void testExchangeDone() throws Exception {
getMockEndpoint("mock:result").expectedMessageCount(1);
- getMockEndpoint("mock:event").expectedMessageCount(9);
+ getMockEndpoint("mock:event").expectedMessageCount(12);
template.sendBody("direct:start", "Hello World");
@@ -59,7 +59,7 @@ public class PublishEventNotifierTest extends ContextTestSupport {
@Test
public void testExchangeFailed() throws Exception {
- getMockEndpoint("mock:event").expectedMessageCount(5);
+ getMockEndpoint("mock:event").expectedMessageCount(7);
try {
template.sendBody("direct:fail", "Hello World");
diff --git a/core/camel-management/src/test/java/org/apache/camel/management/RemoveEventNotifierTest.java b/core/camel-management/src/test/java/org/apache/camel/management/RemoveEventNotifierTest.java
index 471626215a4..8436150bae1 100644
--- a/core/camel-management/src/test/java/org/apache/camel/management/RemoveEventNotifierTest.java
+++ b/core/camel-management/src/test/java/org/apache/camel/management/RemoveEventNotifierTest.java
@@ -66,7 +66,7 @@ public class RemoveEventNotifierTest extends ContextTestSupport {
template.sendBody("direct:start", "Hello World");
assertMockEndpointsSatisfied();
- assertEquals(17, events.size());
+ assertEquals(20, events.size());
// remove and we should not get new events
context.getManagementStrategy().removeEventNotifier(notifier);
@@ -76,7 +76,7 @@ public class RemoveEventNotifierTest extends ContextTestSupport {
template.sendBody("direct:start", "Bye World");
assertMockEndpointsSatisfied();
- assertEquals(17, events.size());
+ assertEquals(20, events.size());
}
@Override
diff --git a/core/camel-support/src/main/java/org/apache/camel/support/EventHelper.java b/core/camel-support/src/main/java/org/apache/camel/support/EventHelper.java
index 07b62e54948..3d88c9826ba 100644
--- a/core/camel-support/src/main/java/org/apache/camel/support/EventHelper.java
+++ b/core/camel-support/src/main/java/org/apache/camel/support/EventHelper.java
@@ -1458,6 +1458,52 @@ public final class EventHelper {
return answer;
}
+ public static boolean notifyExchangeAsyncProcessingStartedEvent(CamelContext context, Exchange exchange) {
+ ManagementStrategy management = context.getManagementStrategy();
+ if (management == null) {
+ return false;
+ }
+
+ EventFactory factory = management.getEventFactory();
+ if (factory == null) {
+ return false;
+ }
+
+ List<EventNotifier> notifiers = management.getStartedEventNotifiers();
+ if (notifiers == null || notifiers.isEmpty()) {
+ return false;
+ }
+
+ if (((ExtendedExchange) exchange).isNotifyEvent()) {
+ // do not generate events for an notify event
+ return false;
+ }
+
+ boolean answer = false;
+ CamelEvent event = null;
+ // optimise for loop using index access to avoid creating iterator object
+ for (int i = 0; i < notifiers.size(); i++) {
+ EventNotifier notifier = notifiers.get(i);
+ if (notifier.isDisabled()) {
+ continue;
+ }
+ if (notifier.isIgnoreExchangeEvents() || notifier.isIgnoreExchangeAsyncProcessingStartedEvents()) {
+ continue;
+ }
+
+ if (event == null) {
+ // only create event once
+ event = factory.createCamelExchangeAsyncProcessingStartedEvent(exchange);
+ if (event == null) {
+ // factory could not create event so exit
+ return false;
+ }
+ }
+ answer |= doNotifyEvent(notifier, event);
+ }
+ return answer;
+ }
+
private static boolean doNotifyEvent(EventNotifier notifier, CamelEvent event) {
if (!notifier.isEnabled(event)) {
LOG.trace("Notifier: {} is not enabled for the event: {}", notifier, event);
@@ -1472,5 +1518,4 @@ public final class EventHelper {
return true;
}
-
}
diff --git a/core/camel-support/src/main/java/org/apache/camel/support/SimpleEventNotifierSupport.java b/core/camel-support/src/main/java/org/apache/camel/support/SimpleEventNotifierSupport.java
index 9996bcc0c07..16a34d9d091 100644
--- a/core/camel-support/src/main/java/org/apache/camel/support/SimpleEventNotifierSupport.java
+++ b/core/camel-support/src/main/java/org/apache/camel/support/SimpleEventNotifierSupport.java
@@ -38,6 +38,7 @@ public abstract class SimpleEventNotifierSupport extends ServiceSupport implemen
boolean ignoreExchangeEvents;
boolean ignoreExchangeCreatedEvent;
boolean ignoreExchangeCompletedEvent;
+ boolean ignoreExchangeAsyncProcessingStartedEvents;
boolean ignoreExchangeFailedEvents;
boolean ignoreExchangeRedeliveryEvents;
boolean ignoreExchangeSendingEvents;
@@ -189,6 +190,16 @@ public abstract class SimpleEventNotifierSupport extends ServiceSupport implemen
this.ignoreStepEvents = ignoreStepEvents;
}
+ @Override
+ public boolean isIgnoreExchangeAsyncProcessingStartedEvents() {
+ return ignoreExchangeAsyncProcessingStartedEvents;
+ }
+
+ @Override
+ public void setIgnoreExchangeAsyncProcessingStartedEvents(boolean ignoreExchangeAsyncProcessingStartedEvents) {
+ this.ignoreExchangeAsyncProcessingStartedEvents = ignoreExchangeAsyncProcessingStartedEvents;
+ }
+
@Override
protected void doStart() throws Exception {
// noop