Fix and re-enable hanging test

acogoluegnes · acogoluegnes · commit 7b6b7af20f41 · 2019-05-15T10:48:16.000+02:00
The test could hang because a connection was in flow control and a queue
purge operation was stuck. The test now uses different connections to
publish and purge and use publisher confirms. Publisher confirms seems
necessary because the test was expecting the exact number of published
messages was purged from the queue. It's maybe expecting too much
considering publishing is asynchronous. Publisher confirms should make
the test more reliable.
diff --git a/src/test/java/com/rabbitmq/client/test/server/EffectVisibilityCrossNodeTest.java b/src/test/java/com/rabbitmq/client/test/server/EffectVisibilityCrossNodeTest.java
@@ -18,7 +18,11 @@
 import static org.junit.Assert.assertEquals;
 
 import java.io.IOException;
+import java.util.concurrent.TimeoutException;
 
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.test.TestUtils;
 import org.junit.Test;
 
 import com.rabbitmq.client.test.functional.ClusteredTestBase;
@@ -29,19 +33,25 @@
  */
 public class EffectVisibilityCrossNodeTest extends ClusteredTestBase {
     private final String[] queues = new String[QUEUES];
+    private Connection purgeConnection;
 
     @Override
-    protected void createResources() throws IOException {
+    protected void createResources() throws IOException, TimeoutException {
         for (int i = 0; i < queues.length ; i++) {
             queues[i] = alternateChannel.queueDeclare("", false, false, true, null).getQueue();
             alternateChannel.queueBind(queues[i], "amq.fanout", "");
         }
+        this.purgeConnection = TestUtils.connectionFactory().newConnection();
     }
 
     @Override
     protected void releaseResources() throws IOException {
-        for (int i = 0; i < queues.length ; i++) {
-            alternateChannel.queueDelete(queues[i]);
+        try {
+            for (int i = 0; i < queues.length ; i++) {
+                alternateChannel.queueDelete(queues[i]);
+            }
+        } finally {
+            TestUtils.close(this.purgeConnection);
         }
     }
 
@@ -52,13 +62,15 @@ protected void releaseResources() throws IOException {
     private static final byte[] msg = "".getBytes();
 
     @Test public void effectVisibility() throws Exception {
-
+        Channel purgeChannel = this.purgeConnection.createChannel();
+        channel.confirmSelect();
         for (int i = 0; i < BATCHES; i++) {
             for (int j = 0; j < MESSAGES_PER_BATCH; j++) {
                 channel.basicPublish("amq.fanout", "", null, msg);
             }
+            channel.waitForConfirmsOrDie(10_000);
             for (int j = 0; j < queues.length ; j++) {
-                assertEquals(MESSAGES_PER_BATCH, channel.queuePurge(queues[j]).getMessageCount());
+                assertEquals(MESSAGES_PER_BATCH, purgeChannel.queuePurge(queues[j]).getMessageCount());
             }
         }
     }
diff --git a/src/test/java/com/rabbitmq/client/test/server/ServerTests.java b/src/test/java/com/rabbitmq/client/test/server/ServerTests.java
@@ -26,7 +26,7 @@
 	Permissions.class,
     DurableBindingLifecycle.class,
     DeadLetterExchangeDurable.class,
-    //EffectVisibilityCrossNodeTest.class,
+    EffectVisibilityCrossNodeTest.class,
     ExclusiveQueueDurability.class,
     AbsentQueue.class,
     AlternateExchangeEquivalence.class,
