Fix completion-related tests

[#154470407]

References #72

Author: acogoluegnes

src/main/java/com/rabbitmq/perf/Producer.java

@@ -64,7 +64,6 @@ public Producer(Channel channel, String exchangeName, String id, boolean randomR
                     long confirm, int confirmTimeout,
                     MessageBodySource messageBodySource,
                     TimestampProvider tsp, Stats stats, MulticastSet.CompletionHandler completionHandler) {
-
         this.channel           = channel;
         this.exchangeName      = exchangeName;
         this.id                = id;

src/test/java/com/rabbitmq/perf/MessageCountTimeLimitTest.java

@@ -169,7 +169,7 @@ public void producerCount(int producersCount, int channelsCount) throws Exceptio
             any(), any());
 
         assertThat(messagesTotal + " messages should have been published by now",
-            publishedLatch.await(5, TimeUnit.SECONDS), is(true));
+            publishedLatch.await(10, TimeUnit.SECONDS), is(true));
         waitAtMost(5, TimeUnit.SECONDS).until(() -> runningTest.isDone(), is(true));
         verify(ch, times(messagesTotal))
             .basicPublish(anyString(), anyString(),

src/test/java/com/rabbitmq/perf/TopologyTest.java

@@ -471,7 +471,7 @@ public void sequenceMoreQueuesThanProducers() throws Exception {
             anyBoolean(), eq(false),
             any(), any(byte[].class));
 
-        MulticastSet set = getMulticastSet(new InterruptThreadHandler(latchPublishing));
+        MulticastSet set = getMulticastSet(new MulticastSet.DefaultThreadingHandler(), latchPublishing);
 
         set.run();
 
@@ -617,14 +617,26 @@ private MulticastSet getMulticastSet() {
     private MulticastSet getMulticastSet(MulticastSet.ThreadingHandler threadingHandler) {
         MulticastSet set = new MulticastSet(
             stats, cf, params, singletonList("amqp://localhost"), new MulticastSet.CompletionHandler() {
-
             @Override
-            public void waitForCompletion() {
-            }
+            public void waitForCompletion() { }
+            @Override
+            public void countDown() { }
+        }
+        );
+
+        set.setThreadingHandler(threadingHandler);
+        return set;
+    }
 
+    private MulticastSet getMulticastSet(MulticastSet.ThreadingHandler threadingHandler, CountDownLatch completionLatch) {
+        MulticastSet set = new MulticastSet(
+            stats, cf, params, singletonList("amqp://localhost"), new MulticastSet.CompletionHandler() {
             @Override
-            public void countDown() {
+            public void waitForCompletion() throws InterruptedException {
+                completionLatch.await(10, TimeUnit.SECONDS);
             }
+            @Override
+            public void countDown() { }
         }
         );