GH-1296: Fix DMLC Recovery: Missing Queue at Start

Resolves https://github.com/spring-projects/spring-amqp/issues/1296

- Add `MissingQueueEvent`
- Fix detection of a deleted queue in recovery - previously incorrectly used the
  absense of the queue in `consumersByQueue`, which can be empty if missing during
  start
- Add an index to `SimpleConsumer`
- When adjusting consumer counts, look for gaps in the index sequence because reducing
  the consumer count can remove any idle consumer.
- Change consumers to restart to a `Set` to avoid OOM when no broker
  (see https://github.com/spring-projects/spring-amqp/pull/642)
- Unconditionally add consumers to `consumersToRestart`

**cherry-pick to 2.2.x, 2.1.x**

# Conflicts:
#	spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java
#	spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainerIntegrationTests.java

# Conflicts:
#	spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java
#	spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java
#	spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java
#	spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java
#	spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainerIntegrationTests.java

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2021 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.
@@ -1651,6 +1651,13 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
 		}
 	}
 
+	protected void publishMissingQueueEvent(String queue) {
+		if (this.applicationEventPublisher != null) {
+			this.applicationEventPublisher
+					.publishEvent(new MissingQueueEvent(this, queue));
+		}
+	}
+
 	protected final void publishIdleContainerEvent(long idleTime) {
 		if (this.applicationEventPublisher != null) {
 			this.applicationEventPublisher.publishEvent(

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2018 the original author or authors.
+ * Copyright 2002-2021 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.
@@ -156,6 +156,8 @@ public class BlockingQueueConsumer {
 
 	private ApplicationEventPublisher applicationEventPublisher;
 
+	private java.util.function.Consumer<String> missingQueuePublisher = str -> { };
+
 	private volatile long abortStarted;
 
 	private volatile boolean normalCancel;
@@ -370,6 +372,15 @@ public class BlockingQueueConsumer {
 		this.applicationEventPublisher = applicationEventPublisher;
 	}
 
+	/**
+	 * Set the publisher for a missing queue event.
+	 * @param missingQueuePublisher the publisher.
+	 * @since 2.1.18
+	 */
+	public void setMissingQueuePublisher(java.util.function.Consumer<String> missingQueuePublisher) {
+		this.missingQueuePublisher = missingQueuePublisher;
+	}
+
 	/**
 	 * Clear the delivery tags when rolling back with an external transaction
 	 * manager.
@@ -691,6 +702,7 @@ public class BlockingQueueConsumer {
 				if (logger.isWarnEnabled()) {
 					logger.warn("Failed to declare queue: " + queueName);
 				}
+				this.missingQueuePublisher.accept(queueName);
 				if (!this.channel.isOpen()) {
 					throw new AmqpIOException(e);
 				}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2018 the original author or authors.
+ * Copyright 2016-2021 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.
@@ -24,12 +24,14 @@ import java.util.Collection;
 import java.util.Collections;
 import java.util.Date;
 import java.util.Iterator;
+import java.util.LinkedHashSet;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Properties;
 import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
@@ -101,7 +103,9 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta
 
 	protected final List<SimpleConsumer> consumers = new LinkedList<>(); // NOSONAR
 
-	private final List<SimpleConsumer> consumersToRestart = new LinkedList<>();
+	private final Set<SimpleConsumer> consumersToRestart = new LinkedHashSet<>();
+
+	private final Set<String> removedQueues = ConcurrentHashMap.newKeySet();
 
 	private final MultiValueMap<String, SimpleConsumer> consumersByQueue = new LinkedMultiValueMap<>();
 
@@ -241,6 +245,7 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta
 		Assert.notNull(queueNames, "'queueNames' cannot be null");
 		Assert.noNullElements(queueNames, "'queueNames' cannot contain null elements");
 		try {
+			Arrays.stream(queueNames).forEach(this.removedQueues::remove);
 			addQueues(Arrays.stream(queueNames));
 		}
 		catch (AmqpIOException e) {
@@ -254,6 +259,9 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta
 		Assert.notNull(queues, "'queues' cannot be null");
 		Assert.noNullElements(queues, "'queues' cannot contain null elements");
 		try {
+			Arrays.stream(queues)
+				.map(q -> q.getActualName())
+				.forEach(this.removedQueues::remove);
 			addQueues(Arrays.stream(queues).map(Queue::getName));
 		}
 		catch (AmqpIOException e) {
@@ -296,7 +304,10 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta
 		if (isRunning()) {
 			synchronized (this.consumersMonitor) {
 				checkStartState();
-				queueNames.map(this.consumersByQueue::remove)
+				queueNames.map(queue -> {
+							this.removedQueues.add(queue);
+							return this.consumersByQueue.remove(queue);
+						})
 						.filter(Objects::nonNull)
 						.flatMap(Collection::stream)
 						.forEach(this::cancelConsumer);
@@ -311,7 +322,21 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta
 			for (String queue : getQueueNames()) {
 				while (this.consumersByQueue.get(queue) == null
 						|| this.consumersByQueue.get(queue).size() < newCount) { // NOSONAR never null
-					doConsumeFromQueue(queue);
+					List<SimpleConsumer> cBQ = this.consumersByQueue.get(queue);
+					int index = 0;
+					if (cBQ != null) {
+						// find a gap or set the index to the end
+						List<Integer> indices = cBQ.stream()
+								.map(cons -> cons.getIndex())
+								.sorted()
+								.collect(Collectors.toList());
+						for (index = 0; index < indices.size(); index++) {
+							if (index < indices.get(index)) {
+								break;
+							}
+						}
+					}
+					doConsumeFromQueue(queue, index);
 				}
 				List<SimpleConsumer> consumerList = this.consumersByQueue.get(queue);
 				if (consumerList != null && consumerList.size() > newCount) {
@@ -428,9 +453,9 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta
 			checkConsumers(now);
 			if (this.lastRestartAttempt + getFailedDeclarationRetryInterval() < now) {
 				synchronized (this.consumersMonitor) {
-					List<SimpleConsumer> restartableConsumers = new ArrayList<>(this.consumersToRestart);
-					this.consumersToRestart.clear();
 					if (this.started) {
+						List<SimpleConsumer> restartableConsumers = new ArrayList<>(this.consumersToRestart);
+						this.consumersToRestart.clear();
 						if (restartableConsumers.size() > 0) {
 							doRedeclareElementsIfNecessary();
 						}
@@ -438,10 +463,9 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta
 						while (iterator.hasNext()) {
 							SimpleConsumer consumer = iterator.next();
 							iterator.remove();
-							if (!DirectMessageListenerContainer.this.consumersByQueue
-									.containsKey(consumer.getQueue())) {
+							if (DirectMessageListenerContainer.this.removedQueues.contains(consumer.getQueue())) {
 								if (this.logger.isDebugEnabled()) {
-									this.logger.debug("Skipping restart of consumer " + consumer);
+									this.logger.debug("Skipping restart of consumer, queue removed " + consumer);
 								}
 								continue;
 							}
@@ -514,11 +538,11 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta
 			if (StringUtils.hasText(actualName)) {
 				namesToQueues.remove(consumer.getQueue());
 				namesToQueues.put(actualName, queue);
-				consumer = new SimpleConsumer(null, null, actualName);
+				consumer = new SimpleConsumer(null, null, actualName, consumer.getIndex());
 			}
 		}
 		try {
-			doConsumeFromQueue(consumer.getQueue());
+			doConsumeFromQueue(consumer.getQueue(), consumer.getIndex());
 			return true;
 		}
 		catch (AmqpConnectException | AmqpIOException e) {
@@ -644,12 +668,12 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta
 		// Possible race with setConsumersPerQueue and the task launched by start()
 		if (CollectionUtils.isEmpty(list)) {
 			for (int i = 0; i < this.consumersPerQueue; i++) {
-				doConsumeFromQueue(queue);
+				doConsumeFromQueue(queue, i);
 			}
 		}
 	}
 
-	private void doConsumeFromQueue(String queue) {
+	private void doConsumeFromQueue(String queue, int index) {
 		if (!isActive()) {
 			if (this.logger.isDebugEnabled()) {
 				this.logger.debug("Consume from queue " + queue + " ignore, container stopping");
@@ -666,7 +690,7 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta
 		}
 		catch (Exception e) {
 			publishConsumerFailedEvent(e.getMessage(), false, e);
-			addConsumerToRestart(new SimpleConsumer(null, null, queue));
+			addConsumerToRestart(new SimpleConsumer(null, null, queue, index));
 			throw e instanceof AmqpConnectException // NOSONAR exception type check
 					? (AmqpConnectException) e
 					: new AmqpConnectException(e);
@@ -676,7 +700,7 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta
 				SimpleResourceHolder.pop(getRoutingConnectionFactory()); // NOSONAR never null here
 			}
 		}
-		SimpleConsumer consumer = consume(queue, connection);
+		SimpleConsumer consumer = consume(queue, index, connection);
 		synchronized (this.consumersMonitor) {
 			if (consumer != null) {
 				this.cancellationLock.add(consumer);
@@ -693,13 +717,13 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta
 	}
 
 	@Nullable
-	private SimpleConsumer consume(String queue, Connection connection) {
+	private SimpleConsumer consume(String queue, int index, Connection connection) {
 		Channel channel = null;
 		SimpleConsumer consumer = null;
 		try {
 			channel = connection.createChannel(isChannelTransacted());
 			channel.basicQos(getPrefetchCount());
-			consumer = new SimpleConsumer(connection, channel, queue);
+			consumer = new SimpleConsumer(connection, channel, queue, index);
 			channel.queueDeclarePassive(queue);
 			consumer.consumerTag = channel.basicConsume(queue, getAcknowledgeMode().isAutoAck(),
 					(getConsumerTagStrategy() != null
@@ -713,13 +737,14 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta
 			RabbitUtils.closeChannel(channel);
 			RabbitUtils.closeConnection(connection);
 
-			consumer = handleConsumeException(queue, consumer, e);
+			consumer = handleConsumeException(queue, index, consumer, e);
 		}
 		return consumer;
 	}
 
 	@Nullable
-	private SimpleConsumer handleConsumeException(String queue, SimpleConsumer consumerArg, Exception e) {
+	private SimpleConsumer handleConsumeException(String queue, int index, @Nullable SimpleConsumer consumerArg,
+			Exception e) {
 
 		SimpleConsumer consumer = consumerArg;
 		if (e.getCause() instanceof ShutdownSignalException
@@ -730,6 +755,7 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta
 		}
 		else if (e.getCause() instanceof ShutdownSignalException
 				&& RabbitUtils.isPassiveDeclarationChannelClose((ShutdownSignalException) e.getCause())) {
+			publishMissingQueueEvent(queue);
 			this.logger.error("Queue not present, scheduling consumer "
 				+ (consumer == null ? "for queue " + queue : consumer) + " for restart", e);
 		}
@@ -739,7 +765,7 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta
 		}
 
 		if (consumer == null) {
-			addConsumerToRestart(new SimpleConsumer(null, null, queue));
+			addConsumerToRestart(new SimpleConsumer(null, null, queue, index));
 		}
 		else {
 			addConsumerToRestart(consumer);
@@ -833,11 +859,9 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta
 	}
 
 	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);
-			}
+		this.consumersToRestart.add(consumer);
+		if (this.logger.isTraceEnabled()) {
+			this.logger.trace("Consumers to restart now: " + this.consumersToRestart);
 		}
 	}
 
@@ -860,6 +884,8 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta
 
 		private final String queue;
 
+		private final int index;
+
 		private final boolean ackRequired;
 
 		private final ConnectionFactory connectionFactory = getConnectionFactory();
@@ -894,10 +920,11 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta
 
 		private volatile boolean ackFailed;
 
-		SimpleConsumer(Connection connection, Channel channel, String queue) {
+		SimpleConsumer(@Nullable Connection connection, @Nullable Channel channel, String queue, int index) {
 			super(channel);
 			this.connection = connection;
 			this.queue = queue;
+			this.index = index;
 			this.ackRequired = !getAcknowledgeMode().isAutoAck() && !getAcknowledgeMode().isManual();
 			if (channel instanceof ChannelProxy) {
 				this.targetChannel = ((ChannelProxy) channel).getTargetChannel();
@@ -907,10 +934,14 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta
 			}
 		}
 
-		private String getQueue() {
+		String getQueue() {
 			return this.queue;
 		}
 
+		int getIndex() {
+			return this.index;
+		}
+
 		@Override
 		public String getConsumerTag() {
 			return this.consumerTag;
@@ -1203,9 +1234,53 @@ public class DirectMessageListenerContainer extends AbstractMessageListenerConta
 			consumerRemoved(this);
 		}
 
+		@Override
+		public int hashCode() {
+			final int prime = 31;
+			int result = 1;
+			result = prime * result + getEnclosingInstance().hashCode();
+			result = prime * result + this.index;
+			result = prime * result + ((this.queue == null) ? 0 : this.queue.hashCode());
+			return result;
+		}
+
+		@Override
+		public boolean equals(Object obj) {
+			if (this == obj) {
+				return true;
+			}
+			if (obj == null) {
+				return false;
+			}
+			if (getClass() != obj.getClass()) {
+				return false;
+			}
+			SimpleConsumer other = (SimpleConsumer) obj;
+			if (!getEnclosingInstance().equals(other.getEnclosingInstance())) {
+				return false;
+			}
+			if (this.index != other.index) {
+				return false;
+			}
+			if (this.queue == null) {
+				if (other.queue != null) {
+					return false;
+				}
+			}
+			else if (!this.queue.equals(other.queue)) {
+				return false;
+			}
+			return true;
+		}
+
+		private DirectMessageListenerContainer getEnclosingInstance() {
+			return DirectMessageListenerContainer.this;
+		}
+
 		@Override
 		public String toString() {
-			return "SimpleConsumer [queue=" + this.queue + ", consumerTag=" + this.consumerTag
+			return "SimpleConsumer [queue=" + this.queue + ", index=" + this.index
+					+ ", consumerTag=" + this.consumerTag
 					+ " identity=" + ObjectUtils.getIdentityHexString(this) + "]";
 		}
 

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

@@ -0,0 +1,57 @@
+/*
+ * Copyright 2021 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.amqp.rabbit.listener;
+
+import org.springframework.amqp.event.AmqpEvent;
+
+/**
+ * Event published when a missing queue is detected.
+ *
+ * @author Gary Russell
+ * @since 2.1.18
+ *
+ */
+public class MissingQueueEvent extends AmqpEvent {
+
+	private static final long serialVersionUID = 1L;
+
+	private final String queue;
+
+	/**
+	 * Construct an instance with the provided source and queue.
+	 * @param source the source.
+	 * @param queue the queue.
+	 */
+	public MissingQueueEvent(Object source, String queue) {
+		super(source);
+		this.queue = queue;
+	}
+
+	/**
+	 * Return the missing queue.
+	 * @return the queue.
+	 */
+	public String getQueue() {
+		return this.queue;
+	}
+
+	@Override
+	public String toString() {
+		return "MissingQueueEvent [queue=" + this.queue + ", source=" + this.source + "]";
+	}
+
+}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2018 the original author or authors.
+ * Copyright 2002-2021 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.
@@ -775,6 +775,7 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta
 		consumer = new BlockingQueueConsumer(getConnectionFactory(), getMessagePropertiesConverter(),
 				this.cancellationLock, getAcknowledgeMode(), isChannelTransacted(), actualPrefetchCount,
 				isDefaultRequeueRejected(), getConsumerArguments(), isNoLocal(), isExclusive(), queues);
+		consumer.setMissingQueuePublisher(this::publishMissingQueueEvent);
 		if (this.declarationRetries != null) {
 			consumer.setDeclarationRetries(this.declarationRetries);
 		}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2018 the original author or authors.
+ * Copyright 2016-2021 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.
@@ -35,6 +35,7 @@ import static org.mockito.Mockito.verify;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Properties;
+import java.util.Set;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -97,6 +98,8 @@ public class DirectMessageListenerContainerIntegrationTests {
 
 	private static final String DLQ1 = "testDLQ1";
 
+	private static final String MISSING = "missing.DirectMessageListenerContainerIntegrationTests";
+
 	@ClassRule
 	public static BrokerRunning brokerRunning = BrokerRunning.isRunningWithEmptyQueues(Q1, Q2, EQ1, EQ2, DLQ1);
 
@@ -277,7 +280,7 @@ public class DirectMessageListenerContainerIntegrationTests {
 		container.setConsumerTagStrategy(new Tag());
 		container.afterPropertiesSet();
 		container.setQueues(new Queue(Q1));
-		assertArrayEquals(new String[] { Q1 }, container.getQueueNames());
+		assertArrayEquals(new String[]{ Q1 }, container.getQueueNames());
 		container.start();
 		container.addQueues(new Queue(Q2));
 		assertTrue(consumersOnQueue(Q1, 2));
@@ -681,6 +684,31 @@ public class DirectMessageListenerContainerIntegrationTests {
 		cf.destroy();
 	}
 
+	@Test
+	public void missingQueueOnStart() throws InterruptedException {
+		CachingConnectionFactory cf = new CachingConnectionFactory("localhost");
+		RabbitAdmin admin = new RabbitAdmin(cf);
+		admin.deleteQueue(MISSING);
+		DirectMessageListenerContainer container = new DirectMessageListenerContainer(cf);
+		container.setQueueNames(MISSING);
+		container.setBeanName("missingQOnStart");
+		final CountDownLatch latch = new CountDownLatch(1);
+		container.setApplicationEventPublisher(event -> {
+			if (event instanceof MissingQueueEvent) {
+				admin.declareQueue(new Queue(MISSING));
+			}
+			else if (event instanceof AsyncConsumerStartedEvent) {
+				latch.countDown();
+			}
+		});
+		container.afterPropertiesSet();
+		container.start();
+		assertTrue(latch.await(10, TimeUnit.SECONDS));
+		container.stop();
+		admin.deleteQueue(MISSING);
+		cf.destroy();
+	}
+
 	private boolean consumersOnQueue(String queue, int expected) throws Exception {
 		int n = 0;
 		Properties queueProperties = admin.getQueueProperties(queue);
@@ -704,8 +732,8 @@ public class DirectMessageListenerContainerIntegrationTests {
 	}
 
 	private boolean restartConsumerCount(AbstractMessageListenerContainer container, int expected) throws Exception {
+		Set<?> consumers = TestUtils.getPropertyValue(container, "consumersToRestart", Set.class);
 		int n = 0;
-		List<?> consumers = TestUtils.getPropertyValue(container, "consumersToRestart", List.class);
 		while (n++ < 600 && consumers.size() != expected) {
 			Thread.sleep(100);
 		}
@@ -722,4 +750,5 @@ public class DirectMessageListenerContainerIntegrationTests {
 		}
 
 	}
+
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2020 the original author or authors.
+ * Copyright 2002-2021 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.
@@ -255,8 +255,10 @@ public class SimpleMessageListenerContainerIntegration2Tests {
 			if (event instanceof ListenerContainerConsumerFailedEvent) {
 				eventRef.set((ListenerContainerConsumerFailedEvent) event);
 			}
-			events.add((AmqpEvent) event);
-			eventLatch.countDown();
+			if (!(event instanceof MissingQueueEvent)) {
+				events.add((AmqpEvent) event);
+				eventLatch.countDown();
+			}
 		});
 		container.start();
 		for (int i = 0; i < 10; i++) {
@@ -544,6 +546,7 @@ public class SimpleMessageListenerContainerIntegration2Tests {
 		ConnectionFactory connectionFactory = new CachingConnectionFactory("localhost", BrokerTestUtils.getPort());
 
 		CountDownLatch latch = new CountDownLatch(1);
+		CountDownLatch missingLatch = new CountDownLatch(1);
 		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
 		container.setMessageListener(new MessageListenerAdapter(new PojoListener(latch)));
 		container.setQueues(queue);
@@ -552,9 +555,16 @@ public class SimpleMessageListenerContainerIntegration2Tests {
 		container.setDeclarationRetries(1);
 		container.setFailedDeclarationRetryInterval(100);
 		container.setRetryDeclarationInterval(30000);
+		container.setApplicationEventPublisher(event -> {
+			if (event instanceof MissingQueueEvent) {
+				missingLatch.countDown();
+			}
+		});
 		container.afterPropertiesSet();
 		container.start();
 
+		assertTrue(missingLatch.await(10, TimeUnit.SECONDS));
+
 		new RabbitAdmin(connectionFactory).declareQueue(queue);
 		this.template.convertAndSend(queue.getName(), "foo");
 

src/reference/asciidoc/amqp.adoc

@@ -1937,6 +1937,7 @@ Several other events are published at various stages of the container lifecycle:
 * `AsyncConsumerStoppedEvent`: When the consumer is stopped - `SimpleMessageListenerContainer` only.
 * `ConsumeOkEvent`: When a `consumeOk` is received from the broker, contains the queue name and `consumerTag`
 * `ListenerContainerIdleEvent`: See <<idle-containers>>.
+* `MissingQueueEvent`: When a missing queue is detected.
 
 [[consumerTags]]
 ===== Consumer Tags