https://issues.apache.org/jira/browse/AMQ-3219 - add basic support for MDC and enable broker name to appear in log entries

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@1081502 13f79535-47bb-0310-9956-ffa450edef68

activemq-core/src/main/java/org/apache/activemq/broker/BrokerService.java

@@ -95,6 +95,8 @@ import org.apache.activemq.usage.SystemUsage;
 import org.apache.activemq.util.*;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.slf4j.MDC;
+
 /**
  * Manages the lifecycle of an ActiveMQ Broker. A BrokerService consists of a
  * number of transport connectors, network connectors and a bunch of properties
@@ -471,6 +473,9 @@ public class BrokerService implements Service {
             // throw new IllegalStateException("Allready started.");
             return;
         }
+
+        MDC.put("broker", brokerName);
+
         try {
         	if (systemExitOnShutdown && useShutdownHook) {
         		throw new ConfigurationException("'useShutdownHook' property cannot be be used with 'systemExitOnShutdown', please turn it off (useShutdownHook=false)");
@@ -537,6 +542,8 @@ public class BrokerService implements Service {
                 LOG.warn("Failed to stop broker after failure in start ", ex);
             }
             throw e;
+        } finally {
+            MDC.remove("broker");
         }
     }
 
@@ -551,6 +558,8 @@ public class BrokerService implements Service {
             return;
         }
 
+        MDC.put("broker", brokerName);
+
         if (systemExitOnShutdown) {
         	new Thread() {
         		@Override
@@ -639,6 +648,8 @@ public class BrokerService implements Service {
             }
         }
 
+        MDC.remove("broker");
+
         stopper.throwFirstException();
     }
     
@@ -2100,9 +2111,11 @@ public class BrokerService implements Service {
                         connector.setBrokerURL(getDefaultSocketURIString());
                     }
                     if (networkConnectorStartExecutor != null) {
+                        final Map context = MDCHelper.getCopyOfContextMap();
                         networkConnectorStartExecutor.execute(new Runnable() {
                             public void run() {
                                 try {
+                                    MDCHelper.setContextMap(context);
                                     LOG.info("Async start of " + connector);
                                     connector.start();
                                 } catch(Exception e) {

activemq-core/src/main/java/org/apache/activemq/broker/TransportConnection.java

@@ -90,10 +90,7 @@ import org.apache.activemq.transport.ResponseCorrelator;
 import org.apache.activemq.transport.Transport;
 import org.apache.activemq.transport.TransportDisposedIOException;
 import org.apache.activemq.transport.TransportFactory;
-import org.apache.activemq.util.IntrospectionSupport;
+import org.apache.activemq.util.*;
-import org.apache.activemq.util.MarshallingSupport;
-import org.apache.activemq.util.ServiceSupport;
-import org.apache.activemq.util.URISupport;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -946,10 +943,12 @@ public class TransportConnection implements Connection, Task, CommandVisitor {
                 cs.getContext().getStopping().set(true);
             }
             try {
+                final Map context = MDCHelper.getCopyOfContextMap();
                 getDefaultTaskRunnerFactory().execute(new Runnable(){
                     public void run() {
                         serviceLock.writeLock().lock();
                         try {
+                            MDCHelper.setContextMap(context);
                             doStop();
                         } catch (Throwable e) {
                             LOG.debug("Error occured while shutting down a connection to '" + transport.getRemoteAddress()

activemq-core/src/main/java/org/apache/activemq/broker/TransportConnector.java

@@ -21,6 +21,7 @@ import java.io.IOException;
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.util.Iterator;
+import java.util.Map;
 import java.util.StringTokenizer;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.regex.Pattern;
@@ -38,6 +39,7 @@ import org.apache.activemq.transport.TransportFactory;
 import org.apache.activemq.transport.TransportServer;
 import org.apache.activemq.transport.discovery.DiscoveryAgent;
 import org.apache.activemq.transport.discovery.DiscoveryAgentFactory;
+import org.apache.activemq.util.MDCHelper;
 import org.apache.activemq.util.ServiceStopper;
 import org.apache.activemq.util.ServiceSupport;
 import org.slf4j.Logger;
@@ -207,11 +209,13 @@ public class TransportConnector implements Connector, BrokerServiceAware {
         brokerInfo.setPeerBrokerInfos(broker.getPeerBrokerInfos());
         brokerInfo.setFaultTolerantConfiguration(broker.isFaultTolerantConfiguration());
         brokerInfo.setBrokerURL(getServer().getConnectURI().toString());
+        final Map context = MDCHelper.getCopyOfContextMap();
         getServer().setAcceptListener(new TransportAcceptListener() {
             public void onAccept(final Transport transport) {
                 try {
                     getDefaultTaskRunnerFactory().execute(new Runnable() {
                         public void run() {
+                            MDCHelper.setContextMap(context);
                             try {
                                 Connection connection = createConnection(transport);
                                 connection.start();

activemq-core/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java

@@ -19,10 +19,7 @@ package org.apache.activemq.network;
 import java.io.IOException;
 import java.security.GeneralSecurityException;
 import java.security.cert.X509Certificate;
-import java.util.Arrays;
+import java.util.*;
-import java.util.Collection;
-import java.util.List;
-import java.util.Properties;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
@@ -75,14 +72,10 @@ import org.apache.activemq.transport.TransportDisposedIOException;
 import org.apache.activemq.transport.TransportFilter;
 import org.apache.activemq.transport.TransportListener;
 import org.apache.activemq.transport.tcp.SslTransport;
-import org.apache.activemq.util.IdGenerator;
+import org.apache.activemq.util.*;
-import org.apache.activemq.util.IntrospectionSupport;
-import org.apache.activemq.util.LongSequenceGenerator;
-import org.apache.activemq.util.MarshallingSupport;
-import org.apache.activemq.util.ServiceStopper;
-import org.apache.activemq.util.ServiceSupport;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.slf4j.MDC;
 
 /**
  * A useful base class for implementing demand forwarding bridges.
@@ -243,8 +236,10 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br
     }
 
     protected void triggerLocalStartBridge() throws IOException {
+        final Map context = MDCHelper.getCopyOfContextMap();
         asyncTaskRunner.execute(new Runnable() {
             public void run() {
+                MDCHelper.setContextMap(context);
                 final String originalName = Thread.currentThread().getName();
                 Thread.currentThread().setName("StartLocalBridge: localBroker=" + localBroker);
                 try {
@@ -259,8 +254,10 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br
     }
 
     protected void triggerRemoteStartBridge() throws IOException {
+        final Map context = MDCHelper.getCopyOfContextMap();
         asyncTaskRunner.execute(new Runnable() {
             public void run() {
+                MDCHelper.setContextMap(context);
                 final String originalName = Thread.currentThread().getName();
                 Thread.currentThread().setName("StartRemotelBridge: localBroker=" + localBroker);
                 try {
@@ -391,9 +388,11 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br
                 try {
                     remoteBridgeStarted.set(false);
                     final CountDownLatch sendShutdown = new CountDownLatch(1);
+                    final Map map = MDCHelper.getCopyOfContextMap();
                     asyncTaskRunner.execute(new Runnable() {
                         public void run() {
                             try {
+                                MDCHelper.setContextMap(map);
                                 localBroker.oneway(new ShutdownInfo());
                                 sendShutdown.countDown();
                                 remoteBroker.oneway(new ShutdownInfo());
@@ -435,8 +434,10 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br
                 LOG.warn("Network connection between " + localBroker + " and " + remoteBroker + " shutdown due to a remote error: " + error);
             }
             LOG.debug("The remote Exception was: " + error, error);
+            final Map map = MDCHelper.getCopyOfContextMap();
             asyncTaskRunner.execute(new Runnable() {
                 public void run() {
+                    MDCHelper.setContextMap(map);
                     ServiceSupport.dispose(getControllingService());
                 }
             });
@@ -651,8 +652,10 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br
         if (!disposed.get()) {
             LOG.info("Network connection between " + localBroker + " and " + remoteBroker + " shutdown due to a local error: " + error);
             LOG.debug("The local Exception was:" + error, error);
+            final Map map = MDCHelper.getCopyOfContextMap();
             asyncTaskRunner.execute(new Runnable() {
                 public void run() {
+                    MDCHelper.setContextMap(map);
                     ServiceSupport.dispose(getControllingService());
                 }
             });
@@ -678,8 +681,10 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br
             subscriptionMapByLocalId.remove(sub.getLocalInfo().getConsumerId());
 
             // continue removal in separate thread to free up this thread for outstanding responses
+            final Map map = MDCHelper.getCopyOfContextMap();
             asyncTaskRunner.execute(new Runnable() {
                 public void run() {
+                    MDCHelper.setContextMap(map);
                     sub.waitForCompletion();
                     try {
                         localBroker.oneway(sub.getLocalInfo().createRemoveCommand());

activemq-core/src/main/java/org/apache/activemq/thread/DedicatedTaskRunner.java

@@ -16,6 +16,10 @@
  */
 package org.apache.activemq.thread;
 
