fix up virtual topic consumer migration test to ensure demand and wai…

tests/arq/src/test/java/org/jboss/test/arquillian/ce/amq/Amq63VirtualTopicSubscriberMigrationTest.java

@@ -45,14 +45,20 @@
                 @TemplateParameter(name = "MQ_QUEUES", value = "QUEUES.FOO,QUEUES.BAR"),
                 @TemplateParameter(name = "MQ_TOPICS", value = "topics.mqtt,TOPICS.FOO"),
                 @TemplateParameter(name = "APPLICATION_NAME", value = "amq-test"),
+                @TemplateParameter(name = "AMQ_MESH_QUERY_INTERVAL", value = "5"), // needs to be respected to the template
                 @TemplateParameter(name = "MQ_USERNAME", value = "${amq.username:amq-test}"),
                 @TemplateParameter(name = "MQ_PASSWORD", value = "${amq.password:redhat}"),
                 @TemplateParameter(name = "MQ_PROTOCOL", value = "openwire,amqp,mqtt,stomp")})
 @RoleBinding(roleRefName = "view", userName = "system:serviceaccount:${kubernetes.namespace}:default")
 @OpenShiftResource("classpath:testrunner-secret.json")
 @OpenShiftResource("classpath:testrunner-secret.json")
 @Replicas(1)
-@Ignore("https://github.com/jboss-openshift/ce-testsite/issues/123")
+//@Ignore("https://github.com/jboss-openshift/ce-testsite/issues/123")
+// there is an issue with being ready before being networked into the mesh - remote demand won't
+// be visible and vt fanout will drop messages for remote consumers
+// AMQ_MESH_QUERY_INTERVAL needs to be in the template to allow mesh discovery to be snappy and
+// the test should verify the presence of network bridges - or the readyness probe should
+// some sleeps in there for the moment
 public class Amq63VirtualTopicSubscriberMigrationTest extends AmqVirtualTopicSubscriberMigrationTestBase {
 
     @Deployment

tests/arq/src/test/java/org/jboss/test/arquillian/ce/amq/support/AmqClient.java

@@ -267,24 +267,20 @@ public void produceStomp(String message) throws Exception {
         }
     }
 
