From 69ec27847d30a3c05bd5d8c33d91a843dcf4192e Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 12 Jan 2021 12:12:21 -0500 Subject: [PATCH] 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 --- .../AbstractMessageListenerContainer.java | 9 +- .../listener/BlockingQueueConsumer.java | 14 +- .../DirectMessageListenerContainer.java | 131 ++++++++++++++---- .../rabbit/listener/MissingQueueEvent.java | 57 ++++++++ .../SimpleMessageListenerContainer.java | 3 +- ...sageListenerContainerIntegrationTests.java | 35 ++++- ...ageListenerContainerIntegration2Tests.java | 16 ++- src/reference/asciidoc/amqp.adoc | 1 + 8 files changed, 229 insertions(+), 37 deletions(-) create mode 100644 spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/MissingQueueEvent.java diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java index edfad096..2d21f45c 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java +++ b/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( diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java index 47d30bef..5cdfc6af 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java +++ b/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 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 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); } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java index e9e3e6fe..b69d8b28 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java +++ b/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 consumers = new LinkedList<>(); // NOSONAR - private final List consumersToRestart = new LinkedList<>(); + private final Set consumersToRestart = new LinkedHashSet<>(); + + private final Set removedQueues = ConcurrentHashMap.newKeySet(); private final MultiValueMap 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 cBQ = this.consumersByQueue.get(queue); + int index = 0; + if (cBQ != null) { + // find a gap or set the index to the end + List 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 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 restartableConsumers = new ArrayList<>(this.consumersToRestart); - this.consumersToRestart.clear(); if (this.started) { + List 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) + "]"; } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/MissingQueueEvent.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/MissingQueueEvent.java new file mode 100644 index 00000000..e0243787 --- /dev/null +++ b/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 + "]"; + } + +} diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java index fedd3fd1..3c8388b6 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java +++ b/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); } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainerIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainerIntegrationTests.java index 4b7e2bc2..380bbbc6 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainerIntegrationTests.java +++ b/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 { } } + } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.java index 3b0f85b9..ea3f12f4 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainerIntegration2Tests.java +++ b/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"); diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index 39722be3..1ba48f39 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/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 <>. +* `MissingQueueEvent`: When a missing queue is detected. [[consumerTags]] ===== Consumer Tags