GH-837: Fix DMLC Recovery with queue removal

Fixes #837

- Skip recovery of consumers who's queue has been deleted during recovery
- If recovery is aborted (no connection) reset `consumersToRestart`

**cherry-pick to 2.0.x**

Author: garyrussell

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

@@ -23,6 +23,7 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Date;
+import java.util.Iterator;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
@@ -448,8 +449,18 @@ protected void actualStart() throws Exception {
 						if (restartableConsumers.size() > 0) {
 							doRedeclareElementsIfNecessary();
 						}
-						for (SimpleConsumer consumer : restartableConsumers) {
-							if (this.logger.isDebugEnabled() && restartableConsumers.size() > 0) {
+						Iterator<SimpleConsumer> iterator = restartableConsumers.iterator();
+						while (iterator.hasNext()) {
+							SimpleConsumer consumer = iterator.next();
+							iterator.remove();
+							if (!DirectMessageListenerContainer.this.consumersByQueue
+									.containsKey(consumer.getQueue())) {
+								if (this.logger.isDebugEnabled()) {
+									this.logger.debug("Skipping restart of consumer " + consumer);
+								}
+								continue;
+							}
+							if (this.logger.isDebugEnabled()) {
 								this.logger.debug("Attempting to restart consumer " + consumer);
 							}
 							Queue queue = namesToQueues.get(consumer.getQueue());
@@ -471,6 +482,10 @@ protected void actualStart() throws Exception {
 									this.logger.error("Application context is closed, terminating");
 									this.taskScheduler.schedule(this::stop, new Date());
 								}
+								this.consumersToRestart.addAll(restartableConsumers);
+								if (this.logger.isTraceEnabled()) {
+									this.logger.trace("After restart exception, consumers to restart now: " + this.consumersToRestart);
+								}
 								break;
 							}
 						}
@@ -772,6 +787,9 @@ private void cancelConsumer(SimpleConsumer consumer) {
 	private void addConsumerToRestart(SimpleConsumer consumer) {
 		if (this.started) {
 			this.consumersToRestart.add(consumer);
+			if (this.logger.isTraceEnabled()) {
+				this.logger.trace("Consumers to restart now: " + this.consumersToRestart);
+			}
 		}
 	}
 

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

@@ -25,9 +25,11 @@
 import static org.mockito.ArgumentMatchers.anyLong;
 import static org.mockito.ArgumentMatchers.anyMap;
 import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.BDDMockito.given;
 import static org.mockito.BDDMockito.willAnswer;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 
 import java.util.concurrent.CountDownLatch;
@@ -198,6 +200,64 @@ else if (i.getArgument(0).equals(17L)) {
 		verify(channel).basicAck(20L, true);
 	}
 
+	@Test
+	public void testRemoveQueuesWhileNotConnected() throws Exception {
+		ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
+		Connection connection = mock(Connection.class);
+		ChannelProxy channel = mock(ChannelProxy.class);
+		Channel rabbitChannel = mock(AutorecoveringChannel.class);
+		given(channel.getTargetChannel()).willReturn(rabbitChannel);
+
+		given(connectionFactory.createConnection()).willReturn(connection);
+		given(connection.createChannel(anyBoolean())).willReturn(channel);
+		final AtomicBoolean isOpen = new AtomicBoolean(true);
+		willAnswer(i -> isOpen.get()).given(channel).isOpen();
+		given(channel.queueDeclarePassive(Mockito.anyString()))
+				.willAnswer(invocation -> mock(AMQP.Queue.DeclareOk.class));
+		given(channel.basicConsume(anyString(), anyBoolean(), anyString(), anyBoolean(), anyBoolean(),
+						anyMap(), any(Consumer.class))).willReturn("consumerTag");
+
+		final CountDownLatch latch1 = new CountDownLatch(2);
+		final CountDownLatch latch3 = new CountDownLatch(3);
+		final AtomicInteger qos = new AtomicInteger();
+		willAnswer(i -> {
+			qos.set(i.getArgument(0));
+			latch1.countDown();
+			latch3.countDown();
+			return null;
+		}).given(channel).basicQos(anyInt());
+		final CountDownLatch latch2 = new CountDownLatch(2);
+		willAnswer(i -> {
+			latch2.countDown();
+			return null;
+		}).given(channel).basicCancel("consumerTag");
+
+		DirectMessageListenerContainer container = new DirectMessageListenerContainer(connectionFactory);
+		container.setQueueNames("test1", "test2");
+		container.setPrefetchCount(2);
+		container.setMonitorInterval(100);
+		container.setFailedDeclarationRetryInterval(100);
+		container.setRecoveryInterval(100);
+		container.setShutdownTimeout(1);
+		container.afterPropertiesSet();
+		container.start();
+
+		assertTrue(latch1.await(10, TimeUnit.SECONDS));
+		assertThat(qos.get(), equalTo(2));
+		isOpen.set(false);
+		assertTrue(latch2.await(10, TimeUnit.SECONDS));
+		container.removeQueueNames("test1");
+		isOpen.set(true);
+		assertTrue(latch3.await(10, TimeUnit.SECONDS));
+
+		verify(channel, times(1)).basicConsume(eq("test1"), anyBoolean(), anyString(), anyBoolean(), anyBoolean(),
+				anyMap(), any(Consumer.class));
+		verify(channel, times(2)).basicConsume(eq("test2"), anyBoolean(), anyString(), anyBoolean(), anyBoolean(),
+				anyMap(), any(Consumer.class));
+
+		container.stop();
+	}
+
 	private Envelope envelope(long tag) {
 		return new Envelope(tag,  false, "", "");
 	}