QPID-327 : Patch supplied by Rob Godfrey - [race condition] PoolingFilter : Possible race condition when completing a Job

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

2 files changed, 22 insertions, 11 deletions

diff --git a/java/common/src/test/java/org/apache/qpid/pool/PoolingFilterTest.java b/java/common/src/test/java/org/apache/qpid/pool/PoolingFilterTest.java
index 9a5208662b..6383d52298 100644
--- a/java/common/src/test/java/org/apache/qpid/pool/PoolingFilterTest.java
+++ b/java/common/src/test/java/org/apache/qpid/pool/PoolingFilterTest.java
@@ -36,25 +36,32 @@ public class PoolingFilterTest extends TestCase
 
     public void setUp()
     {
+
         //Create Pool
         _executorService = ReferenceCountingExecutorService.getInstance();
         _executorService.acquireExecutorService();
-        _pool = PoolingFilter.createAynschWritePoolingFilter(_executorService, 
+        _pool = PoolingFilter.createAynschWritePoolingFilter(_executorService,
                                   "AsynchronousWriteFilter");
 
     }
 
     public void testRejectedExecution() throws Exception
     {
-        _pool.filterWrite(new NoOpFilter(), new TestSession(), new IoFilter.WriteRequest("Message"));
+
+        TestSession testSession = new TestSession();
+        _pool.createNewJobForSession(testSession);
+        _pool.filterWrite(new NoOpFilter(), testSession, new IoFilter.WriteRequest("Message"));
 
         //Shutdown the pool
         _executorService.getPool().shutdownNow();
 
         try
         {
+
+            testSession = new TestSession();
+            _pool.createNewJobForSession(testSession);
             //prior to fix for QPID-172 this would throw RejectedExecutionException
-            _pool.filterWrite(null, new TestSession(), null);
+            _pool.filterWrite(null, testSession, null);
         }
         catch (RejectedExecutionException rje)
         {
diff --git a/java/common/src/test/java/org/apache/qpid/session/TestSession.java b/java/common/src/test/java/org/apache/qpid/session/TestSession.java
index f10d55e9d0..aafc91b03b 100644
--- a/java/common/src/test/java/org/apache/qpid/session/TestSession.java
+++ b/java/common/src/test/java/org/apache/qpid/session/TestSession.java
@@ -24,9 +24,13 @@ import org.apache.mina.common.*;
 
 import java.net.SocketAddress;
 import java.util.Set;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ConcurrentHashMap;
 
 public class TestSession implements IoSession
 {
+    private final ConcurrentMap attributes = new ConcurrentHashMap();
+
     public TestSession()
     {
     }
@@ -68,42 +72,42 @@ public class TestSession implements IoSession
 
     public Object getAttachment()
     {
-        return null;  //TODO
+        return getAttribute("");
     }
 
     public Object setAttachment(Object attachment)
     {
-        return null;  //TODO
+        return setAttribute("",attachment);
     }
 
     public Object getAttribute(String key)
     {
-        return null;  //TODO
+        return attributes.get(key);
     }
 
     public Object setAttribute(String key, Object value)
     {
-        return null;  //TODO
+        return attributes.put(key,value);
     }
 
     public Object setAttribute(String key)
     {
-        return null;  //TODO
+        return attributes.put(key, Boolean.TRUE);
     }
 
     public Object removeAttribute(String key)
     {
-        return null;  //TODO
+        return attributes.remove(key);
     }
 
     public boolean containsAttribute(String key)
     {
-        return false;  //TODO
+        return attributes.containsKey(key);
     }
 
     public Set getAttributeKeys()
     {
-        return null;  //TODO
+        return attributes.keySet();
     }
 
     public TransportType getTransportType()