GH-1246: SMLC: Fix addQueueNames

Resolves #1246

- called `queuesChanged()` twice
- didn't wait for old consumers to exit, causing problems with exclusive consumers

Author: garyrussell

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java

@@ -413,8 +413,7 @@ public void setQueueNames(String... queueName) {
 	 */
 	@Override
 	public void addQueueNames(String... queueName) {
-		super.addQueueNames(queueName);
-		queuesChanged();
+		super.addQueueNames(queueName); // calls addQueues() which will cycle consumers
 	}
 
 	/**
@@ -812,7 +811,13 @@ private void queuesChanged() {
 					consumerIterator.remove();
 					count++;
 				}
-				this.addAndStartConsumers(count);
+				try {
+					this.cancellationLock.await(getShutdownTimeout(), TimeUnit.MILLISECONDS);
+				}
+				catch (InterruptedException e) {
+					Thread.currentThread().interrupt();
+				}
+				addAndStartConsumers(count);
 			}
 		}
 	}

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -19,6 +19,7 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.BDDMockito.willAnswer;
 import static org.mockito.Mockito.atLeastOnce;
 import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.spy;
@@ -132,14 +133,21 @@ public void clear() throws Exception {
 	@Test
 	public void testChangeQueues() throws Exception {
 		CountDownLatch latch = new CountDownLatch(30);
-		container =
-				createContainer(new MessageListenerAdapter(new PojoListener(latch)), queue.getName(), queue1.getName());
+		AtomicInteger restarts = new AtomicInteger();
+		container = spy(createContainer(new MessageListenerAdapter(new PojoListener(latch)), false, queue.getName(),
+				queue1.getName()));
+		willAnswer(invocation -> {
+			restarts.incrementAndGet();
+			invocation.callRealMethod();
+			return null;
+		}).given(container).addAndStartConsumers(1);
 		final CountDownLatch consumerLatch = new CountDownLatch(1);
 		this.container.setApplicationEventPublisher(e -> {
 			if (e instanceof AsyncConsumerStoppedEvent) {
 				consumerLatch.countDown();
 			}
 		});
+		this.container.start();
 		for (int i = 0; i < 10; i++) {
 			template.convertAndSend(queue.getName(), i + "foo");
 			template.convertAndSend(queue1.getName(), i + "foo");
@@ -153,6 +161,7 @@ public void testChangeQueues() throws Exception {
 		assertThat(waited).as("Timed out waiting for message").isTrue();
 		assertThat(template.receiveAndConvert(queue.getName())).isNull();
 		assertThat(template.receiveAndConvert(queue1.getName())).isNull();
+		assertThat(restarts.get()).isEqualTo(1);
 	}
 
 	@Test