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__

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;
 		}
 

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<String> 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);

src/reference/asciidoc/amqp.adoc

@@ -4402,6 +4402,18 @@ See <<listener-concurrency>>.
 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 <<threading>>.
+Default 60000 (60 seconds).
+
+a| image::images/tickmark.png[]
+a|
+
 | startConsumerMin
 Interval
 (min-start-interval)