QPID-1670: Implement an UncaughtExceptionHandler to log exceptions causing the permature termination of Qpid client threads.

Applied patch from Keith Wall <keith.wall@gmail.com>


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

2 files changed, 209 insertions, 13 deletions

diff --git a/java/common/src/test/java/org/apache/qpid/pool/ReferenceCountingExecutorServiceTest.java b/java/common/src/test/java/org/apache/qpid/pool/ReferenceCountingExecutorServiceTest.java
new file mode 100644
index 0000000000..35998de3a1
--- /dev/null
+++ b/java/common/src/test/java/org/apache/qpid/pool/ReferenceCountingExecutorServiceTest.java
@@ -0,0 +1,159 @@
+/*
+ *
+ * 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.qpid.pool;
+
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+
+import junit.framework.TestCase;
+
+
+public class ReferenceCountingExecutorServiceTest extends TestCase
+{
+
+
+    private ReferenceCountingExecutorService _executorService = ReferenceCountingExecutorService.getInstance();  // Class under test
+    private ThreadFactory _beforeExecutorThreadFactory;
+
+
+    @Override
+    protected void setUp() throws Exception
+    {
+        super.setUp();
+        _beforeExecutorThreadFactory = _executorService.getThreadFactory();
+    }
+
+    @Override
+    protected void tearDown() throws Exception
+    {
+        super.tearDown();
+        _executorService.setThreadFactory(_beforeExecutorThreadFactory);
+    }
+
+
+
+    /**
+     * Tests that the ReferenceCountingExecutorService correctly manages the reference count.
+     */
+    public void testReferenceCounting() throws Exception
+    {
+        final int countBefore = _executorService.getReferenceCount();
+
+        try
+        {
+            _executorService.acquireExecutorService();
+            _executorService.acquireExecutorService();
+
+            assertEquals("Reference count should now be +2", countBefore + 2, _executorService.getReferenceCount());
+        }
+        finally
+        {
+            _executorService.releaseExecutorService();
+            _executorService.releaseExecutorService();
+        }
+        assertEquals("Reference count should have returned to the initial value", countBefore, _executorService.getReferenceCount());
+    }
+
+    /**
+     * Tests that the executor creates and executes a task using the default thread pool.
+     */
+    public void testExecuteCommandWithDefaultExecutorThreadFactory() throws Exception
+    {
+        final CountDownLatch latch = new CountDownLatch(1);
+        final Set<ThreadGroup> threadGroups = new HashSet<ThreadGroup>();
+
+        _executorService.acquireExecutorService();
+
+        try
+        {
+            _executorService.getPool().execute(createRunnable(latch, threadGroups));
+
+            latch.await(3, TimeUnit.SECONDS);
+
+            assertTrue("Expect that executor created a thread using default thread factory",
+                    threadGroups.contains(Thread.currentThread().getThreadGroup()));
+        }
+        finally
+        {
+            _executorService.releaseExecutorService();
+        }
+    }
+
+    /**
+     * Tests that the executor creates and executes a task using an overridden thread pool.
+     */
+    public void testExecuteCommandWithOverriddenExecutorThreadFactory() throws Exception
+    {
+        final CountDownLatch latch = new CountDownLatch(1);
+        final ThreadGroup expectedThreadGroup = new ThreadGroup("junit");
+        _executorService.setThreadFactory(new ThreadGroupChangingThreadFactory(expectedThreadGroup));
+        _executorService.acquireExecutorService();
+
+        final Set<ThreadGroup> threadGroups = new HashSet<ThreadGroup>();
+
+        try
+        {
+            _executorService.getPool().execute(createRunnable(latch, threadGroups));
+
+            latch.await(3, TimeUnit.SECONDS);
+
+            assertTrue("Expect that executor created a thread using overridden thread factory",
+                    threadGroups.contains(expectedThreadGroup));
+        }
+        finally
+        {
+            _executorService.releaseExecutorService();
+        }
+    }
+
+    private Runnable createRunnable(final CountDownLatch latch, final Set<ThreadGroup> threadGroups)
+    {
+        return new Runnable()
+        {
+
+            public void run()
+            {
+                threadGroups.add(Thread.currentThread().getThreadGroup());
+                latch.countDown();
+            }
+
+        };
+    }
+
+   private final class ThreadGroupChangingThreadFactory implements ThreadFactory
+   {
+       private final ThreadGroup _newGroup;
+
+       private ThreadGroupChangingThreadFactory(final ThreadGroup newGroup)
+       {
+           this._newGroup = newGroup;
+       }
+
+       public Thread newThread(Runnable r)
+       {
+           return new Thread(_newGroup, r);
+       }
+   }
+
+}
diff --git a/java/common/src/test/java/org/apache/qpid/thread/ThreadFactoryTest.java b/java/common/src/test/java/org/apache/qpid/thread/ThreadFactoryTest.java
index 7f17592893..7b0f93700a 100644
--- a/java/common/src/test/java/org/apache/qpid/thread/ThreadFactoryTest.java
+++ b/java/common/src/test/java/org/apache/qpid/thread/ThreadFactoryTest.java
@@ -1,4 +1,3 @@
-package org.apache.qpid.thread;
 /*
  * 
  * Licensed to the Apache Software Foundation (ASF) under one
@@ -20,18 +19,22 @@ package org.apache.qpid.thread;
  * 
  */
 
