1 files changed, 71 insertions, 7 deletions

diff --git a/java/systests/src/main/java/org/apache/qpid/server/persistent/NoLocalAfterRecoveryTest.java b/java/systests/src/main/java/org/apache/qpid/server/persistent/NoLocalAfterRecoveryTest.java
index 2e259191aa..bcad59a1fa 100644
--- a/java/systests/src/main/java/org/apache/qpid/server/persistent/NoLocalAfterRecoveryTest.java
+++ b/java/systests/src/main/java/org/apache/qpid/server/persistent/NoLocalAfterRecoveryTest.java
@@ -42,7 +42,7 @@ public class NoLocalAfterRecoveryTest extends QpidBrokerTestCase
     protected final String MY_TOPIC_SUBSCRIPTION_NAME = this.getName();
     protected static final int SEND_COUNT = 10;
 
-    public void test() throws Exception
+    public void testNoLocalNotQueued() throws Exception
     {
         if(!isBrokerStorePersistent())
         {
@@ -66,6 +66,8 @@ public class NoLocalAfterRecoveryTest extends QpidBrokerTestCase
         // Check messages can be received as expected.
         connection.start();
 
+        //As the no-local subscriber was on the same connection the messages were
+        //published on, tit will receive no messages as they will be discarded on the broker
         List<Message> received = receiveMessage(noLocalSubscriber, SEND_COUNT);
         assertEquals("No Local Subscriber Received messages", 0, received.size());
 
@@ -73,10 +75,12 @@ public class NoLocalAfterRecoveryTest extends QpidBrokerTestCase
         assertEquals("Normal Subscriber Received no messages",
                      SEND_COUNT, received.size());
         session.commit();
+
+        normalSubscriber.close();
         connection.close();
 
-        //We didn't receive the messages on the durable queue for the no-local subscriber
-        //so they are still on the broker. Restart the broker, prompting their recovery.
+        //Ensure the no-local subscribers messages were discarded by restarting the broker
+        //and reconnecting to the subscription to ensure they were not recovered.
         restartBroker();
 
         Connection connection2 = getConnection();
@@ -89,12 +93,72 @@ public class NoLocalAfterRecoveryTest extends QpidBrokerTestCase
                 createDurableSubscriber(topic2, MY_TOPIC_SUBSCRIPTION_NAME + "-NoLocal",
                                         null, true);
 
-        // The NO-local subscriber should now get ALL the messages
-        // as they are being consumed on a different connection to
-        // the one that they were published on.
+        // The NO-local subscriber should not get any messages
+        received = receiveMessage(noLocalSubscriber2, SEND_COUNT);
+        session2.commit();
+        assertEquals("No Local Subscriber Received messages", 0, received.size());
+
+        noLocalSubscriber2.close();
+
+
+    }
+
+
+    public void testNonNoLocalQueued() throws Exception
+    {
+        if(!isBrokerStorePersistent())
+        {
+            fail("This test requires a broker with a persistent store");
+        }
+
+        Connection connection = getConnection();
+        Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
+        Topic topic = session.createTopic(MY_TOPIC_SUBSCRIPTION_NAME);
+
+        TopicSubscriber noLocalSubscriber =
+                session.createDurableSubscriber(topic, MY_TOPIC_SUBSCRIPTION_NAME + "-NoLocal", null, true);
+
+
+        sendMessage(session, topic, SEND_COUNT);
+
+        // Check messages can be received as expected.
+        connection.start();
+
+        List<Message> received = receiveMessage(noLocalSubscriber, SEND_COUNT);
+        assertEquals("No Local Subscriber Received messages", 0, received.size());
+
+
+
+        session.commit();
+
+        Connection connection3 = getConnection();
+        Session session3 = connection3.createSession(true, Session.SESSION_TRANSACTED);
+        sendMessage(session3, topic, SEND_COUNT);
+
+
+        connection.close();
+
+        //We didn't receive the messages on the durable queue for the no-local subscriber
+        //so they are still on the broker. Restart the broker, prompting their recovery.
+        restartBroker();
+
+        Connection connection2 = getConnection();
+        connection2.start();
+
+        Session session2 = connection2.createSession(true, Session.SESSION_TRANSACTED);
+        Topic topic2 = session2.createTopic(MY_TOPIC_SUBSCRIPTION_NAME);
+
+        TopicSubscriber noLocalSubscriber2 =
+                session2.createDurableSubscriber(topic2, MY_TOPIC_SUBSCRIPTION_NAME + "-NoLocal",null, true);
+
+        // The NO-local subscriber should receive messages sent from connection3
         received = receiveMessage(noLocalSubscriber2, SEND_COUNT);
         session2.commit();
-        assertEquals("No Local Subscriber Received messages", SEND_COUNT, received.size());
+        assertEquals("No Local Subscriber did not receive expected messages", SEND_COUNT, received.size());
+
+        noLocalSubscriber2.close();
+
+
     }
 
     protected List<Message> receiveMessage(MessageConsumer messageConsumer,