QPID-4271: improve behaviour when embedding the broker inside a container

Avoid potential ThreadLocal leaks on Container owned threads for CurrentActor, AMQShortString and SecurityManager.
Have LogRecorder unregistered itself from Log4J.
Allow SIGHUP handling to be turned off (inappropiate to install signal handling when deployed inside Container.
Allow use of custom RMI socket factory to be disabled.  (The registration of a custom RMI socket with the JRE cannot be reversed
(deficiency in JRE API) and this causes a large perm-gen leak).

Work of Robbie Gemmell <robbie@apache.org> and myself.

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

11 files changed, 128 insertions, 48 deletions

diff --git a/java/broker-plugins/management-http/src/main/java/org/apache/qpid/server/management/plugin/servlet/rest/AbstractServlet.java b/java/broker-plugins/management-http/src/main/java/org/apache/qpid/server/management/plugin/servlet/rest/AbstractServlet.java
index 843ba9816a..521ad69abe 100644
--- a/java/broker-plugins/management-http/src/main/java/org/apache/qpid/server/management/plugin/servlet/rest/AbstractServlet.java
+++ b/java/broker-plugins/management-http/src/main/java/org/apache/qpid/server/management/plugin/servlet/rest/AbstractServlet.java
@@ -36,6 +36,7 @@ import javax.servlet.http.HttpServletResponse;
 import javax.servlet.http.HttpSession;
 import org.apache.commons.codec.binary.Base64;
 import org.apache.log4j.Logger;
+import org.apache.qpid.framing.AMQShortString;
 import org.apache.qpid.server.logging.LogActor;
 import org.apache.qpid.server.logging.RootMessageLogger;
 import org.apache.qpid.server.logging.actors.CurrentActor;
@@ -218,7 +219,14 @@ public abstract class AbstractServlet extends HttpServlet
         }
         finally
         {
-            org.apache.qpid.server.security.SecurityManager.setThreadSubject(null);
+            try
+            {
+                org.apache.qpid.server.security.SecurityManager.setThreadSubject(null);
+            }
+            finally
+            {
+                AMQShortString.clearLocalCache();
+            }
         }
     }
 
diff --git a/java/broker-plugins/management-jmx/src/main/java/org/apache/qpid/server/jmx/JMXManagedObjectRegistry.java b/java/broker-plugins/management-jmx/src/main/java/org/apache/qpid/server/jmx/JMXManagedObjectRegistry.java
index 44fe4787bb..cb2e9f5e54 100644
--- a/java/broker-plugins/management-jmx/src/main/java/org/apache/qpid/server/jmx/JMXManagedObjectRegistry.java
+++ b/java/broker-plugins/management-jmx/src/main/java/org/apache/qpid/server/jmx/JMXManagedObjectRegistry.java
@@ -211,10 +211,12 @@ public class JMXManagedObjectRegistry implements ManagedObjectRegistry
         System.setProperty("java.rmi.server.randomIDs", "true");
         if(_useCustomSocketFactory)
         {
+            _log.debug("Using custom RMIServerSocketFactory");
             _rmiRegistry = LocateRegistry.createRegistry(_jmxPortRegistryServer, null, new CustomRMIServerSocketFactory());
         }
         else
         {
+            _log.debug("Using default RMIServerSocketFactory");
             _rmiRegistry = LocateRegistry.createRegistry(_jmxPortRegistryServer, null, null);
         }
 