+import org.apache.activemq.util.MDCHelper;
+
+import java.util.Map;
+
 /**
  * 
  */
@@ -31,8 +35,10 @@ class DedicatedTaskRunner implements TaskRunner {
 
     public DedicatedTaskRunner(Task task, String name, int priority, boolean daemon) {
         this.task = task;
+        final Map context = MDCHelper.getCopyOfContextMap();
         thread = new Thread(name) {
             public void run() {
+                MDCHelper.setContextMap(context);
                 runTask();
             }
         };

activemq-core/src/main/java/org/apache/activemq/thread/DeterministicTaskRunner.java

@@ -16,6 +16,10 @@
  */
 package org.apache.activemq.thread;
 
+import org.apache.activemq.util.MDCHelper;
+import org.slf4j.MDC;
+
+import java.util.Map;
 import java.util.concurrent.Executor;
 
 /**
@@ -35,9 +39,13 @@ public class DeterministicTaskRunner implements TaskRunner {
     public DeterministicTaskRunner(Executor executor, Task task) {
         this.executor = executor;
         this.task = task;
+        final Map context = MDCHelper.getCopyOfContextMap();
         this.runable = new Runnable() {
             public void run() {
                 Thread.currentThread();
+                if (context != null) {
+                    MDCHelper.setContextMap(context);
+                }
                 runTask();
             }
         };

activemq-core/src/main/java/org/apache/activemq/thread/PooledTaskRunner.java

@@ -16,6 +16,9 @@
  */
 package org.apache.activemq.thread;
 
+import org.apache.activemq.util.MDCHelper;
+
+import java.util.Map;
 import java.util.concurrent.Executor;
 
 /**
@@ -36,8 +39,10 @@ class PooledTaskRunner implements TaskRunner {
         this.executor = executor;
         this.maxIterationsPerRun = maxIterationsPerRun;
         this.task = task;
+        final Map context = MDCHelper.getCopyOfContextMap();
         runable = new Runnable() {
             public void run() {
+                MDCHelper.setContextMap(context);
                 runningThread = Thread.currentThread();
                 try {
                     runTask();

activemq-core/src/main/java/org/apache/activemq/transport/discovery/simple/SimpleDiscoveryAgent.java

@@ -18,6 +18,7 @@ package org.apache.activemq.transport.discovery.simple;
 
 import java.io.IOException;
 import java.net.URI;
+import java.util.Map;
 import java.util.concurrent.SynchronousQueue;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.ThreadPoolExecutor;
@@ -28,6 +29,7 @@ import org.apache.activemq.command.DiscoveryEvent;
 import org.apache.activemq.thread.DefaultThreadPools;
 import org.apache.activemq.transport.discovery.DiscoveryAgent;
 import org.apache.activemq.transport.discovery.DiscoveryListener;
+import org.apache.activemq.util.MDCHelper;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -110,9 +112,12 @@ public class SimpleDiscoveryAgent implements DiscoveryAgent {
         if (event.failed.compareAndSet(false, true)) {
 
             listener.onServiceRemove(event);
+            final Map context = MDCHelper.getCopyOfContextMap();
             DefaultThreadPools.getDefaultTaskRunnerFactory().execute(new Runnable() {
                 public void run() {
 
+                    MDCHelper.setContextMap(context);
+
                     // We detect a failed connection attempt because the service
                     // fails right
                     // away.

activemq-core/src/test/resources/log4j.properties

@@ -30,11 +30,11 @@ log4j.rootLogger=INFO, out, stdout
 # CONSOLE appender not used by default
 log4j.appender.stdout=org.apache.log4j.ConsoleAppender
 log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
-log4j.appender.stdout.layout.ConversionPattern=%d [%-15.15t] %-5p %-30.30c{1} - %m%n
+log4j.appender.stdout.layout.ConversionPattern=%d [%-15.15t] %-10.10X{broker} - %-5p %-30.30c{1} - %m%n
 
 # File appender
 log4j.appender.out=org.apache.log4j.FileAppender
 log4j.appender.out.layout=org.apache.log4j.PatternLayout
-log4j.appender.out.layout.ConversionPattern=%d [%-15.15t] %-5p %-30.30c{1} - %m%n
+log4j.appender.out.layout.ConversionPattern=%d [%-15.15t] %-10.10X{broker} - %-5p %-30.30c{1} - %m%n
 log4j.appender.out.file=target/activemq-test.log
 log4j.appender.out.append=true

assembly/src/release/conf/log4j.properties

@@ -39,6 +39,8 @@ log4j.logger.org.apache.camel=INFO
 log4j.appender.console=org.apache.log4j.ConsoleAppender
 log4j.appender.console.layout=org.apache.log4j.PatternLayout
 log4j.appender.console.layout.ConversionPattern=%5p | %m%n
+# use the pattern below if you want to see mdc data, like broker name, etc
+#log4j.appender.console.layout.ConversionPattern=%5p | %-10.10X{broker} | %m%n
 log4j.appender.console.threshold=INFO
 
 # File appender
@@ -49,6 +51,8 @@ log4j.appender.logfile.maxBackupIndex=5
 log4j.appender.logfile.append=true
 log4j.appender.logfile.layout=org.apache.log4j.PatternLayout
 log4j.appender.logfile.layout.ConversionPattern=%d | %-5p | %m | %c | %t%n
+# use the pattern below if you want to see mdc data, like broker name, etc
+#log4j.appender.logfile.layout.ConversionPattern=%d | %-10.10X{broker} | %-5p | %m | %c | %t%n
 
 ###########
 # Audit log