Using a predeclared exchange that does not exist creates it
Running the following command creates an exchange:

```
make run ARGS="--producers 1 --rate 1 --consumers 0 --exchange idonotexist --predeclared true"
```

Since this exchange does not exist, it should not be created since we use the `--predeclared true` flag.
diff --git a/src/main/java/com/rabbitmq/perf/Consumer.java b/src/main/java/com/rabbitmq/perf/Consumer.java
index 38e5aba..014cded 100644
--- a/src/main/java/com/rabbitmq/perf/Consumer.java
+++ b/src/main/java/com/rabbitmq/perf/Consumer.java
@@ -17,6 +17,10 @@ package com.rabbitmq.perf;
 
 import com.rabbitmq.client.AMQP.BasicProperties;
 import com.rabbitmq.client.*;
+import com.rabbitmq.client.AMQP.Queue.DeclareOk;
+import com.rabbitmq.perf.PerfTest.EXIT_WHEN;
+import java.time.Duration;
+import java.util.stream.Collectors;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -41,6 +45,8 @@ public class Consumer extends AgentBase implements Runnable {
     private static final AckNackOperation NACK_OPERATION =
             (ch, envelope, multiple, requeue) -> ch.basicNack(envelope.getDeliveryTag(), multiple, requeue);
     static final String STOP_REASON_CONSUMER_REACHED_MESSAGE_LIMIT = "Consumer reached message limit";
+    static final String STOP_REASON_CONSUMER_IDLE = "Consumer is idle for more than 1 second";
+    static final String STOP_REASON_CONSUMER_QUEUE_EMPTY = "Consumer queue(s) empty";
 
     private volatile ConsumerImpl       q;
     private final Channel               channel;
@@ -76,6 +82,8 @@ public class Consumer extends AgentBase implements Runnable {
 
     private final Map<String, Object> consumerArguments;
 
+    private final EXIT_WHEN exitWhen;
+
     public Consumer(ConsumerParameters parameters) {
         this.channel           = parameters.getChannel();
         this.id                = parameters.getId();
@@ -91,6 +99,7 @@ public class Consumer extends AgentBase implements Runnable {
         this.polling           = parameters.isPolling();
         this.pollingInterval   = parameters.getPollingInterval();
         this.consumerArguments = parameters.getConsumerArguments();
+        this.exitWhen          = parameters.getExitWhen();
 
         this.queueNames.set(new ArrayList<>(parameters.getQueueNames()));
         this.initialQueueNames = new ArrayList<>(parameters.getQueueNames());
@@ -133,7 +142,7 @@ public class Consumer extends AgentBase implements Runnable {
         }
 
 
-        this.state = new ConsumerState(parameters.getRateLimit());
+        this.state = new ConsumerState(parameters.getRateLimit(), timestampProvider);
         this.recoveryProcess = parameters.getRecoveryProcess();
         this.recoveryProcess.init(this);
     }
@@ -232,9 +241,10 @@ public class Consumer extends AgentBase implements Runnable {
 
         void handleMessage(Envelope envelope, BasicProperties properties, byte[] body, Channel ch) throws IOException {
             int currentMessageCount = state.incrementMessageCount();
+            long nowTimestamp = timestampProvider.getCurrentTime();
+            state.setLastActivityTimestamp(nowTimestamp);
             if (msgLimit == 0 || currentMessageCount <= msgLimit) {
                 long messageTimestamp = timestampExtractor.apply(properties, body);
-                long nowTimestamp = timestampProvider.getCurrentTime();
                 long diff_time = timestampProvider.getDifference(nowTimestamp, messageTimestamp);
 
                 stats.handleRecv(id.equals(envelope.getRoutingKey()) ? diff_time : 0L);
@@ -339,6 +349,47 @@ public class Consumer extends AgentBase implements Runnable {
         }
     }
 
+    void maybeStopIfNoActivityOrQueueEmpty() {
+        LOGGER.debug("Checking consumer activity");
+        if (this.exitWhen == EXIT_WHEN.NEVER) {
+            return;
+        }
+        TimestampProvider tp = state.getTimestampProvider();
+        long lastActivityTimestamp = state.getLastActivityTimestamp();
+        if (lastActivityTimestamp == -1) {
+            // this avoids not terminating a consumer that never consumes
+            state.setLastActivityTimestamp(tp.getCurrentTime());
+            return;
+        }
+        Duration idleDuration = tp.difference(tp.getCurrentTime(), lastActivityTimestamp);
+        if (idleDuration.toMillis() > 1000) {
+            LOGGER.debug("Consumer idle for {}", idleDuration);
+            List<String> queues = queueNames.get();
+            if (this.exitWhen == EXIT_WHEN.IDLE) {
+                LOGGER.debug("Terminating consumer {} because of inactivity", this);
+                countDown(STOP_REASON_CONSUMER_IDLE);
+            } else if (this.exitWhen == EXIT_WHEN.EMPTY){
+                LOGGER.debug("Checking content of consumer queue(s)");
+                boolean empty = false;
+                for (String queue : queues) {
+                    try {
+                        DeclareOk declareOk = this.channel.queueDeclarePassive(queue);
+                        LOGGER.debug("Message count for queue {}: {}", queue, declareOk.getMessageCount());
+                        if (declareOk.getMessageCount() == 0) {
+                            empty = true;
+                        }
+                    } catch (IOException e) {
+                        LOGGER.info("Error when calling queue.declarePassive({}) in consumer {}", queue, this);
+                    }
+                }
+                if (empty) {
+                    LOGGER.debug("Terminating consumer {} because its queue(s) is (are) empty", this);
+                    countDown(STOP_REASON_CONSUMER_QUEUE_EMPTY);
+                }
+            }
+        }
+    }
+
     private static String queueName(TopologyRecording recording, String queue) {
         TopologyRecording.RecordedQueue queueRecord = recording.queue(queue);
         // The recording is missing when using pre-declared, so just using the initial name.
@@ -352,10 +403,14 @@ public class Consumer extends AgentBase implements Runnable {
 
         private final float rateLimit;
         private volatile long  lastStatsTime;
+        private volatile long lastActivityTimestamp = -1;
         private final AtomicInteger msgCount = new AtomicInteger(0);
+        private final TimestampProvider timestampProvider;
 
-        protected ConsumerState(float rateLimit) {
+        protected ConsumerState(float rateLimit,
+            TimestampProvider timestampProvider) {
             this.rateLimit = rateLimit;
+            this.timestampProvider = timestampProvider;
         }
 
         public float getRateLimit() {
@@ -370,10 +425,22 @@ public class Consumer extends AgentBase implements Runnable {
             this.lastStatsTime = lastStatsTime;
         }
 
+        public void setLastActivityTimestamp(long lastActivityTimestamp) {
+            this.lastActivityTimestamp = lastActivityTimestamp;
+        }
+
+        public long getLastActivityTimestamp() {
+            return lastActivityTimestamp;
+        }
+
         public int getMsgCount() {
             return msgCount.get();
         }
 
+        public TimestampProvider getTimestampProvider() {
+            return timestampProvider;
+        }
+
         protected void setMsgCount(int msgCount) {
             this.msgCount.set(msgCount);
         }
diff --git a/src/main/java/com/rabbitmq/perf/ConsumerParameters.java b/src/main/java/com/rabbitmq/perf/ConsumerParameters.java
index 7296453..e6cd6b7 100644
--- a/src/main/java/com/rabbitmq/perf/ConsumerParameters.java
+++ b/src/main/java/com/rabbitmq/perf/ConsumerParameters.java
@@ -17,6 +17,7 @@ package com.rabbitmq.perf;
 
 import com.rabbitmq.client.Channel;
 
+import com.rabbitmq.perf.PerfTest.EXIT_WHEN;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ExecutorService;
@@ -47,6 +48,8 @@ public class ConsumerParameters {
 
     private int pollingInterval;
 
+    private EXIT_WHEN exitWhen = EXIT_WHEN.NEVER;
+
     private Map<String, Object> consumerArguments = null;
 
     public Channel getChannel() {
@@ -219,4 +222,13 @@ public class ConsumerParameters {
     public Map<String, Object> getConsumerArguments() {
         return consumerArguments;
     }
+
+    public ConsumerParameters setExitWhen(EXIT_WHEN exitWhen) {
+        this.exitWhen = exitWhen;
+        return this;
+    }
+
+    public EXIT_WHEN getExitWhen() {
+        return exitWhen;
+    }
 }
diff --git a/src/main/java/com/rabbitmq/perf/MulticastParams.java b/src/main/java/com/rabbitmq/perf/MulticastParams.java
index 640e1ba..e3f1803 100644
--- a/src/main/java/com/rabbitmq/perf/MulticastParams.java
+++ b/src/main/java/com/rabbitmq/perf/MulticastParams.java
@@ -20,6 +20,7 @@ import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.ShutdownSignalException;
 
+import com.rabbitmq.perf.PerfTest.EXIT_WHEN;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collections;
@@ -120,6 +121,8 @@ public class MulticastParams {
 
     private Map<String, Object> consumerArguments = null;
 
+    private EXIT_WHEN exitWhen = EXIT_WHEN.NEVER;
+
     // for random JSON body generation
     private AtomicReference<MessageBodySource> messageBodySourceReference = new AtomicReference<>();
 
@@ -285,6 +288,10 @@ public class MulticastParams {
         this.consumerArguments = consumerArguments;
     }
 
+    public void setExitWhen(EXIT_WHEN exitWhen) {
+        this.exitWhen = exitWhen;
+    }
+
     public int getConsumerCount() {
         return consumerCount;
     }
@@ -422,6 +429,10 @@ public class MulticastParams {
         return consumerLatencies;
     }
 
+    public EXIT_WHEN getExitWhen() {
+        return exitWhen;
+    }
+
     public void setPolling(boolean polling) {
         this.polling = polling;
     }
@@ -546,6 +557,7 @@ public class MulticastParams {
                 .setNack(this.nack)
                 .setRequeue(this.requeue)
                 .setConsumerArguments(this.consumerArguments)
+                .setExitWhen(this.exitWhen)
         );
         this.topologyHandler.next();
         return consumer;
@@ -599,7 +611,9 @@ public class MulticastParams {
     }
 
     public boolean hasLimit() {
-        return this.timeLimit > 0 || this.consumerMsgCount > 0 || this.producerMsgCount > 0;
+        return this.timeLimit > 0 || this.consumerMsgCount > 0 || this.producerMsgCount > 0
+            || this.exitWhen == EXIT_WHEN.EMPTY
+            || this.exitWhen == EXIT_WHEN.IDLE;
     }
 
     public void setExclusive(boolean exclusive) {
diff --git a/src/main/java/com/rabbitmq/perf/MulticastSet.java b/src/main/java/com/rabbitmq/perf/MulticastSet.java
index ddbab3d..8ab91d0 100644
--- a/src/main/java/com/rabbitmq/perf/MulticastSet.java
+++ b/src/main/java/com/rabbitmq/perf/MulticastSet.java
@@ -15,25 +15,46 @@
 
 package com.rabbitmq.perf;
 
-import com.rabbitmq.client.*;
-import com.rabbitmq.client.impl.recovery.AutorecoveringConnection;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import static com.rabbitmq.perf.Utils.isRecoverable;
+import static java.lang.Math.min;
+import static java.lang.String.format;
 
+import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.Address;
+import com.rabbitmq.client.AlreadyClosedException;
+import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.ConnectionFactory;
+import com.rabbitmq.client.Recoverable;
+import com.rabbitmq.client.RecoveryListener;
+import com.rabbitmq.client.impl.recovery.AutorecoveringConnection;
+import com.rabbitmq.perf.PerfTest.EXIT_WHEN;
 import java.io.IOException;
 import java.net.URISyntaxException;
 import java.security.KeyManagementException;
 import java.security.NoSuchAlgorithmException;
-import java.util.*;
-import java.util.concurrent.*;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Function;
 import java.util.function.Supplier;
 import java.util.stream.IntStream;
-
-import static com.rabbitmq.perf.Utils.isRecoverable;
-import static java.lang.Math.min;
-import static java.lang.String.format;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class MulticastSet {
 
@@ -165,7 +186,7 @@ public class MulticastSet {
 
             this.params.resetTopologyHandler();
 
-            Runnable[] consumerRunnables = new Runnable[params.getConsumerThreadCount()];
+            Consumer[] consumerRunnables = new Consumer[params.getConsumerThreadCount()];
             Connection[] consumerConnections = new Connection[params.getConsumerCount()];
             Function<Integer, ExecutorService> consumersExecutorsFactory;
             consumersExecutorsFactory = createConsumersExecutorsFactory();
@@ -188,6 +209,24 @@ public class MulticastSet {
             startConsumers(consumerRunnables);
             startProducers(producerStates);
 
+            if (params.getExitWhen() == EXIT_WHEN.EMPTY || params.getExitWhen() == EXIT_WHEN.IDLE) {
+                ScheduledExecutorService scheduledExecutorService =
+                    this.threadingHandler.scheduledExecutorService(
+                        "perf-test-queue-empty-consumer-idle-scheduler", 1);
+                    scheduledExecutorService.scheduleAtFixedRate(
+                    () -> {
+                      for (Consumer consumer : consumerRunnables) {
+                        try {
+                            consumer.maybeStopIfNoActivityOrQueueEmpty();
+                        } catch (Exception e) {
+                            LOGGER.info("Error while checking exit-when for consumer {}: {}", consumer, e.getMessage());
+                        }
+                      }
+                    },
+                    2,
+                    1,
+                    TimeUnit.SECONDS);
+            }
 
             AutoCloseable shutdownSequence;
             int shutdownTimeout = this.params.getShutdownTimeout();
@@ -313,7 +352,7 @@ public class MulticastSet {
     private void createConsumers(boolean announceStartup,
                                  Runnable[] consumerRunnables,
                                  Connection[] consumerConnections,
-                                 Function<Integer, ExecutorService> consumersExecutorsFactory) throws URISyntaxException, NoSuchAlgorithmException, KeyManagementException, IOException, TimeoutException {
+                                 Function<Integer, ExecutorService> consumersExecutorsFactory) throws IOException, TimeoutException {
         for (int i = 0; i < consumerConnections.length; i++) {
             if (announceStartup) {
                 System.out.println("id: " + testID + ", starting consumer #" + i);
@@ -327,7 +366,8 @@ public class MulticastSet {
                 if (announceStartup) {
                     System.out.println("id: " + testID + ", starting consumer #" + i + ", channel #" + j);
                 }
-                consumerRunnables[(i * params.getConsumerChannelCount()) + j] = params.createConsumer(consumerConnection, stats, this.consumerLatencyIndicator, this.completionHandler, executorService);
+                Consumer consumer = params.createConsumer(consumerConnection, stats, this.consumerLatencyIndicator, this.completionHandler, executorService);
+                consumerRunnables[(i * params.getConsumerChannelCount()) + j] = consumer;
             }
         }
     }
diff --git a/src/main/java/com/rabbitmq/perf/PerfTest.java b/src/main/java/com/rabbitmq/perf/PerfTest.java
index 921ccf7..2c8824c 100644
--- a/src/main/java/com/rabbitmq/perf/PerfTest.java
+++ b/src/main/java/com/rabbitmq/perf/PerfTest.java
@@ -212,6 +212,18 @@ public class PerfTest {
                 queueArguments.put("x-queue-type", "quorum");
             }
 
+            String exitWhenParameter = strArg(cmd, "ew", null);
+            EXIT_WHEN exitWhen;
+            if (exitWhenParameter != null) {
+                if (!"empty".equals(exitWhenParameter) && !"idle".equals(exitWhenParameter)) {
+                    System.out.println("--exit-when must be 'empty' or 'idle'.");
+                    systemExiter.exit(1);
+                }
+                exitWhen = EXIT_WHEN.valueOf(exitWhenParameter.toUpperCase(Locale.ENGLISH));
+            } else {
+                exitWhen = EXIT_WHEN.NEVER;
+            }
+
             ConnectionFactory factory = new ConnectionFactory();
             if (disableConnectionRecovery) {
                 factory.setAutomaticRecoveryEnabled(false);
@@ -379,6 +391,7 @@ public class PerfTest {
             p.setBodyCount(bodyCount);
             p.setConsumerArguments(convertKeyValuePairs(consumerArgs));
             p.setQueuesInSequence(queueFile != null);
+            p.setExitWhen(exitWhen);
 
             ConcurrentMap<String, Integer> completionReasons = new ConcurrentHashMap<>();
 
@@ -490,9 +503,13 @@ public class PerfTest {
             if (p.getProducerMsgCount() > 0) {
                 countLimit += p.getProducerThreadCount();
             }
-            if (p.getConsumerMsgCount() > 0) {
+            if (p.getConsumerMsgCount() > 0 || p.getExitWhen() == EXIT_WHEN.EMPTY
+                || p.getExitWhen() == EXIT_WHEN.IDLE) {
                 countLimit += p.getConsumerThreadCount();
             }
+            LOGGER.debug("Creating completion handler with time limit {} and count limit {}",
+                p.getTimeLimit(), countLimit
+                );
             completionHandler = new MulticastSet.DefaultCompletionHandler(
                 p.getTimeLimit(),
                 countLimit,
@@ -699,6 +716,7 @@ public class PerfTest {
         options.addOption(new Option("qf", "queue-file", true, "file to look up queue names from"));
         options.addOption(new Option("sni", "server-name-indication", true, "server names for Server Name Indication TLS parameter, separated by commas"));
         options.addOption(new Option("qq", "quorum-queue", false,"create quorum queue(s)"));
+        options.addOption(new Option("ew", "exit-when", true, "exit when queue(s) empty or consumer(s) idle for 1 second, valid values are empty or idle"));
         return options;
     }
 
@@ -867,4 +885,7 @@ public class PerfTest {
 
     private static final Function<String, String> ENVIRONMENT_VARIABLE_LOOKUP = name -> System.getenv(name);
 
+    enum EXIT_WHEN {
+        NEVER, EMPTY, IDLE
+    }
 }
diff --git a/src/main/java/com/rabbitmq/perf/TimestampProvider.java b/src/main/java/com/rabbitmq/perf/TimestampProvider.java
index 0fe3b96..22138fc 100644
--- a/src/main/java/com/rabbitmq/perf/TimestampProvider.java
+++ b/src/main/java/com/rabbitmq/perf/TimestampProvider.java
@@ -15,6 +15,8 @@
 
 package com.rabbitmq.perf;
 
+import java.time.Duration;
+
 public class TimestampProvider {
     private final boolean useMillis;
     private final boolean isTimestampInHeader;
@@ -39,4 +41,13 @@ public class TimestampProvider {
     public long getDifference(long ts1, long ts2) {
         return Math.abs(ts1 - ts2);
     }
+
+    Duration difference(long ts1, long ts2) {
+        long difference = getDifference(ts1, ts2);
+        if (useMillis) {
+            return Duration.ofMillis(difference);
+        } else {
+            return Duration.ofNanos(difference);
+        }
+    }
 }
diff --git a/src/test/java/com/rabbitmq/perf/MessageCountTimeLimitAndPublishingIntervalRateTest.java b/src/test/java/com/rabbitmq/perf/MessageCountTimeLimitAndPublishingIntervalRateTest.java
index 9675063..4611571 100644
--- a/src/test/java/com/rabbitmq/perf/MessageCountTimeLimitAndPublishingIntervalRateTest.java
+++ b/src/test/java/com/rabbitmq/perf/MessageCountTimeLimitAndPublishingIntervalRateTest.java
@@ -15,8 +15,13 @@
 
 package com.rabbitmq.perf;
 
+import com.rabbitmq.client.AMQP.Queue.DeclareOk;
 import com.rabbitmq.client.Consumer;
 import com.rabbitmq.client.*;
+import com.rabbitmq.perf.PerfTest.EXIT_WHEN;
+import java.util.stream.Collectors;
+import org.assertj.core.api.Assertions;
+import org.assertj.core.api.Condition;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -32,7 +37,6 @@ import java.util.*;
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.IntStream;
 import java.util.stream.Stream;
@@ -43,6 +47,7 @@ import static com.rabbitmq.perf.TestUtils.waitAtMost;
 import static java.lang.String.format;
 import static java.util.Arrays.asList;
 import static java.util.Collections.singletonList;
+import static org.assertj.core.api.Assertions.anyOf;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
@@ -78,6 +83,17 @@ public class MessageCountTimeLimitAndPublishingIntervalRateTest {
         );
     }
 
+    static Arguments[] consumerActivityArguments() {
+        List<Arguments> consumerArguments = consumerCountArguments().collect(Collectors.toList());
+        List<Arguments> result = new ArrayList<>();
+        Stream.of("idle", "empty").forEach(exitWhen -> {
+            for (Arguments consumerArgument : consumerArguments) {
+                result.add(Arguments.arguments(exitWhen, consumerArgument.get()[0], consumerArgument.get()[1]));
+            }
+        });
+        return result.toArray(new Arguments[0]);
+    }
+
     static Stream<Arguments> pollingWithBasicGetArguments() {
         return Stream.of(
                 Arguments.of(1, 1),
@@ -297,6 +313,64 @@ public class MessageCountTimeLimitAndPublishingIntervalRateTest {
                 .containsValue(consumersCount * channelsCount);
     }
 
+    @ParameterizedTest
+    @MethodSource("consumerActivityArguments")
+    public void consumerActivity(String exitWhen, int consumersCount, int channelsCount) throws Exception {
+        int messagesCount = consumersCount * channelsCount;
+        params.setExitWhen(EXIT_WHEN.valueOf(exitWhen.toUpperCase(Locale.ENGLISH)));
+        params.setConsumerCount(consumersCount);
+        params.setConsumerChannelCount(channelsCount);
+        params.setQueueNames(asList("queue"));
+
+        CountDownLatch consumersLatch = new CountDownLatch(consumersCount * channelsCount);
+        AtomicInteger consumerTagCounter = new AtomicInteger(0);
+        List<Consumer> consumers = new CopyOnWriteArrayList<>();
+        Channel channel = proxy(Channel.class,
+            callback("basicConsume", (proxy, method, args) -> {
+                consumers.add((Consumer) args[3]);
+                consumersLatch.countDown();
+                return consumerTagCounter.getAndIncrement() + "";
+            }),
+            callback("queueDeclarePassive", (proxy, method, args) -> new DeclareOk.Builder().queue("test").messageCount(0).build())
+        );
+
+        Connection connection = proxy(Connection.class,
+            callback("createChannel", (proxy, method, args) -> channel)
+        );
+
+        MulticastSet multicastSet = getMulticastSet(connectionFactoryThatReturns(connection));
+        run(multicastSet);
+
+        waitForRunToStart();
+
+        assertThat(consumersLatch.await(5, TimeUnit.SECONDS))
+            .as(consumersCount * channelsCount + " consumer(s) should have been registered by now")
+            .isTrue();
+
+        waitAtMost(20, () -> consumers.size() == (consumersCount * channelsCount));
+
+        Collection<Future<?>> sendTasks = new ArrayList<>(consumers.size());
+        for (Consumer consumer : consumers) {
+            Collection<Future<?>> tasks = sendMessagesToConsumer(messagesCount, consumer);
+            sendTasks.addAll(tasks);
+        }
+
+        for (Future<?> latch : sendTasks) {
+            latch.get(10, TimeUnit.SECONDS);
+        }
+
+        waitAtMost(20, () -> testIsDone.get());
+        assertThat(shutdownReasons)
+            .hasSize(1)
+            .hasKeySatisfying(anyOf(isKey(com.rabbitmq.perf.Consumer.STOP_REASON_CONSUMER_IDLE), isKey(
+                com.rabbitmq.perf.Consumer.STOP_REASON_CONSUMER_QUEUE_EMPTY)))
+            .containsValue(consumersCount * channelsCount);
+    }
+
+    static Condition<String> isKey(String value) {
+        return new Condition<>(key -> key.equals(value), "'%s' is a key", value);
+    }
+
     // --time 5 -x 1 --pmessages 10 -y 1 --cmessages 10
     @Test
     public void timeLimitTakesPrecedenceOverCounts() throws Exception {
diff --git a/src/test/resources/logback-test.xml b/src/test/resources/logback-test.xml
index 4d3c69f..e6dce39 100644
--- a/src/test/resources/logback-test.xml
+++ b/src/test/resources/logback-test.xml
@@ -7,6 +7,7 @@
 
     <logger name="com.rabbitmq.perf" level="info"/>
     <logger name="com.rabbitmq.perf.MulticastSet" level="info"/>
+    <logger name="com.rabbitmq.perf.Consumer" level="info"/>
 
 
     <root level="warn">
['src/main/java/com/rabbitmq/perf/ConsumerParameters.java', 'src/test/java/com/rabbitmq/perf/MessageCountTimeLimitAndPublishingIntervalRateTest.java', 'src/main/java/com/rabbitmq/perf/MulticastParams.java', 'src/test/resources/logback-test.xml', 'src/main/java/com/rabbitmq/perf/PerfTest.java', 'src/main/java/com/rabbitmq/perf/Consumer.java', 'src/main/java/com/rabbitmq/perf/MulticastSet.java', 'src/main/java/com/rabbitmq/perf/TimestampProvider.java']