QPID-2059 : CI failures were due to threading races to exhaust prefetch before delivered messages could be acked. So adjusted test to have the client dispatcher ready to deliver and relax the constraint that we validate which actor is doing the logging. The fact the state has changed correctly is the important factor.

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

1 files changed, 11 insertions, 19 deletions

diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/server/logging/SubscriptionLoggingTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/server/logging/SubscriptionLoggingTest.java
index 3da45d8976..5dd56fb0f9 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/server/logging/SubscriptionLoggingTest.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/server/logging/SubscriptionLoggingTest.java
@@ -328,12 +328,15 @@ public class SubscriptionLoggingTest extends AbstractTestLogging
         int PREFETCH = 15;
 
         //Create new session with small prefetch
-        _session = ((AMQConnection) _connection).createSession(true, Session.AUTO_ACKNOWLEDGE, PREFETCH);
+        _session = ((AMQConnection) _connection).createSession(true, Session.SESSION_TRANSACTED, PREFETCH);
 
         MessageConsumer consumer = _session.createConsumer(_queue);
 
         _connection.start();
 
+        //Start the dispatcher & Unflow the channel.
+        consumer.receiveNoWait();
+
         //Fill the prefetch and two extra so that our receive bellow allows the
         // subscription to become active
         // Previously we set this to 17 so that it would return to a suspended
@@ -371,15 +374,13 @@ public class SubscriptionLoggingTest extends AbstractTestLogging
         try
         {
             // Validation expects three messages.
-            // The first will be logged by the QueueActor as part of the processQueue thread
-// INFO - MESSAGE [vh(/test)/qu(example.queue)] [sub:6(qu(example.queue))] SUB-1003 : State : SUSPENDED
-            // The second will be by the connnection as it acknowledges and activates the subscription
-// INFO - MESSAGE [con:6(guest@anonymous(26562441)/test)/ch:3] [sub:6(qu(example.queue))] SUB-1003 : State : ACTIVE
-            // The final one can be either the connection or the subscription suspending as part of the SubFlushRunner or the processQueue thread
-            // As a result validating the actor is more complicated and doesn't add anything. The goal of this test is
-            // to ensure the State is correct not that a particular Actor performs the logging.
-// INFO - MESSAGE [sub:6(vh(test)/qu(example.queue))] [sub:6(qu(example.queue))] SUB-1003 : State : SUSPENDED
-// INFO - MESSAGE [vh(/test)/qu(example.queue)] [sub:6(qu(example.queue))] SUB-1003 : State : SUSPENDED
+            // The Actor can be any one of the following depending on the exactly what is going on on the broker.
+            // Ideally we would test that we can get all of them but setting up
+            // the timing to do this in a consistent way is not benefitial.
+            // Ensuring the State is as expected is sufficient.
+// INFO - MESSAGE [vh(/test)/qu(example.queue)] [sub:6(qu(example.queue))] SUB-1003 : State :
+// INFO - MESSAGE [con:6(guest@anonymous(26562441)/test)/ch:3] [sub:6(qu(example.queue))] SUB-1003 : State :
+// INFO - MESSAGE [sub:6(vh(test)/qu(example.queue))] [sub:6(qu(example.queue))] SUB-1003 : State :
 
             assertEquals("Result set not expected size:", 3, results.size());
 
@@ -388,19 +389,10 @@ public class SubscriptionLoggingTest extends AbstractTestLogging
             String log = getLog(results.get(0));
             validateSubscriptionState(log, expectedState);
 
-            // Validate that the logActor is the the queue
-            String actor = fromActor(log);
-            assertTrue("Actor string does not contain expected queue("
-                       + _queue.getQueueName() + ") name." + actor,
-                       actor.contains("qu(" + _queue.getQueueName() + ")"));
-
             // After being suspended the subscription should become active.
             expectedState = "ACTIVE";
             log = getLog(results.get(1));
             validateSubscriptionState(log, expectedState);
-            // Validate we have a connection Actor
-            actor = fromActor(log);
-            assertTrue("The actor is not a connection actor:" + actor, actor.startsWith("con:"));
 
             // Validate that it was re-suspended
             expectedState = "SUSPENDED";