-    public void createVirtualTopicSubscriber(String subscriptionName) throws Exception {
+    public Connection createVirtualTopicSubscriberDemand(String subscriptionName) throws Exception {
         Connection conn = null;
-        try {
-            ConnectionFactory cf = getAMQConnectionFactory(false);
-            conn = createConnection(cf, username, password);
-            conn.setClientID("tmp123");
+        ConnectionFactory cf = getAMQConnectionFactory(false);
+        conn = createConnection(cf, username, password);
 
-            conn.start();
+        conn.start();
 
-            Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
-            Destination tFoo = session.createQueue("Consumer." + subscriptionName + ".VirtualTopic.FOO");
+        // this consumer won't ack
+        Session session = conn.createSession(false, Session.CLIENT_ACKNOWLEDGE);
+        Destination tFoo = session.createQueue("Consumer." + subscriptionName + ".VirtualTopic.FOO");
 
-            // This may need to remain open
-            MessageConsumer consumer = session.createConsumer(tFoo);
-            consumer.close();
-        } finally {
-            close(conn);
-        }
+        // This needs to remain open, hence the returned connection handle
+        MessageConsumer consumer = session.createConsumer(tFoo);
+        return conn;
     }
 
     public void produceVirtualTopic(String message) throws Exception {
@@ -321,7 +317,11 @@ public List<String> consumeVirtualTopic(int N, long timeout, String subscription
             while (N > 0) {
                 TextMessage msg = (TextMessage)consumer.receive(timeout);
                 if (msg == null) {
-                    throw new IllegalStateException("Missing " + N + " messages.");
+                    // one more try
+                    msg = (TextMessage)consumer.receive(4000);
+                }
+                if (msg == null) {
+                    throw new IllegalStateException("Missing " + N + " messages on sub: " + subscriptionName);
                 }
                 msgs.add(msg.getText());
                 N--;

tests/arq/src/test/java/org/jboss/test/arquillian/ce/amq/support/AmqNMigrationsTestBase.java

@@ -79,7 +79,7 @@ public void testMigrate(@ArquillianResource OpenShiftHandle adapter) throws Exce
             }
             Assert.assertTrue("No pod with msgs!?!", pi < pods.size()); // such pod should exist
 
-            log.info(String.format("Deleting pod: %s", pods.get(pi)));
+            log.info(String.format("[%d] Deleting pod: %s", i, pods.get(pi)));
             adapter.deletePod(pods.get(pi), -1);
 
             adapter.waitForReadyPods("amq-test-amq", REPLICAS);

tests/arq/src/test/java/org/jboss/test/arquillian/ce/amq/support/AmqVirtualTopicSubscriberMigrationTestBase.java

@@ -23,8 +23,13 @@
 
 package org.jboss.test.arquillian.ce.amq.support;
 
+import javax.jms.Connection;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Set;
 import java.util.TreeSet;
+import java.util.concurrent.TimeUnit;
+import java.util.logging.Logger;
 
 import org.arquillian.cube.openshift.api.OpenShiftHandle;
 import org.jboss.arquillian.container.test.api.RunAsClient;
@@ -39,25 +44,41 @@
 public class AmqVirtualTopicSubscriberMigrationTestBase extends AmqMigrationTestBase {
 
     private static final int N = 10;
+    private static final Logger log = Logger.getLogger(AmqVirtualTopicSubscriberMigrationTestBase.class.getName());
 
     @Test
     @RunAsClient
     @InSequence(1)
     public void testScaleUp(@ArquillianResource OpenShiftHandle adapter) throws Exception {
         adapter.scaleDeployment("amq-test-amq", 2); // scale up
         adapter.waitForReadyPods("amq-test-amq", 2);
+
+        log.info("Wait on mesh formation...");
+        TimeUnit.SECONDS.sleep(30);
     }
 
     @Test
     @InSequence(2)
     public void testSendMsgs() throws Exception {
         AmqClient client = createAmqClient("tcp://" + System.getenv("AMQ_TEST_AMQ_TCP_SERVICE_HOST") + ":61616");
+        // the consumer demand needs to outlive message production
+        List<Connection> connections = new ArrayList<>();
         for (int i = 1; i <= N; i++) {
-            client.createVirtualTopicSubscriber("Sub" + i);
+            connections.add(client.createVirtualTopicSubscriberDemand("Sub" + i));
         }
+
+        log.info("Wait on mesh propagation of demand...");
+        TimeUnit.SECONDS.sleep(30);
+
         for (int i = 1; i <= N; i++) {
             client.produceVirtualTopic("Text" + i);
         }
+
+        for (Connection c: connections) {
+            try {
+                c.close();
+            } catch (Exception ignored) {}
+        }
     }
 
     @Test
@@ -76,10 +97,11 @@ public void testSubscriberConsume() throws Exception {
         AmqClient client = createAmqClient("tcp://" + System.getenv("AMQ_TEST_AMQ_TCP_SERVICE_HOST") + ":61616");
         for (int i = 1; i <= N; i++) {
             Set<Integer> msgNumbers = new TreeSet<>();
-            for (String msg : client.consumeVirtualTopic(N, 2000, "Sub" + i)) {
+            log.info("try consume N for Sub:" + i);
+            for (String msg : client.consumeVirtualTopic(N, 10000, "Sub" + i)) {
                 msgNumbers.add(Integer.parseInt(msg.substring(4)));
             }
-            Assert.assertEquals(N, msgNumbers.size());
+            Assert.assertEquals("for sub[" + i+ "]", N, msgNumbers.size());
         }
     }
 }