@@ -235,7 +237,7 @@ public class JMXManagedObjectRegistry implements ManagedObjectRegistry
 
             /**
              * Override makeClient so we can cache the username of the client in a Map keyed by connectionId.
-             * ConnectionId is guaranteed to be unique per client connection, according to the JMS spec.
+             * ConnectionId is guaranteed to be unique per client connection, according to the JMX spec.
              * An instance of NotificationListener (mapCleanupListener) will be responsible for removing these Map
              * entries.
              *
diff --git a/java/broker/etc/log4j.xml b/java/broker/etc/log4j.xml
index b1b31248c1..7392260a0a 100644
--- a/java/broker/etc/log4j.xml
+++ b/java/broker/etc/log4j.xml
@@ -88,9 +88,9 @@
     </appender>
 
     <!-- Provide warnings to standard output -->
-    <category additivity="true" name="org.apache.qpid">
-        <priority value="warn"/>
-    </category>
+    <logger additivity="true" name="org.apache.qpid">
+        <level value="warn"/>
+    </logger>
 
     <!-- Enable info messages for the status-logging hierarchy -->
     <logger additivity="true" name="qpid.message">
@@ -108,21 +108,14 @@
       <level value="info"/>
     </logger>
 
-    <!-- Examples of additional logging settings -->
-    <!-- Used to generate extra debug. See debug.log4j.xml -->
-    
-    <!--<category additivity="true" name="org.apache.qpid.server.store">
-        <priority value="debug"/>
-    </category-->
-
     <!-- Set the commons logging that the XML parser uses to WARN, it is very chatty at debug -->
     <logger name="org.apache.commons">
-        <level value="WARN"/>
+        <level value="warn"/>
     </logger>
 
     <!-- Log all info events to file -->
     <root>
-        <priority value="info"/>
+        <level value="info"/>
         <appender-ref ref="FileAppender"/>
         <!--appender-ref ref="ArchivingFileAppender"/-->
     </root>
diff --git a/java/broker/src/main/java/org/apache/qpid/server/Broker.java b/java/broker/src/main/java/org/apache/qpid/server/Broker.java
index d58a0d5bb4..e92da34a49 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/Broker.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/Broker.java
@@ -29,6 +29,7 @@ import java.util.*;
 import javax.net.ssl.SSLContext;
 import org.apache.log4j.Logger;
 import org.apache.log4j.PropertyConfigurator;
+import org.apache.qpid.framing.AMQShortString;
 import org.apache.qpid.server.configuration.ServerConfiguration;
 import org.apache.qpid.server.configuration.ServerNetworkTransportConfiguration;
 import org.apache.qpid.server.logging.SystemOutMessageLogger;
@@ -73,7 +74,14 @@ public class Broker
         }
         finally
         {
-            ApplicationRegistry.remove();
+            try
+            {
+                ApplicationRegistry.remove();
+            }
+            finally
+            {
+                clearAMQShortStringCache();
+            }
         }
     }
 
@@ -84,15 +92,22 @@ public class Broker
 
     public void startup(final BrokerOptions options) throws Exception
     {
+        CurrentActor.set(new BrokerActor(new SystemOutMessageLogger()));
         try
         {
-            CurrentActor.set(new BrokerActor(new SystemOutMessageLogger()));
             startupImpl(options);
             addShutdownHook();
         }
         finally
         {
-            CurrentActor.remove();
+            try
+            {
+                CurrentActor.remove();
+            }
+            finally
+            {
+                clearAMQShortStringCache();
+            }
         }
     }
 
@@ -368,7 +383,7 @@ public class Broker
 
         if (!configFile.exists() && throwOnFileNotFound)
         {
-            String error = "File " + fileName + " could not be found. Check the file exists and is readable.";
+            String error = "File " + configFile + " could not be found. Check the file exists and is readable.";
 
             if (qpidHome == null)
             {
@@ -540,4 +555,14 @@ public class Broker
             Broker.this.shutdown();
         }
     }
+
+    /**
+     * Workaround that prevents AMQShortStrings cache from being left in the thread local. This is important
+     * when embedding the Broker in containers where the starting thread may not belong to Qpid.
+     * The long term solution here is to stop our use of AMQShortString outside the AMQP transport layer.
+     */
+    private void clearAMQShortStringCache()
+    {
+        AMQShortString.clearLocalCache();
+    }
 }
diff --git a/java/broker/src/main/java/org/apache/qpid/server/configuration/ServerConfiguration.java b/java/broker/src/main/java/org/apache/qpid/server/configuration/ServerConfiguration.java
index 99625a22ad..f94bd684e9 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/configuration/ServerConfiguration.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/configuration/ServerConfiguration.java
@@ -67,6 +67,8 @@ public class ServerConfiguration extends ConfigurationPlugin
     public static final int DEFAULT_HTTP_MANAGEMENT_PORT = 8080;
     public static final int DEFAULT_HTTPS_MANAGEMENT_PORT = 8443;
     public static final long DEFAULT_MINIMUM_ALERT_REPEAT_GAP = 30000l;
+    public static final String SKIP_SIGHUP_HANDLER_REGISTRATION = "qpid.skip_sighup_handler_registration";
+    public static final String USE_CUSTOM_RMI_SOCKET_FACTORY = "qpid.use_custom_rmi_socket_factory";
 
     public static final String QPID_HOME = "QPID_HOME";
     public static final String QPID_WORK = "QPID_WORK";
@@ -156,6 +158,18 @@ public class ServerConfiguration extends ConfigurationPlugin
         this(parseConfig(configurationURL));
         _configFile = configurationURL;
 
+        if(!Boolean.getBoolean(SKIP_SIGHUP_HANDLER_REGISTRATION))
+        {
+            registerSigHupHandler();
+        }
+        else
+        {
+            _logger.info("Skipping registration of Signal HUP handler.");
+        }
+    }
+
+    private void registerSigHupHandler()
+    {
         SignalHandlerTask hupReparseTask = new SignalHandlerTask()
         {
             public void handle()
@@ -562,7 +576,8 @@ public class ServerConfiguration extends ConfigurationPlugin
 
     public boolean getUseCustomRMISocketFactory()
     {
-        return getBooleanValue(MGMT_CUSTOM_REGISTRY_SOCKET, true);
+        return getBooleanValue(MGMT_CUSTOM_REGISTRY_SOCKET,
+                               Boolean.parseBoolean(System.getProperty(USE_CUSTOM_RMI_SOCKET_FACTORY, "true")));
     }
 
     public void setUseCustomRMISocketFactory(boolean bool)
diff --git a/java/broker/src/main/java/org/apache/qpid/server/logging/Log4jMessageLogger.java b/java/broker/src/main/java/org/apache/qpid/server/logging/Log4jMessageLogger.java
index ec506ab51c..62f0e75ceb 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/logging/Log4jMessageLogger.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/logging/Log4jMessageLogger.java
@@ -20,15 +20,11 @@
  */
 package org.apache.qpid.server.logging;
 
-import org.apache.log4j.Level;
 import org.apache.log4j.Logger;
-
 import org.apache.qpid.server.configuration.ServerConfiguration;
 
 public class Log4jMessageLogger extends AbstractRootMessageLogger
 {
-    public static final Level LEVEL = Level.toLevel("INFO");
-    
     public Log4jMessageLogger()
     {
         super();
@@ -51,7 +47,7 @@ public class Log4jMessageLogger extends AbstractRootMessageLogger
         if(isEnabled())
         {
             Logger logger = Logger.getLogger(logHierarchy);
-            return logger.isEnabledFor(LEVEL);
+            return logger.isInfoEnabled();
         }
         else
         {
@@ -69,7 +65,6 @@ public class Log4jMessageLogger extends AbstractRootMessageLogger
     public void rawMessage(String message, Throwable throwable, String logHierarchy)
     {
         Logger logger = Logger.getLogger(logHierarchy);
-        
-        logger.log(LEVEL, message, throwable);
+        logger.info(message, throwable);
     }
 }
diff --git a/java/broker/src/main/java/org/apache/qpid/server/logging/LogRecorder.java b/java/broker/src/main/java/org/apache/qpid/server/logging/LogRecorder.java
index a1065319d3..d053dd3fe2 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/logging/LogRecorder.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/logging/LogRecorder.java
@@ -90,7 +90,6 @@ public class LogRecorder implements Appender, Iterable<LogRecorder.Record>
 
     public LogRecorder()
     {
-
         Logger.getRootLogger().addAppender(this);
     }
 
@@ -109,7 +108,11 @@ public class LogRecorder implements Appender, Iterable<LogRecorder.Record>
     @Override
     public void close()
     {
-        //TODO - Implement
+    }
+
+    public void closeLogRecorder()
+    {
+        Logger.getRootLogger().removeAppender(this);
     }
 
     @Override
diff --git a/java/broker/src/main/java/org/apache/qpid/server/logging/actors/CurrentActor.java b/java/broker/src/main/java/org/apache/qpid/server/logging/actors/CurrentActor.java
index 97134515a0..6251471139 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/logging/actors/CurrentActor.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/logging/actors/CurrentActor.java
@@ -105,6 +105,11 @@ public class CurrentActor
     {
         Stack<LogActor> stack = _currentActor.get();
         stack.pop();
+
+        if (stack.isEmpty())
+        {
+            _currentActor.remove();
+        }
     }
 
     /**
diff --git a/java/broker/src/main/java/org/apache/qpid/server/registry/ApplicationRegistry.java b/java/broker/src/main/java/org/apache/qpid/server/registry/ApplicationRegistry.java
index 967c58debb..63d00fd0fc 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/registry/ApplicationRegistry.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/registry/ApplicationRegistry.java
@@ -479,6 +479,8 @@ public abstract class ApplicationRegistry implements IApplicationRegistry
             close(_pluginManager);
 
             CurrentActor.get().message(BrokerMessages.STOPPED());
+
+            _logRecorder.closeLogRecorder();
         }
         finally
         {
diff --git a/java/broker/src/main/java/org/apache/qpid/server/security/SecurityManager.java b/java/broker/src/main/java/org/apache/qpid/server/security/SecurityManager.java
index 088d120821..ce0ea2faea 100755
--- a/java/broker/src/main/java/org/apache/qpid/server/security/SecurityManager.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/security/SecurityManager.java
@@ -69,13 +69,8 @@ public class SecurityManager
     
     /** Container for the {@link java.security.Principal} that is using to this thread. */
     private static final ThreadLocal<Subject> _subject = new ThreadLocal<Subject>();
-    private static final ThreadLocal<Boolean> _accessChecksDisabled = new ThreadLocal<Boolean>()
-    {
-        protected Boolean initialValue()
-        {
-            return false;
-        }
-    };
+
+    public static final ThreadLocal<Boolean> _accessChecksDisabled = new ClearingThreadLocal(false);
 
     private PluginManager _pluginManager;
     private Map<String, SecurityPluginFactory> _pluginFactories = new HashMap<String, SecurityPluginFactory>();
@@ -114,6 +109,50 @@ public class SecurityManager
         }
     }
 
+    /**
+     * A special ThreadLocal, which calls remove() on itself whenever the value is
+     * the default, to avoid leaving a default value set after its use has passed.
+     */
+    private static final class ClearingThreadLocal extends ThreadLocal<Boolean>
+    {
+        private Boolean _defaultValue;
+
+        public ClearingThreadLocal(Boolean defaultValue)
+        {
+            super();
+            _defaultValue = defaultValue;
+        }
+
+        @Override
+        protected Boolean initialValue()
+        {
+            return _defaultValue;
+        }
+
+        @Override
+        public void set(Boolean value)
+        {
+            if (value == _defaultValue)
+            {
+                super.remove();
+            }
+            else
+            {
+                super.set(value);
+            }
+        }
+
+        @Override
+        public Boolean get()
+        {
+            Boolean value = super.get();
+            if (value == _defaultValue)
+            {
+                super.remove();
+            }
+            return value;
+        }
+    }
 
     public SecurityManager(SecurityManager parent) throws ConfigurationException
     {
diff --git a/java/common/src/main/java/org/apache/qpid/framing/AMQShortString.java b/java/common/src/main/java/org/apache/qpid/framing/AMQShortString.java
index fdc71e31f9..1381390640 100644
--- a/java/common/src/main/java/org/apache/qpid/framing/AMQShortString.java
+++ b/java/common/src/main/java/org/apache/qpid/framing/AMQShortString.java
@@ -110,7 +110,7 @@ public final class AMQShortString implements CharSequence, Comparable<AMQShortSt
                 {
                     return new LinkedHashMap<AMQShortString, AMQShortString>()
                     {
-
+                        @Override
                         protected boolean removeEldestEntry(Map.Entry<AMQShortString, AMQShortString> eldest)
                         {
                             return size() > LOCAL_INTERN_CACHE_SIZE;
@@ -845,22 +845,15 @@ public final class AMQShortString implements CharSequence, Comparable<AMQShortSt
         return internString;
     }
 
-
-    public static void main(String args[])
+    public static String toString(AMQShortString amqShortString)
     {
-        AMQShortString s = new AMQShortString("a.b.c.d.e.f.g.h.i.j.k");
-        AMQShortString s2 = s.substring(2, 7);
-
-        AMQShortStringTokenizer t = s2.tokenize((byte) '.');
-        while(t.hasMoreTokens())
-        {
-            System.err.println(t.nextToken());
-        }
+        return amqShortString == null ? null : amqShortString.asString();
     }
 
-    public static String toString(AMQShortString amqShortString)
+    public static void clearLocalCache()
     {
-        return amqShortString == null ? null : amqShortString.asString();
+        _localInternMap.remove();
+        _localStringMap.remove();
     }
 
 }