fixed a race condition in rollback() that leads to intermittant failures of TransactedTest, also modified TransactedTest to be slightly more robust

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

1 files changed, 17 insertions, 7 deletions

diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java
index af469ee291..bcd9337bcc 100644
--- a/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java
+++ b/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java
@@ -109,6 +109,7 @@ import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
 
 /**
  *
@@ -221,6 +222,11 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi
      */
     private final FlowControllingBlockingQueue _queue;
 
+    /**
+     * Holds the highest received delivery tag.
+     */
+    private final AtomicLong _highestDeliveryTag = new AtomicLong(-1);
+
     /** Holds the dispatcher thread for this session. */
     private Dispatcher _dispatcher;
 
@@ -1281,6 +1287,7 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi
         }
         else
         {
+            _highestDeliveryTag.set(message.getDeliverBody().deliveryTag);
             _queue.add(message);
         }
     }
@@ -2553,6 +2560,7 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi
         private final AtomicBoolean _closed = new AtomicBoolean(false);
 
         private final Object _lock = new Object();
+        private final AtomicLong _rollbackMark = new AtomicLong(-1);
 
         public Dispatcher()
         {
@@ -2609,7 +2617,7 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi
                     setConnectionStopped(true);
                 }
 
-                rejectAllMessages(true);
+                _rollbackMark.set(_highestDeliveryTag.get());
 
                 _dispatcherLogger.debug("Session Pre Dispatch Queue cleared");
 
@@ -2645,7 +2653,7 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi
             // Allow disptacher to start stopped
             synchronized (_lock)
             {
-                while (connectionStopped())
+                while (!_closed.get() && connectionStopped())
                 {
                     try
                     {
@@ -2670,14 +2678,16 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi
                             _lock.wait();
                         }
 
-                        synchronized (_messageDeliveryLock)
+                        if (message.getDeliverBody().deliveryTag <= _rollbackMark.get())
                         {
-                            dispatchMessage(message);
+                            rejectMessage(message, true);
                         }
-
-                        while (connectionStopped())
+                        else
                         {
-                            _lock.wait();
+                            synchronized (_messageDeliveryLock)
+                            {
+                                dispatchMessage(message);
+                            }
                         }
 
                     }