fix(components): enable any async EIP to reuse main MDC context

catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/others/mdc.json

@@ -2,7 +2,7 @@
   "other": {
     "kind": "other",
     "name": "mdc",
-    "title": "Mdc",
+    "title": "MDC Logging",
     "description": "Logging MDC (Mapped Diagnostic Context) Service",
     "deprecated": false,
     "firstVersion": "4.15.0",

components/camel-mdc/pom.xml

@@ -31,6 +31,7 @@
     <firstVersion>4.15.0</firstVersion>
     <label>logging</label>
     <supportLevel>Preview</supportLevel>
+    <title>MDC Logging</title>
   </properties>
 
   <artifactId>camel-mdc</artifactId>

components/camel-mdc/src/generated/resources/mdc.json

@@ -2,7 +2,7 @@
   "other": {
     "kind": "other",
     "name": "mdc",
-    "title": "Mdc",
+    "title": "MDC Logging",
     "description": "Logging MDC (Mapped Diagnostic Context) Service",
     "deprecated": false,
     "firstVersion": "4.15.0",

components/camel-mdc/src/main/docs/mdc.adoc

@@ -1,5 +1,5 @@
-= Mdc Component
-:doctitle: Mdc
+= MDC Logging Component
+:doctitle: MDC Logging
 :shortname: mdc
 :artifactid: camel-mdc
 :description: Logging MDC (Mapped Diagnostic Context) Service
@@ -19,26 +19,32 @@ When you enable the MDC Service provided in this component, you will instruct yo
 
 If you want to use the feature, you need to include the `camel-mdc` dependency in your `pom.xml` and configure it with the parameters in the `application.properties` configuration file (at least set `camel.mdc.enabled=true`).
 
+NOTE: don't set legacy MDC via `context.setUseMDCLogging(true)` on `CamelContext` at the same time. Instead, only use `camel-mdc`.
+
 The goal of this component is to avoid to work on low level API in Java. In older MDC implementations you had to hack into the code to include MDC such as:
 
-```java
+[source,java]
+----
     org.slf4j.MDC.put("myCustomMDCKey", "myCustomKeyValue");
-```
+----
 
 And later you had to make sure to provide MDC context propagation in async components (eg, `wiretap`) in order to make sure to have such context available in the new executing async thread. With this new service, the only thing to do is to add the value as a Camel Exchange header (or a property), for example:
 
-```java
+[source,java]
+----
     .setHeader("myCustomMDCKey", simple("myCustomKeyValue"))
-```
+----
+
 include the MDC service and additionally instruct the Camel application to treat that header as a MDC trace (via `camel.mdc.customHeaders=myCustomMDCKey` or `*` to include all headers). You won't need any longer to worry about context propagation as the propagation will be done via Camel Exchange instead.
 
 NOTE: you won't also need any longer to hack the code using Java DSL, as you can put headers in any Camel DSL.
 
 Depending on the logging technology used, you can now include the MDC parameters you want to trace in your logging configuration. For example, in `log4j2` configuration you can include them as shown below:
 
-```
+[source,text]
+----
 ... [%X{camel.contextId}, %X{camel.routeId}, %X{camel.exchangeId}, %X{camel.messageId}, %X{customHead}, %X{customProp}]
-```
+----
 
 During the execution you can verify the output of the log to see the traces appended to your logger.
 
@@ -54,7 +60,7 @@ This is the list of default MDC values included in each execution:
 * camel.contextId
 * camel.threadId
 
-If they exists in the Exchange, then, they will be included in the MDC. You can use `camel.mdc.customHeaders` and `camel.mdc.customProperties` to include any further header and property you need to trace.
+If they exist in the Exchange, then, they will be included in the MDC. You can use `camel.mdc.customHeaders` and `camel.mdc.customProperties` to include any further header and property you need to trace.
 
 == Configuration
 

components/camel-mdc/src/main/java/org/apache/camel/mdc/MDCProcessorsInterceptStrategy.java

@@ -16,6 +16,7 @@
  */
 package org.apache.camel.mdc;
 
+import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 
 import org.apache.camel.AsyncCallback;
@@ -26,10 +27,15 @@
 import org.apache.camel.Processor;
 import org.apache.camel.spi.InterceptStrategy;
 import org.apache.camel.support.AsyncProcessorConverterHelper;
+import org.slf4j.MDC;
 
 /**
  * MDCProcessorsInterceptStrategy is used to wrap each processor calls and generate the MDC context for each process
- * execution.
+ * execution. IMPORTANT NOTE: When working in async mode there is no possible way to clean the thread MDC context
+ * reliably as any spinoff process (for example, InterceptSendToEndpoint EIP) would loose the possibility to reuse the
+ * context map previously set by this InterceptStrategy. This is not a consistency problem, since, the MDC service is in
+ * charge to reset the MDC context at every exchange execution with the values expected for each execution (either
+ * synchronous or asynchronous).
  */
 public class MDCProcessorsInterceptStrategy implements InterceptStrategy {
 
@@ -53,12 +59,20 @@ public Processor wrapProcessorInInterceptors(
 
             @Override
             public boolean process(Exchange exchange, AsyncCallback callback) {
+                Map<String, String> previousContext = MDC.getCopyOfContextMap();
                 mdcService.setMDC(exchange);
-                boolean answer = asyncProcessor.process(exchange, doneSync -> {
-                    callback.done(doneSync);
+                return asyncProcessor.process(exchange, doneSync -> {
+                    try {
+                        callback.done(doneSync);
+                    } finally {
+                        mdcService.unsetMDC(exchange);
+                        if (previousContext != null) {
+                            MDC.setContextMap(previousContext);
+                        } else {
+                            MDC.clear();
+                        }
+                    }
                 });
-                mdcService.unsetMDC(exchange);
-                return answer;
             }
 
             @Override
@@ -74,15 +88,20 @@ public void process(Exchange exchange) throws Exception {
             @Override
             public CompletableFuture<Exchange> processAsync(Exchange exchange) {
                 CompletableFuture<Exchange> future = new CompletableFuture<>();
+                Map<String, String> previousContext = MDC.getCopyOfContextMap();
                 mdcService.setMDC(exchange);
                 asyncProcessor.process(exchange, doneSync -> {
                     if (exchange.getException() != null) {
                         future.completeExceptionally(exchange.getException());
                     } else {
                         future.complete(exchange);
                     }
+                    if (previousContext != null) {
+                        MDC.setContextMap(previousContext);
+                    } else {
+                        MDC.clear();
+                    }
                 });
-                mdcService.unsetMDC(exchange);
                 return future;
             }
         };

components/camel-mdc/src/main/java/org/apache/camel/mdc/MDCService.java

@@ -16,6 +16,8 @@
  */
 package org.apache.camel.mdc;
 
+import java.util.Map;
+
 import org.apache.camel.CamelContext;
 import org.apache.camel.Exchange;
 import org.apache.camel.ExchangePropertyKey;
@@ -137,15 +139,21 @@ protected void unsetMDC(Exchange exchange) {
 
     private final class MDCLogListener implements LogListener {
 
+        // NOTE: the onLog and afterLog are executed on the same thread, so we can
+        // reliably store the context here.
+        Map<String, String> previousContext;
+
         @Override
         public String onLog(Exchange exchange, CamelLogger camelLogger, String message) {
+            previousContext = MDC.getCopyOfContextMap();
             setMDC(exchange);
             return message;
         }
 
         @Override
         public void afterLog(Exchange exchange, CamelLogger camelLogger, String message) {
             unsetMDC(exchange);
+            MDC.setContextMap(previousContext);
         }
     }
 

components/camel-mdc/src/test/java/org/apache/camel/mdc/MDCAllHeadersTest.java

@@ -22,19 +22,16 @@
 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.test.junit5.ExchangeTestSupport;
 import org.junit.jupiter.api.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import org.slf4j.MDC;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 
 public class MDCAllHeadersTest extends ExchangeTestSupport {
 
-    private static final Logger LOG = LoggerFactory.getLogger(MDCSelectedPropertiesTest.class);
-
     @Override
     protected CamelContext createCamelContext() throws Exception {
         MDCService mdcSvc = new MDCService();
@@ -46,10 +43,22 @@ protected CamelContext createCamelContext() throws Exception {
     }
 
     @Test
-    void testRouteSingleRequest() throws IOException {
+    void testRouteSingleRequest() throws IOException, InterruptedException {
+        MockEndpoint mock = getMockEndpoint("mock:assertMdc");
+        mock.expectedMessageCount(1);
+        mock.whenAnyExchangeReceived(exchange -> {
+            assertNotNull(MDC.get(MDCService.MDC_MESSAGE_ID));
+            assertNotNull(MDC.get(MDCService.MDC_EXCHANGE_ID));
+            assertNotNull(MDC.get(MDCService.MDC_ROUTE_ID));
+            assertNotNull(MDC.get(MDCService.MDC_CAMEL_CONTEXT_ID));
+
+            assertEquals("Header1", MDC.get("head1"));
+            assertEquals("Header2", MDC.get("head2"));
+        });
+
         template.request("direct:start", null);
-        // We should get no MDC after the route has been executed
-        assertEquals(0, MDC.getCopyOfContextMap().size());
+
+        mock.assertIsSatisfied();
     }
 
     @Override
@@ -62,15 +71,7 @@ public void configure() {
                         .log("A message")
                         .setHeader("head1", simple("Header1"))
                         .setHeader("head2", simple("Header2"))
-                        .process(exchange -> {
-                            LOG.info("A process");
-                            assertNotNull(MDC.get(MDCService.MDC_MESSAGE_ID));
-                            assertNotNull(MDC.get(MDCService.MDC_EXCHANGE_ID));
-                            assertNotNull(MDC.get(MDCService.MDC_ROUTE_ID));
-                            assertNotNull(MDC.get(MDCService.MDC_CAMEL_CONTEXT_ID));
-                            assertEquals("Header1", MDC.get("head1"));
-                            assertEquals("Header2", MDC.get("head2"));
-                        })
+                        .to("mock:assertMdc")
                         .to("log:info");
             }
         };

components/camel-mdc/src/test/java/org/apache/camel/mdc/MDCAllPropertiesTest.java

@@ -16,25 +16,20 @@
  */
 package org.apache.camel.mdc;
 
-import java.io.IOException;
-
 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.test.junit5.ExchangeTestSupport;
 import org.junit.jupiter.api.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import org.slf4j.MDC;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 
 public class MDCAllPropertiesTest extends ExchangeTestSupport {
 
-    private static final Logger LOG = LoggerFactory.getLogger(MDCSelectedPropertiesTest.class);
-
     @Override
     protected CamelContext createCamelContext() throws Exception {
         MDCService mdcSvc = new MDCService();
@@ -46,10 +41,26 @@ protected CamelContext createCamelContext() throws Exception {
     }
 
     @Test
-    void testRouteSingleRequest() throws IOException {
+    void testRouteSingleRequest() throws Exception {
+        MockEndpoint mock = getMockEndpoint("mock:assertMdc");
+
+        mock.expectedMessageCount(1);
+        mock.whenAnyExchangeReceived(exchange -> {
+            // Assert MDC values
+            assertNotNull(MDC.get(MDCService.MDC_MESSAGE_ID));
+            assertNotNull(MDC.get(MDCService.MDC_EXCHANGE_ID));
+            assertNotNull(MDC.get(MDCService.MDC_ROUTE_ID));
+            assertNotNull(MDC.get(MDCService.MDC_CAMEL_CONTEXT_ID));
+            // Assert exchange properties are correctly set
+            assertEquals("Property1", exchange.getProperty("prop1"));
+            assertEquals("Property2", exchange.getProperty("prop2"));
+        });
+
+        // Trigger the route
         template.request("direct:start", null);
-        // We should get no MDC after the route has been executed
-        assertEquals(0, MDC.getCopyOfContextMap().size());
+
+        // Wait for assertions to pass
+        mock.assertIsSatisfied();
     }
 
     @Override
@@ -60,17 +71,9 @@ public void configure() {
                 from("direct:start")
                         .routeId("start")
                         .log("A message")
-                        .setProperty("prop1", simple("Property1"))
-                        .setProperty("prop2", simple("Property2"))
-                        .process(exchange -> {
-                            LOG.info("A process");
-                            assertNotNull(MDC.get(MDCService.MDC_MESSAGE_ID));
-                            assertNotNull(MDC.get(MDCService.MDC_EXCHANGE_ID));
-                            assertNotNull(MDC.get(MDCService.MDC_ROUTE_ID));
-                            assertNotNull(MDC.get(MDCService.MDC_CAMEL_CONTEXT_ID));
-                            assertEquals("Property1", MDC.get("prop1"));
-                            assertEquals("Property2", MDC.get("prop2"));
-                        })
+                        .setProperty("prop1", constant("Property1"))
+                        .setProperty("prop2", constant("Property2"))
+                        .to("mock:assertMdc")
                         .to("log:info");
             }
         };

components/camel-mdc/src/test/java/org/apache/camel/mdc/MDCAsyncTest.java

@@ -21,6 +21,7 @@
 import org.apache.camel.Exchange;
 import org.apache.camel.Processor;
 import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
 import org.apache.camel.test.junit5.ExchangeTestSupport;
 import org.junit.jupiter.api.Test;
 import org.slf4j.MDC;
@@ -44,15 +45,21 @@ protected CamelContext createCamelContext() throws Exception {
 
     @Test
     public void testAsyncEndpoint() throws Exception {
-        getMockEndpoint("mock:before").expectedBodiesReceived("Hello Camel");
-        getMockEndpoint("mock:after").expectedBodiesReceived("Bye Camel");
-        getMockEndpoint("mock:result").expectedBodiesReceived("Bye Camel");
+        MockEndpoint before = getMockEndpoint("mock:before");
+        MockEndpoint after = getMockEndpoint("mock:after");
+        MockEndpoint result = getMockEndpoint("mock:result");
+        before.expectedBodiesReceived("Hello Camel");
+        after.expectedBodiesReceived("Bye Camel");
+        result.expectedBodiesReceived("Bye Camel");
 
         String reply = template.requestBody("direct:start", "Hello Camel", String.class);
         assertEquals("Bye Camel", reply);
 
-        // We should get no MDC after the route has been executed
-        assertEquals(0, MDC.getCopyOfContextMap().size());
+        before.assertIsSatisfied();
+        after.assertIsSatisfied();
+        result.assertIsSatisfied();
+
+        // NOTE: more assertions directly in process as it was simpler to verify the condition while executing the async process.
     }
 
     @Override

components/camel-mdc/src/test/java/org/apache/camel/mdc/MDCAsyncWiretapTest.java

@@ -56,8 +56,8 @@ void testRouteSingleRequest() throws IOException, InterruptedException {
         mock.setAssertPeriod(5000);
         context.createProducerTemplate().sendBody("direct:start", null);
         mock.assertIsSatisfied(1000);
-        // We should get no MDC after the route has been executed
-        assertEquals(0, MDC.getCopyOfContextMap().size());
+
+        // NOTE: more assertions directly in process as it was simpler to verify the condition while executing the async process.
     }
 
     @Override