QPID-1206: Fix failover and failover tests

AMQConnection: remove dead and confusingly misnamed method

AMQSession: rename failedOver to failedOverDirty to convey actual usage, only set it if we failed over while dirty. Ewww!

BasicMessageConsumer: if we're in client ack mode, mark as dirty when we receive a message

PingPongProducer: calculate expected replies properly if we fail after a send or before a commit

FailoverTest: test transacted case


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

3 files changed, 13 insertions, 36 deletions

diff --git a/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java b/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java
index 90dce8eb60..27294562e5 100644
--- a/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java
+++ b/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java
@@ -1322,24 +1322,6 @@ public class AMQConnection extends Closeable implements Connection, QueueConnect
         _sessions.remove(channelId);
     }
 
-    /**
-     * For all sessions, and for all consumers in those sessions, resubscribe. This is called during failover handling.
-     * The caller must hold the failover mutex before calling this method.
-     */
-    public void resubscribeSesssions() throws JMSException, AMQException, FailoverException
-    {
-        ArrayList sessions = new ArrayList(_sessions.values());
-        _logger.info(MessageFormat.format("Resubscribing sessions = {0} sessions.size={1}", sessions, sessions.size())); // FIXME: removeKey?
-        for (Iterator it = sessions.iterator(); it.hasNext();)
-        {
-            AMQSession s = (AMQSession) it.next();
-            // _protocolHandler.addSessionByChannel(s.getChannelId(), s);
-            reopenChannel(s.getChannelId(), s.getDefaultPrefetchHigh(), s.getDefaultPrefetchLow(), s.getTransacted());
-            s.resubscribe();
-            s.setFlowControl(true);
-        }
-    }
-
     public String toString()
     {
         StringBuffer buf = new StringBuffer("AMQConnection:\n");
diff --git a/java/client/src/main/java/org/apache/qpid/client/AMQSession.java b/java/client/src/main/java/org/apache/qpid/client/AMQSession.java
index 801cf53d83..c253b8b641 100644
--- a/java/client/src/main/java/org/apache/qpid/client/AMQSession.java
+++ b/java/client/src/main/java/org/apache/qpid/client/AMQSession.java
@@ -368,8 +368,8 @@ public abstract class AMQSession extends Closeable implements Session, QueueSess
 
     /** Session state : used to detect if commit is a) required b) allowed , i.e. does the tx span failover. */
     private boolean _dirty;
-    /** Has failover occured on this session */
-    private boolean _failedOver;
+    /** Has failover occured on this session with outstanding actions to commit? */
+    private boolean _failedOverDirty;
 
     private static final class FlowControlIndicator
     {
@@ -740,6 +740,7 @@ public abstract class AMQSession extends Closeable implements Session, QueueSess
             }
             // Commits outstanding messages and acknowledgments
             sendCommit();
+            markClean();
         }
         catch (AMQException e)
         {
@@ -1796,7 +1797,10 @@ public abstract class AMQSession extends Closeable implements Session, QueueSess
      */
     void resubscribe() throws AMQException
     {
-        _failedOver = true;
+        if (_dirty)
+        {
+            _failedOverDirty = true;
+        }
         resubscribeProducers();
         resubscribeConsumers();
     }
@@ -2586,7 +2590,7 @@ public abstract class AMQSession extends Closeable implements Session, QueueSess
     public void markClean()
     {
         _dirty = false;
-        _failedOver = false;
+        _failedOverDirty = false;
     }
 
     /**
@@ -2596,7 +2600,7 @@ public abstract class AMQSession extends Closeable implements Session, QueueSess
      */
     public boolean hasFailedOver()
     {
-        return _failedOver;
+        return _failedOverDirty;
     }
 
     /**
diff --git a/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java b/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java
index 9acc69ec55..aaf56e8493 100644
--- a/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java
+++ b/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java
@@ -730,18 +730,9 @@ public abstract class BasicMessageConsumer<H, B> extends Closeable implements Me
         {
             if (isMessageListenerSet())
             {
-                // we do not need a lock around the test above, and the dispatch below as it is invalid
-                // for an application to alter an installed listener while the session is started
-                // synchronized (_closed)
-                {
-                    // if (!_closed.get())
-                    {
-
-                        preApplicationProcessing(jmsMessage);
-                        getMessageListener().onMessage(jmsMessage);
-                        postDeliver(jmsMessage);
-                    }
-                }
+                preApplicationProcessing(jmsMessage);
+                getMessageListener().onMessage(jmsMessage);
+                postDeliver(jmsMessage);
             }
             else
             {
@@ -802,7 +793,7 @@ public abstract class BasicMessageConsumer<H, B> extends Closeable implements Me
                 {
                     _session.acknowledgeMessage(msg.getDeliveryTag(), false);
                 }
-
+                _session.markDirty();
                 break;
 
             case Session.DUPS_OK_ACKNOWLEDGE: