CAMEL-22717: camel-mdc - MDC is lost when using interceptSendToEndpoi…

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,31 @@ 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`).
 
+IMPORTANT: Do you 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 +59,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/MDCEventNotifier.java

@@ -0,0 +1,60 @@
+/*
+ * 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.mdc;
+
+import org.apache.camel.spi.CamelEvent;
+import org.apache.camel.support.SimpleEventNotifierSupport;
+
+/**
+ * Use {@link org.apache.camel.spi.EventNotifier} to ensure MDC is unset when exchange is either completed/failed or
+ * handed over to another thread via asynchronous processing.
+ *
+ * @see MDCProcessor
+ */
+class MDCEventNotifier extends SimpleEventNotifierSupport {
+
+    private final MDCService mdc;
+
+    public MDCEventNotifier(MDCService mdc) {
+        this.mdc = mdc;
+    }
+
+    @Override
+    protected void setupIgnore(boolean ignore) {
+        setIgnoreCamelContextEvents(true);
+        setIgnoreCamelContextInitEvents(true);
+        setIgnoreRouteEvents(true);
+        setIgnoreServiceEvents(true);
+        setIgnoreStepEvents(true);
+        // we need also async processing started events
+        setIgnoreExchangeAsyncProcessingStartedEvents(false);
+    }
+
+    @Override
+    public void notify(CamelEvent event) throws Exception {
+        if (event instanceof CamelEvent.ExchangeAsyncProcessingStartedEvent eap) {
+            // exchange is continued processed on another thread so unset MDC
+            mdc.unsetMDC(eap.getExchange());
+        } else if (event instanceof CamelEvent.ExchangeCompletedEvent ec) {
+            // exchange is completed so unset MDC
+            mdc.unsetMDC(ec.getExchange());
+        } else if (event instanceof CamelEvent.ExchangeFailedEvent ef) {
+            // exchange is failed so unset MDC
+            mdc.unsetMDC(ef.getExchange());
+        }
+    }
+}

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

@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.mdc;
+
+import org.apache.camel.AsyncCallback;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.support.processor.DelegateAsyncProcessor;
+
+/**
+ * MDC {@link Processor} that sets MDC context before processing.
+ *
+ * Important: The {@link MDCEventNotifier} is responsible for unsetting MDC because of the exchange can become
+ * asynchronously processed by another thread, which this processor cannot do.
+ *
+ * @see MDCEventNotifier
+ */
+class MDCProcessor extends DelegateAsyncProcessor {
+
+    private final MDCService mdcService;
+
+    public MDCProcessor(MDCService mdcService, Processor processor) {
+        super(processor);
+        this.mdcService = mdcService;
+    }
+
+    @Override
+    public boolean process(Exchange exchange, AsyncCallback callback) {
+        mdcService.setMDC(exchange);
+        return super.process(exchange, callback);
+    }
+}

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

@@ -20,17 +20,19 @@
 import org.apache.camel.Exchange;
 import org.apache.camel.ExchangePropertyKey;
 import org.apache.camel.RuntimeCamelException;
-import org.apache.camel.spi.CamelLogger;
 import org.apache.camel.spi.CamelMDCService;
 import org.apache.camel.spi.InterceptStrategy;
-import org.apache.camel.spi.LogListener;
 import org.apache.camel.spi.annotations.JdkService;
+import org.apache.camel.support.service.ServiceHelper;
 import org.apache.camel.support.service.ServiceSupport;
 import org.apache.camel.util.ObjectHelper;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.slf4j.MDC;
 
+/**
+ * MDC {@link org.apache.camel.Service} that setup MDC support via the camel-mdc component.
+ */
 @JdkService("mdc-service")
 public class MDCService extends ServiceSupport implements CamelMDCService {
 
@@ -43,6 +45,7 @@ public class MDCService extends ServiceSupport implements CamelMDCService {
     static String MDC_CAMEL_CONTEXT_ID = "camel.contextId";
 
     private static final Logger LOG = LoggerFactory.getLogger(MDCService.class);
+    private final MDCEventNotifier eventNotifier = new MDCEventNotifier(this);
 
     private CamelContext camelContext;
 
@@ -92,17 +95,28 @@ public void init(CamelContext camelContext) {
     @Override
     public void doInit() {
         ObjectHelper.notNull(camelContext, "CamelContext", this);
-        camelContext.getCamelContextExtension().addLogListener(new MDCLogListener());
-        InterceptStrategy interceptStrategy = new MDCProcessorsInterceptStrategy(this);
+
+        camelContext.getManagementStrategy().addEventNotifier(eventNotifier);
+
+        InterceptStrategy interceptStrategy
+                = (context, definition, target, nextTarget) -> new MDCProcessor(MDCService.this, target);
         camelContext.getCamelContextExtension().addInterceptStrategy(interceptStrategy);
     }
 
     @Override
     protected void doStart() throws Exception {
         super.doStart();
+        ServiceHelper.startService(eventNotifier);
         LOG.info("Mapped Diagnostic Context (MDC) enabled");
     }
 
+    @Override
+    protected void doShutdown() throws Exception {
+        super.doShutdown();
+        camelContext.getManagementStrategy().removeEventNotifier(eventNotifier);
+        ServiceHelper.stopService(eventNotifier);
+    }
+
     private void setOrUnsetMDC(Exchange exchange, boolean push) {
         try {
             // Default values
@@ -135,20 +149,6 @@ protected void unsetMDC(Exchange exchange) {
         setOrUnsetMDC(exchange, false);
     }
 
-    private final class MDCLogListener implements LogListener {
-
-        @Override
-        public String onLog(Exchange exchange, CamelLogger camelLogger, String message) {
-            setMDC(exchange);
-            return message;
-        }
-
-        @Override
-        public void afterLog(Exchange exchange, CamelLogger camelLogger, String message) {
-            unsetMDC(exchange);
-        }
-    }
-
     // Default basic MDC properties to set/unset MDC context.
     private void prepareMDC(Exchange exchange, boolean push) {
         if (push) {