QPID-4503: Producer transaction timeout detection feature may produce suprious open/idle alerts and close client connections/sessions without good cause

Race conditon existed between the initial check that determined transaction timeout detection was required and the subsequent re-observeration
required to perform the open and idle calculation.  In the unlucky timing, the state of the transaction changed between these two points.  (To produce
the time-since-epoch type the transaction needs to be committed between these two points).

Changed checkTransactionStatus so that transactionStartTime and transactionUpdateTime are observed once only.

There still exists the possibility that the transaction status change change between the reading of transactionStartTime and transactionUpdateTime
times, but I do not see how this could produce a suprious report (or close).  The alternative (to take a locks to ensure consistent observations
are made) seems unjustifiably costly.

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

3 files changed, 13 insertions, 20 deletions

diff --git a/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java b/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
index 4367d7ee53..76a3a7f224 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
@@ -70,7 +70,6 @@ import org.apache.qpid.server.message.MessageReference;
 import org.apache.qpid.server.message.ServerMessage;
 import org.apache.qpid.server.output.ProtocolOutputConverter;
 import org.apache.qpid.server.protocol.AMQConnectionModel;
-import org.apache.qpid.server.protocol.AMQProtocolEngine;
 import org.apache.qpid.server.protocol.AMQProtocolSession;
 import org.apache.qpid.server.protocol.AMQSessionModel;
 import org.apache.qpid.server.queue.AMQQueue;
@@ -113,7 +112,7 @@ public class AMQChannel implements AMQSessionModel, AsyncAutoCommitTransaction.F
      */
     private long _deliveryTag = 0;
 
-    /** A channel has a default queue (the last declared) that is used when no queue name is explictily set */
+    /** A channel has a default queue (the last declared) that is used when no queue name is explicitly set */
     private AMQQueue _defaultQueue;
 
     /** This tag is unique per subscription to a queue. The server returns this in response to a basic.consume request. */
@@ -209,10 +208,6 @@ public class AMQChannel implements AMQSessionModel, AsyncAutoCommitTransaction.F
     }
 
 
-    public boolean inTransaction()
-    {
-        return isTransactional() && _txnUpdateTime.get() > 0 && _transaction.getTransactionStartTime() > 0;
-    }
 
     private void incrementOutstandingTxnsIfNecessary()
     {
@@ -1487,11 +1482,13 @@ public class AMQChannel implements AMQSessionModel, AsyncAutoCommitTransaction.F
 
     public void checkTransactionStatus(long openWarn, long openClose, long idleWarn, long idleClose) throws AMQException
     {
-        if (inTransaction())
+        final long transactionStartTime = _transaction.getTransactionStartTime();
+        final long transactionUpdateTime = _txnUpdateTime.get();
+        if (isTransactional() && transactionUpdateTime > 0 && transactionStartTime > 0)
         {
             long currentTime = System.currentTimeMillis();
-            long openTime = currentTime - _transaction.getTransactionStartTime();
-            long idleTime = currentTime - _txnUpdateTime.get();
+            long openTime = currentTime - transactionStartTime;
+            long idleTime = currentTime - transactionUpdateTime;
 
             _transactionTimeoutHelper.logIfNecessary(idleTime, idleWarn, ChannelMessages.IDLE_TXN(idleTime),
                                                      TransactionTimeoutHelper.IDLE_TRANSACTION_ALERT);
diff --git a/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSession.java b/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSession.java
index eed55a2e85..075ed2a87c 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSession.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSession.java
@@ -42,7 +42,6 @@ import javax.security.auth.Subject;
 import org.apache.qpid.AMQException;
 import org.apache.qpid.AMQStoreException;
 import org.apache.qpid.protocol.AMQConstant;
-import org.apache.qpid.protocol.ProtocolEngine;
 import org.apache.qpid.server.TransactionTimeoutHelper;
 import org.apache.qpid.server.logging.LogActor;
 import org.apache.qpid.server.logging.LogSubject;
@@ -449,11 +448,6 @@ public class ServerSession extends Session
         return _transaction.isTransactional();
     }
 
-    public boolean inTransaction()
-    {
-        return isTransactional() && _txnUpdateTime.get() > 0 && _transaction.getTransactionStartTime() > 0;
-    }
-
     public void selectTx()
     {
         _transaction = new LocalTransaction(this.getMessageStore());
@@ -591,7 +585,7 @@ public class ServerSession extends Session
     /**
      * Update last transaction activity timestamp
      */
-    public void updateTransactionalActivity()
+    private void updateTransactionalActivity()
     {
         if (isTransactional())
         {
@@ -709,11 +703,13 @@ public class ServerSession extends Session
 
     public void checkTransactionStatus(long openWarn, long openClose, long idleWarn, long idleClose) throws AMQException
     {
-        if (inTransaction())
+        final long transactionStartTime = _transaction.getTransactionStartTime();
+        final long transactionUpdateTime = _txnUpdateTime.get();
+        if (isTransactional() && transactionUpdateTime > 0 && transactionStartTime > 0)
         {
             long currentTime = System.currentTimeMillis();
-            long openTime = currentTime - _transaction.getTransactionStartTime();
-            long idleTime = currentTime - _txnUpdateTime.get();
+            long openTime = currentTime - transactionStartTime;
+            long idleTime = currentTime - transactionUpdateTime;
 
             _transactionTimeoutHelper.logIfNecessary(idleTime, idleWarn, ChannelMessages.IDLE_TXN(idleTime),
                                                      TransactionTimeoutHelper.IDLE_TRANSACTION_ALERT);
diff --git a/java/broker/src/main/java/org/apache/qpid/server/txn/LocalTransaction.java b/java/broker/src/main/java/org/apache/qpid/server/txn/LocalTransaction.java
index 3fbcff7e2c..f11fb1086e 100755
--- a/java/broker/src/main/java/org/apache/qpid/server/txn/LocalTransaction.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/txn/LocalTransaction.java
@@ -50,7 +50,7 @@ public class LocalTransaction implements ServerTransaction
 
     private volatile Transaction _transaction;
     private MessageStore _transactionLog;
-    private long _txnStartTime = 0L;
+    private volatile long _txnStartTime = 0L;
     private StoreFuture _asyncTran;
 
     public LocalTransaction(MessageStore transactionLog)