Fix for dirty sessions, start to test that sessions are dirty when required.

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

2 files changed, 90 insertions, 21 deletions

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 2dfecc80ac..667785c441 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
@@ -779,6 +779,7 @@ public abstract class BasicMessageConsumer<U> extends Closeable implements Messa
                 else
                 {
                     _session.addDeliveredMessage(msg.getDeliveryTag());
+                    _session.markDirty();
                 }
 
                 break;
diff --git a/java/systests/src/main/java/org/apache/qpid/test/unit/ack/AcknowledgeAfterFailoverTest.java b/java/systests/src/main/java/org/apache/qpid/test/unit/ack/AcknowledgeAfterFailoverTest.java
index cad54131f7..be268946cd 100644
--- a/java/systests/src/main/java/org/apache/qpid/test/unit/ack/AcknowledgeAfterFailoverTest.java
+++ b/java/systests/src/main/java/org/apache/qpid/test/unit/ack/AcknowledgeAfterFailoverTest.java
@@ -28,6 +28,9 @@ import javax.jms.JMSException;
 import javax.jms.Message;
 import javax.jms.Session;
 
+/**
+ *
+ */
 public class AcknowledgeAfterFailoverTest extends AcknowledgeTest
 {
 
@@ -35,15 +38,24 @@ public class AcknowledgeAfterFailoverTest extends AcknowledgeTest
     public void setUp() throws Exception
     {
         super.setUp();
+        // This must be even for the test to run correctly.
+        // Otherwise we will kill the standby broker
+        // not the one we are connected to.
+        // The test will still pass but it will not be exactly
+        // as described.
         NUM_MESSAGES = 10;
     }
 
     protected void prepBroker(int count) throws Exception
     {
-        //Stop the connection whilst we repopulate the broker, or the no_ack
-        // test will drain the msgs before we can check we put the right number
-        // back on again.
-//        _connection.stop();
+        if (count % 2 == 1)
+        {
+            failBroker(getFailingPort());
+        }
+        else
+        {
+            failBroker(getPort());
+        }
 
         Connection connection = getConnection();
         Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
@@ -60,7 +72,21 @@ public class AcknowledgeAfterFailoverTest extends AcknowledgeTest
 
         connection.close();
 
-//        _connection.start();
+        try
+        {
+            if (count % 2 == 1)
+            {
+                startBroker(getFailingPort());
+            }
+            else
+            {
+                startBroker(getPort());
+            }
+        }
+        catch (Exception e)
+        {
+            fail("Unable to start failover broker," + e.getMessage());
+        }
     }
 
     @Override
@@ -69,41 +95,83 @@ public class AcknowledgeAfterFailoverTest extends AcknowledgeTest
         //Acknowledge current message
         super.doAcknowlegement(msg);
 
-        int msgCount = msg.getIntProperty(INDEX);
-
-        if (msgCount % 2 == 0)
+        try
         {
-            failBroker(getFailingPort());
+            prepBroker(NUM_MESSAGES - msg.getIntProperty(INDEX) - 1);
         }
-        else
+        catch (Exception e)
         {
-            failBroker(getPort());
+            fail("Unable to prep new broker," + e.getMessage());
         }
 
+    }
+
+    /**
+     * @param transacted
+     * @param mode
+     *
+     * @throws Exception
+     */
+    protected void testDirtyAcking(boolean transacted, int mode) throws Exception
+    {
+        NUM_MESSAGES = 2;
+        //Test Dirty Failover Fails
+        init(transacted, mode);
+
+        _connection.start();
+
+        Message msg = _consumer.receive(1500);
+
+        int count = 0;
+        assertNotNull("Message " + count + " not correctly received.", msg);
+        assertEquals("Incorrect message received", count, msg.getIntProperty(INDEX));
+        count++;
+
+        //Don't acknowledge just prep the next broker
+
         try
         {
-            prepBroker(NUM_MESSAGES - msgCount - 1);
+            prepBroker(count);
         }
         catch (Exception e)
         {
             fail("Unable to prep new broker," + e.getMessage());
         }
 
-        try
+        // Consume the next message
+        msg = _consumer.receive(1500);
+        assertNotNull("Message " + count + " not correctly received.", msg);
+        assertEquals("Incorrect message received", count, msg.getIntProperty(INDEX));
+
+        if (_consumerSession.getTransacted())
+        {
+            _consumerSession.commit();
+        }
+        else
         {
-            if (msgCount % 2 == 0)
+            try
             {
-                startBroker(getFailingPort());
+                msg.acknowledge();
+                fail("Session is dirty we should get an IllegalStateException");
             }
-            else
+            catch (IllegalStateException ise)
             {
-                startBroker(getPort());
+                assertEquals("Incorrect Exception thrown", "has failed over", ise.getMessage());
             }
         }
-        catch (Exception e)
-        {
-            fail("Unable to start failover broker," + e.getMessage());
-        }
 
+        assertEquals("Wrong number of messages on queue", 0,
+                     ((AMQSession) _consumerSession).getQueueDepth((AMQDestination) _queue));
     }
+
+    public void testDirtyClientAck() throws Exception
+    {
+        testDirtyAcking(false, Session.CLIENT_ACKNOWLEDGE);
+    }
+
+    public void testDirtyTransacted() throws Exception
+    {
+        testDirtyAcking(true, Session.SESSION_TRANSACTED);
+    }
+
 }