+package org.apache.qpid.thread;
 
 import junit.framework.TestCase;
 
+/**
+ * Tests the ThreadFactory.
+ */
 public class ThreadFactoryTest extends TestCase
 {
     public void testThreadFactory()
     {
-        Class threadFactoryClass = null;
+        Class<? extends ThreadFactory> threadFactoryClass = null;
         try
         {
             threadFactoryClass = Class.forName(System.getProperty("qpid.thread_factory",
-                    "org.apache.qpid.thread.DefaultThreadFactory"));            
+                    "org.apache.qpid.thread.DefaultThreadFactory")).asSubclass(ThreadFactory.class);
         }
         // If the thread factory class was wrong it will flagged way before it gets here.
         catch(Exception e)
@@ -41,20 +44,19 @@ public class ThreadFactoryTest extends TestCase
         
         assertEquals(threadFactoryClass, Threading.getThreadFactory().getClass());
     }
-    
-    public void testThreadCreate()
+
+    /**
+     * Tests creating a thread without a priority.   Also verifies that the factory sets the
+     * uncaught exception handler so uncaught exceptions are logged to SLF4J.
+     */
+    public void testCreateThreadWithDefaultPriority()
     {
-        Runnable r = new Runnable(){
-          
-            public void run(){
-                
-            }            
-        };
+        Runnable r = createRunnable();
         
         Thread t = null;
         try
         {
-            t = Threading.getThreadFactory().createThread(r,5);
+            t = Threading.getThreadFactory().createThread(r);
         }
         catch(Exception e)
         {
@@ -62,6 +64,41 @@ public class ThreadFactoryTest extends TestCase
         }
         
         assertNotNull(t);
-        assertEquals(5,t.getPriority());
+        assertEquals(Thread.NORM_PRIORITY, t.getPriority());
+        assertTrue(t.getUncaughtExceptionHandler() instanceof LoggingUncaughtExceptionHandler);
+    }
+
+    /**
+     * Tests creating thread with a priority.   Also verifies that the factory sets the
+     * uncaught exception handler so uncaught exceptions are logged to SLF4J.
+     */
+    public void testCreateThreadWithSpecifiedPriority()
+    {
+        Runnable r = createRunnable();
+
+        Thread t = null;
+        try
+        {
+            t = Threading.getThreadFactory().createThread(r, 4);
+        }
+        catch(Exception e)
+        {
+            fail("Error creating thread using Qpid thread factory");
+        }
+
+        assertNotNull(t);
+        assertEquals(4, t.getPriority());
+        assertTrue(t.getUncaughtExceptionHandler() instanceof LoggingUncaughtExceptionHandler);
+    }
+
+    private Runnable createRunnable()
+    {
+        Runnable r = new Runnable(){
+
+            public void run(){
+
+            }
+        };
+        return r;
     }
 }