From d9a8d1c7267a6e068aa32c77a34f70d65fc36a86 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 9 Nov 2017 10:30:02 -0500 Subject: [PATCH] AMQP-780: Configurable consumer start timeout JIRA: https://jira.spring.io/browse/AMQP-780 Log an error if a consumer doesn't start within the timeout; make the timeout configurable. This can happen if the task executor doesn't have enough threads to support the concurrency. __cherry-pick to 1.7.x__ --- .../SimpleMessageListenerContainer.java | 27 +++++++++++++++++-- ...ageListenerContainerIntegration2Tests.java | 19 +++++++++++++ src/reference/asciidoc/amqp.adoc | 12 +++++++++ 3 files changed, 56 insertions(+), 2 deletions(-) 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 5648f945..6b3107dd 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 @@ -77,6 +77,8 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta private static final long DEFAULT_STOP_CONSUMER_MIN_INTERVAL = 60000; + private static final long DEFAULT_CONSUMER_START_TIMEOUT = 60000L; + private static final int DEFAULT_CONSECUTIVE_ACTIVE_TRIGGER = 10; private static final int DEFAULT_CONSECUTIVE_IDLE_TRIGGER = 10; @@ -115,6 +117,8 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta private TransactionTemplate transactionTemplate; + private long consumerStartTimeout = DEFAULT_CONSUMER_START_TIMEOUT; + /** * Default constructor for convenient dependency injection via setters. */ @@ -381,6 +385,18 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta this.retryDeclarationInterval = retryDeclarationInterval; } + /** + * When starting a consumer, if this time (ms) elapses before the consumer starts, an + * error log is written; one possible cause would be if the + * {@link #setTaskExecutor(java.util.concurrent.Executor) taskExecutor} has + * insufficient threads to support the container concurrency. Default 60000. + * @param consumerStartTimeout the timeout. + * @since 1.7.5 + */ + public void setConsumerStartTimeout(long consumerStartTimeout) { + this.consumerStartTimeout = consumerStartTimeout; + } + /** * Avoid the possibility of not configuring the CachingConnectionFactory in sync with the number of concurrent * consumers. @@ -859,8 +875,15 @@ public class SimpleMessageListenerContainer extends AbstractMessageListenerConta * @throws TimeoutException if the consumer hasn't started * @throws InterruptedException if the consumer startup is interrupted */ - private FatalListenerStartupException getStartupException() throws TimeoutException, InterruptedException { - this.start.await(60000L, TimeUnit.MILLISECONDS); //NOSONAR - ignore return value + private FatalListenerStartupException getStartupException() throws TimeoutException, + InterruptedException { + if (!this.start.await( + SimpleMessageListenerContainer.this.consumerStartTimeout, TimeUnit.MILLISECONDS)) { + logger.error("Consumer failed to start in " + + SimpleMessageListenerContainer.this.consumerStartTimeout + + " milliseconds; does the task executor have enough threads to support the container " + + "concurrency?"); + } return this.startupException; } 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 cc6973c1..c5cc5415 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 @@ -79,6 +79,7 @@ import org.springframework.beans.factory.DisposableBean; import org.springframework.context.ApplicationEvent; import org.springframework.context.ApplicationEventPublisher; import org.springframework.context.support.GenericApplicationContext; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import com.rabbitmq.client.AMQP.Queue.DeclareOk; import com.rabbitmq.client.Channel; @@ -568,6 +569,24 @@ public class SimpleMessageListenerContainerIntegration2Tests { this.container.stop(); } + @Test + public void testTooSmallExecutor() { + this.container = createContainer((MessageListener) (m) -> { }, false, this.queue.getName()); + ThreadPoolTaskExecutor exec = new ThreadPoolTaskExecutor(); + exec.initialize(); + this.container.setTaskExecutor(exec); + this.container.setConcurrentConsumers(2); + this.container.setConsumerStartTimeout(100); + Log logger = spy(TestUtils.getPropertyValue(container, "logger", Log.class)); + new DirectFieldAccessor(container).setPropertyValue("logger", logger); + this.container.start(); + this.container.stop(); + ArgumentCaptor captor = ArgumentCaptor.forClass(String.class); + verify(logger).error(captor.capture()); + assertThat(captor.getValue(), equalTo("Consumer failed to start in 100 milliseconds; does the task " + + "executor have enough threads to support the container concurrency?")); + } + private boolean containerStoppedForAbortWithBadListener() throws InterruptedException { Log logger = spy(TestUtils.getPropertyValue(container, "logger", Log.class)); new DirectFieldAccessor(container).setPropertyValue("logger", logger); diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index be298261..f6b93427 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/src/reference/asciidoc/amqp.adoc @@ -4402,6 +4402,18 @@ See <>. a| image::images/tickmark.png[] a| +| consumerStartTimeout +(N/A) + +| The time in milliseconds to wait for a consumer thread to start. +If this time elapses an error log is written; an example of when this might happen is if a `taskExecutor` was configured that has insufficient threads to support the container `concurrentConsumers`. + +See <>. +Default 60000 (60 seconds). + +a| image::images/tickmark.png[] +a| + | startConsumerMin Interval (min-start-interval)