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

3 files changed, 34 insertions, 21 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);
+        }
     }
 
 
diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/xa/FaultTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/xa/FaultTest.java
index e940a73bbb..2ba509e96a 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/xa/FaultTest.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/xa/FaultTest.java
@@ -1,6 +1,6 @@
 package org.apache.qpid.test.unit.xa;
 /*
- * 
+ *
  * 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
@@ -8,16 +8,16 @@ package org.apache.qpid.test.unit.xa;
  * 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.
- * 
+ *
  */
 
 
@@ -344,7 +344,7 @@ public class FaultTest extends AbstractXATestCase
         {
             assertEquals("Wrong error code: ", XAException.XAER_PROTO, e.errorCode);
         }
-    }    
+    }
 
     /**
      * Strategy:
@@ -366,27 +366,28 @@ public class FaultTest extends AbstractXATestCase
     /**
      * Strategy:
      * Check that a transaction timeout as expected
-     * - set timeout to 10ms
-     * - sleep 1000ms
+     * - set timeout to 1s
+     * - sleep 1500ms
      * - call end and check that the expected exception is thrown
      */
     public void testTransactionTimeout() throws Exception
     {
+        _xaResource.setTransactionTimeout(1);
+
         Xid xid = getNewXid();
         try
         {
             _xaResource.start(xid, XAResource.TMNOFLAGS);
-            assertEquals("Wrong timeout", _xaResource.getTransactionTimeout(), 0);
-            _xaResource.setTransactionTimeout(10);
-            Thread.sleep(1000);
+            Thread.sleep(1500);
             _xaResource.end(xid, XAResource.TMSUCCESS);
+            fail("Timeout expected ");
         }
         catch (XAException e)
         {
             assertEquals("Wrong error code: ", XAException.XA_RBTIMEOUT, e.errorCode);
         }
     }
-    
+
     /**
      * Strategy:
      * Set the transaction timeout to 1000
@@ -394,18 +395,18 @@ public class FaultTest extends AbstractXATestCase
     public void testTransactionTimeoutAfterCommit() throws Exception
     {
         Xid xid = getNewXid();
-        
+
         _xaResource.start(xid, XAResource.TMNOFLAGS);
         _xaResource.setTransactionTimeout(1000);
         assertEquals("Wrong timeout", 1000,_xaResource.getTransactionTimeout());
-        
+
         //_xaResource.prepare(xid);
         _xaResource.end(xid, XAResource.TMSUCCESS);
         _xaResource.commit(xid, true);
-        
+
         _xaResource.setTransactionTimeout(2000);
         assertEquals("Wrong timeout", 2000,_xaResource.getTransactionTimeout());
-        
+
         xid = getNewXid();
         _xaResource.start(xid, XAResource.TMNOFLAGS);
         assertEquals("Wrong timeout", 2000, _xaResource.getTransactionTimeout());