QPID-4091 : [Java Broker] correct handling of dtx set timeout

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

2 files changed, 18 insertions, 6 deletions

diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/DtxBranch.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/DtxBranch.java
index 3ac71fc6a6..900e2ef222 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/DtxBranch.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/DtxBranch.java
@@ -103,7 +103,7 @@ public class DtxBranch
             _timeoutFuture.cancel(false);
         }
         _timeout = timeout;
-        _expiration = timeout == 0 ? 0 : System.currentTimeMillis() + timeout;
+        _expiration = timeout == 0 ? 0 : System.currentTimeMillis() + (1000 * timeout);
 
         if(_timeout == 0)
         {
@@ -111,7 +111,7 @@ public class DtxBranch
         }
         else
         {
-            _timeoutFuture = _vhost.scheduleTask(_timeout, new Runnable()
+            _timeoutFuture = _vhost.scheduleTask(1000*_timeout, new Runnable()
             {
                 public void run()
                 {
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/DtxRegistry.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/DtxRegistry.java
index e3bc076d72..6a91cc39c5 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/DtxRegistry.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/txn/DtxRegistry.java
@@ -37,7 +37,7 @@ public class DtxRegistry
     private static final class ComparableXid
     {
         private final Xid _xid;
-        
+
         private ComparableXid(Xid xid)
         {
             _xid = xid;
@@ -58,7 +58,7 @@ public class DtxRegistry
             ComparableXid that = (ComparableXid) o;
 
             return compareBytes(_xid.getBranchId(), that._xid.getBranchId())
-                    && compareBytes(_xid.getGlobalId(), that._xid.getGlobalId()); 
+                    && compareBytes(_xid.getGlobalId(), that._xid.getGlobalId());
         }
 
         private static boolean compareBytes(byte[] a, byte[] b)
@@ -94,7 +94,7 @@ public class DtxRegistry
             return result;
         }
     }
-    
+
     public synchronized DtxBranch getBranch(Xid xid)
     {
         return _branches.get(new ComparableXid(xid));
@@ -308,16 +308,28 @@ public class DtxRegistry
         return inDoubt;
     }
 
-    public synchronized void endAssociations(AMQSessionModel session)
+    public synchronized void endAssociations(AMQSessionModel session) throws TimeoutDtxException
     {
+        boolean timeout = false;
+        Xid id = null;
         for(DtxBranch branch : _branches.values())
         {
             if(branch.isAssociated(session))
             {
+                if(branch.expired() || branch.getState() == DtxBranch.State.TIMEDOUT)
+                {
+                    timeout = true;
+                    id = branch.getXid();
+                }
                 branch.setState(DtxBranch.State.ROLLBACK_ONLY);
                 branch.disassociateSession(session);
             }
         }
+
+        if(timeout)
+        {
+            throw new TimeoutDtxException(id);
+        }
     }