Do not block by default (#8580)

Currently, many timeouts in the project are like `-1` or other negative value
with a meaning to wait indefinitely.

According to distributed systems design and bad demo developing experience
it is not OK to block forever.

* Rework most of the timeouts in the framework to be `30` seconds.
Only one remained as `1` seconds is a `PollingConsumer` where it is
better to not block even for those 30 seconds when no messages in the queue,
but let the polling task be rescheduled.
* Remove the `MessagingGatewaySupport.replyTimeout` propagation down to the
`PollingConsumer` correlator where it was a `-1` before and blocked
the polling thread on the `Queue.poll()`.
This fixed the problem with a single thread in a pool for auto-configured `TaskScheduler`.
Now with 1 seconds wait time we are able to switch to other scheduled tasks
even with only 1 thread in the pool

Author: artembilan

spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp.xsd

@@ -721,7 +721,7 @@ property set to TRUE.
 						</xsd:appinfo>
 					</xsd:annotation>
 				</xsd:attribute>
-				<xsd:attribute name="reply-timeout" type="xsd:string">
+				<xsd:attribute name="reply-timeout" type="xsd:string" default="30000">
 					<xsd:annotation>
 						<xsd:documentation><![CDATA[
 								Allows you to specify how long this gateway will wait for
@@ -738,9 +738,7 @@ property set to TRUE.
 								The "reply-timeout" attribute maps to the "sendTimeout" property of the
 								underlying 'MessagingTemplate' instance (org.springframework.integration.core.MessagingTemplate).
 
-								The attribute will default, if not specified, to '-1', meaning that
-								by default, the Gateway will wait indefinitely. The value is
-								specified in milliseconds.
+								The value is specified in milliseconds.
 							]]></xsd:documentation>
 					</xsd:annotation>
 				</xsd:attribute>
@@ -752,7 +750,7 @@ property set to TRUE.
 		which itself is a Map.
 		This can only be provided if the 'header-mapper' reference is not being set directly. The values in
 		this list can also be simple patterns to be matched against the header names (e.g. "foo*" or "*foo").
-		A special token 'STANDARD_REPLY_HEADERS' represents all the standard AMQP headers (replyTo, correlationId etc);
+		A special token 'STANDARD_REPLY_HEADERS' represents all the standard AMQP headers (replyTo, correlationId etc.);
 		it is included by default. If you wish to add your own headers, you must also include this token if you wish the
 		standard headers to also be mapped. To map all non-standard headers the 'NON_STANDARD_HEADERS' token can be used.
 							]]></xsd:documentation>

spring-integration-core/src/main/java/org/springframework/integration/annotation/Aggregator.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2022 the original author or authors.
+ * Copyright 2002-2023 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.
@@ -61,7 +61,7 @@
 	/**
 	 * Specify the maximum amount of time in milliseconds to wait when sending a reply
 	 * {@link org.springframework.messaging.Message} to the {@link #outputChannel()}.
-	 * Defaults to {@code -1} - blocking indefinitely.
+	 * Defaults to {@code 30} seconds.
 	 * It is applied only if the output channel has some 'sending' limitations, e.g.
 	 * {@link org.springframework.integration.channel.QueueChannel} with
 	 * a fixed 'capacity' and is currently full.
@@ -78,7 +78,7 @@
 	 * or {@code replyChannel} from message headers. Messages are expired when their containing
 	 * {@link org.springframework.integration.store.MessageGroup} expires. One of the ways of expiring MessageGroups
 	 * is by configuring a {@link org.springframework.integration.store.MessageGroupStoreReaper}.
-	 * However MessageGroups can alternatively be expired by simply calling
+	 * However, MessageGroups can alternatively be expired by simply calling
 	 * {@code MessageGroupStore.expireMessageGroup(groupId)}. That could be accomplished via a ControlBus operation
 	 * or by simply invoking that method if you have a reference to the
 	 * {@link org.springframework.integration.store.MessageGroupStore} instance.

spring-integration-core/src/main/java/org/springframework/integration/annotation/Filter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2022 the original author or authors.
+ * Copyright 2002-2023 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.
@@ -100,7 +100,7 @@
 	/**
 	 * Specify the maximum amount of time in milliseconds to wait when sending a reply
 	 * {@link org.springframework.messaging.Message} to the {@link #outputChannel()}.
-	 * Defaults to {@code -1} - blocking indefinitely.
+	 * Defaults to {@code 30} seconds.
 	 * It is applied only if the output channel has some 'sending' limitations, e.g.
 	 * {@link org.springframework.integration.channel.QueueChannel} with
 	 * fixed a 'capacity'. In this case a {@link org.springframework.messaging.MessageDeliveryException} is thrown.

spring-integration-core/src/main/java/org/springframework/integration/annotation/Gateway.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2022 the original author or authors.
+ * Copyright 2002-2023 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.
@@ -22,6 +22,8 @@
 import java.lang.annotation.RetentionPolicy;
 import java.lang.annotation.Target;
 
+import org.springframework.integration.context.IntegrationContextUtils;
+
 /**
  * Indicates that an interface method is capable of mapping its parameters
  * to a message or message payload. These method-level annotations are detected
@@ -82,7 +84,7 @@
 	 * @return the timeout.
 	 * @see #requestTimeoutExpression()
 	 */
-	long requestTimeout() default Long.MIN_VALUE;
+	long requestTimeout() default IntegrationContextUtils.DEFAULT_TIMEOUT;
 
 	/**
 	 * Specify a SpEL Expression to determine the timeout (ms) when sending to the request
@@ -101,10 +103,10 @@
 	 * @return the timeout.
 	 * @see #replyTimeoutExpression()
 	 */
-	long replyTimeout() default Long.MIN_VALUE;
+	long replyTimeout() default IntegrationContextUtils.DEFAULT_TIMEOUT;
 
 	/**
-	 * Specify a SpEL Expression to determine the the time (ms) that the thread sending
+	 * Specify a SpEL Expression to determine the time (ms) that the thread sending
 	 * the request will wait for a reply. The timer starts when the thread returns to the
 	 * gateway, not when the request message is sent. Overrides the encompassing gateway's
 	 * default reply timeout. Overrides {@link #replyTimeout()}.

spring-integration-core/src/main/java/org/springframework/integration/annotation/MessagingGateway.java

@@ -22,6 +22,7 @@
 import java.lang.annotation.Target;
 
 import org.springframework.core.annotation.AliasFor;
+import org.springframework.integration.context.IntegrationContextUtils;
 
 /**
  * A stereotype annotation to provide an Integration Messaging Gateway Proxy
@@ -103,17 +104,17 @@
 	 * See {@link Gateway#requestTimeout()} for per-method configuration.
 	 * @return the suggested timeout in milliseconds, if any
 	 */
-	String defaultRequestTimeout() default "-9223372036854775808";
+	String defaultRequestTimeout() default IntegrationContextUtils.DEFAULT_TIMEOUT_STRING;
 
 	/**
 	 * Allows to specify how long this gateway will wait for the reply {@code Message}
-	 * before returning. By default, it will wait indefinitely. {@code null} is returned if
-	 * the gateway times out. Value is specified in milliseconds; it can be a simple long
+	 * before returning. The {@code null} is returned if the gateway times out.
+	 * Value is specified in milliseconds; it can be a simple long
 	 * value or a SpEL expression; array variable #args is available.
 	 * See {@link Gateway#replyTimeout()} for per-method configuration.
 	 * @return the suggested timeout in milliseconds, if any
 	 */
-	String defaultReplyTimeout() default "-9223372036854775808";
+	String defaultReplyTimeout() default IntegrationContextUtils.DEFAULT_TIMEOUT_STRING;
 
 	/**
 	 * Provide a reference to an implementation of {@link java.util.concurrent.Executor}

spring-integration-core/src/main/java/org/springframework/integration/annotation/Router.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2022 the original author or authors.
+ * Copyright 2002-2023 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.
@@ -123,7 +123,7 @@
 	/**
 	 * Specify the maximum amount of time in milliseconds to wait when sending a reply
 	 * {@link org.springframework.messaging.Message} to the {@code outputChannel}.
-	 * Defaults to {@code -1} - blocking indefinitely.
+	 * Defaults to {@code 30} seconds.
 	 * It is applied only if the output channel has some 'sending' limitations, e.g.
 	 * {@link org.springframework.integration.channel.QueueChannel} with
 	 * fixed a 'capacity'. In this case a {@link org.springframework.messaging.MessageDeliveryException} is thrown.

spring-integration-core/src/main/java/org/springframework/integration/annotation/ServiceActivator.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2022 the original author or authors.
+ * Copyright 2002-2023 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.
@@ -86,7 +86,7 @@
 	/**
 	 * Specify the maximum amount of time in milliseconds to wait when sending a reply
 	 * {@link org.springframework.messaging.Message} to the {@code outputChannel}.
-	 * Defaults to {@code -1} - blocking indefinitely.
+	 * Defaults to {@code 30} seconds.
 	 * It is applied only if the output channel has some 'sending' limitations, e.g.
 	 * {@link org.springframework.integration.channel.QueueChannel} with
 	 * fixed a 'capacity'. In this case a {@link org.springframework.messaging.MessageDeliveryException} is thrown.

spring-integration-core/src/main/java/org/springframework/integration/annotation/Splitter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2022 the original author or authors.
+ * Copyright 2002-2023 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.
@@ -88,7 +88,7 @@
 	/**
 	 * Specify the maximum amount of time in milliseconds to wait when sending a reply
 	 * {@link org.springframework.messaging.Message} to the {@code outputChannel}.
-	 * Defaults to {@code -1} - blocking indefinitely.
+	 * Defaults to {@code 30} seconds.
 	 * It is applied only if the output channel has some 'sending' limitations, e.g.
 	 * {@link org.springframework.integration.channel.QueueChannel} with
 	 * fixed a 'capacity'. In this case a {@link org.springframework.messaging.MessageDeliveryException} is thrown.

spring-integration-core/src/main/java/org/springframework/integration/annotation/Transformer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2022 the original author or authors.
+ * Copyright 2002-2023 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.
@@ -64,7 +64,7 @@
 	/**
 	 * Specify the maximum amount of time in milliseconds to wait when sending a reply
 	 * {@link org.springframework.messaging.Message} to the {@code outputChannel}.
-	 * Defaults to {@code -1} - blocking indefinitely.
+	 * Defaults to {@code 30} seconds.
 	 * It is applied only if the output channel has some 'sending' limitations, e.g.
 	 * {@link org.springframework.integration.channel.QueueChannel} with
 	 * fixed a 'capacity'. In this case a {@link org.springframework.messaging.MessageDeliveryException} is thrown.

spring-integration-core/src/main/java/org/springframework/integration/config/xml/DefaultRouterParser.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2023 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.
@@ -50,14 +50,14 @@ boolean hasDefaultOption() {
 	protected void postProcess(BeanDefinitionBuilder builder, Element element, ParserContext parserContext) {
 		List<Element> mappingElements = DomUtils.getChildElementsByTagName(element, "mapping");
 		if (!CollectionUtils.isEmpty(mappingElements)) {
-			ManagedMap<String, String> channelMappings = new ManagedMap<String, String>();
+			ManagedMap<String, String> channelMappings = new ManagedMap<>();
 			for (Element mappingElement : mappingElements) {
 				channelMappings.put(mappingElement.getAttribute("value"), mappingElement.getAttribute("channel"));
 			}
 			builder.addPropertyValue("channelMappings", channelMappings);
 		}
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "default-output-channel");
-		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "timeout");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "send-timeout");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "resolution-required");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "apply-sequence");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "ignore-send-failures");

spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationContextUtils.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2022 the original author or authors.
+ * Copyright 2002-2023 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.
@@ -101,6 +101,18 @@ public abstract class IntegrationContextUtils {
 
 	public static final String LIST_MESSAGE_HANDLER_FACTORY_BEAN_NAME = "integrationListMessageHandlerMethodFactory";
 
+	/**
+	 * The default timeout for blocking operations like send and receive messages.
+	 * @since 6.1
+	 */
+	public static final long DEFAULT_TIMEOUT = 30000L;
+
+	/**
+	 * A string representation for {@link #DEFAULT_TIMEOUT}, e.g. for annotation attributes.
+	 * @since 6.1
+	 */
+	public static final String DEFAULT_TIMEOUT_STRING = "" + DEFAULT_TIMEOUT;
+
 	/**
 	 * @param beanFactory BeanFactory for lookup, must not be null.
 	 * @return The {@link MetadataStore} bean whose name is "metadataStore".

spring-integration-core/src/main/java/org/springframework/integration/dsl/GatewayProxySpec.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019-2022 the original author or authors.
+ * Copyright 2019-2023 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.
@@ -145,7 +145,7 @@ public GatewayProxySpec requestTimeout(long requestTimeout) {
 
 	/**
 	 * Allows to specify how long this gateway will wait for the reply {@code Message}
-	 * before returning. By default, it will wait indefinitely. {@code null} is returned if
+	 * before returning. By default, it will wait 30 seconds. {@code null} is returned if
 	 * the gateway times out. Value is specified in milliseconds.
 	 * @param replyTimeout the timeout for replies in milliseconds.
 	 * @return current {@link GatewayProxySpec}.

spring-integration-core/src/main/java/org/springframework/integration/dsl/ScatterGatherSpec.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2020 the original author or authors.
+ * Copyright 2016-2023 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.
@@ -50,7 +50,7 @@ public ScatterGatherSpec gatherChannel(MessageChannel gatherChannel) {
 	 * Specify a timeout (in milliseconds) for the
 	 * {@link org.springframework.messaging.PollableChannel#receive(long)} operation
 	 * to wait for gathering results to output.
-	 * Defaults to {@code -1} - to wait indefinitely.
+	 * Defaults to {@code 30} seconds.
 	 * @param gatherTimeout the {@link org.springframework.messaging.PollableChannel} receive timeout.
 	 * @return the current {@link ScatterGatherSpec} instance.
 	 */

spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2022 the original author or authors.
+ * Copyright 2002-2023 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.
@@ -782,14 +782,9 @@ private Expression extractRequestTimeoutFromAnnotationOrMetadata(@Nullable Gatew
 		Expression requestTimeout = this.defaultRequestTimeout;
 
 		if (gatewayAnnotation != null) {
-			/*
-			 * INT-2636 Unspecified annotation attributes should not
-			 * override the default values supplied by explicit configuration.
-			 * There is a small risk that someone has used Long.MIN_VALUE explicitly
-			 * to indicate an indefinite timeout on a gateway method and that will
-			 * no longer work as expected; they will need to use, say, -1 instead.
-			 */
-			if (requestTimeout == null || gatewayAnnotation.requestTimeout() != Long.MIN_VALUE) {
+			if (requestTimeout == null ||
+					gatewayAnnotation.requestTimeout() != IntegrationContextUtils.DEFAULT_TIMEOUT) {
+
 				requestTimeout = new ValueExpression<>(gatewayAnnotation.requestTimeout());
 			}
 			if (StringUtils.hasText(gatewayAnnotation.requestTimeoutExpression())) {
@@ -813,14 +808,7 @@ private Expression extractReplyTimeoutFromAnnotationOrMetadata(@Nullable Gateway
 		Expression replyTimeout = this.defaultReplyTimeout;
 
 		if (gatewayAnnotation != null) {
-			/*
-			 * INT-2636 Unspecified annotation attributes should not
-			 * override the default values supplied by explicit configuration.
-			 * There is a small risk that someone has used Long.MIN_VALUE explicitly
-			 * to indicate an indefinite timeout on a gateway method and that will
-			 * no longer work as expected; they will need to use, say, -1 instead.
-			 */
-			if (replyTimeout == null || gatewayAnnotation.replyTimeout() != Long.MIN_VALUE) {
+			if (replyTimeout == null || gatewayAnnotation.replyTimeout() != IntegrationContextUtils.DEFAULT_TIMEOUT) {
 				replyTimeout = new ValueExpression<>(gatewayAnnotation.replyTimeout());
 			}
 			if (StringUtils.hasText(gatewayAnnotation.replyTimeoutExpression())) {
@@ -968,31 +956,27 @@ private void channels(@Nullable String requestChannelName, @Nullable String repl
 
 	private void timeouts(@Nullable Expression requestTimeout, @Nullable Expression replyTimeout,
 			GatewayMethodInboundMessageMapper messageMapper, MethodInvocationGateway gateway) {
-		if (requestTimeout == null) {
-			gateway.setRequestTimeout(-1);
-		}
-		else if (requestTimeout instanceof ValueExpression) {
-			Long timeout = requestTimeout.getValue(Long.class);
-			if (timeout != null) {
-				gateway.setRequestTimeout(timeout);
+		if (requestTimeout != null) {
+			if (requestTimeout instanceof ValueExpression) {
+				Long timeout = requestTimeout.getValue(Long.class);
+				if (timeout != null) {
+					gateway.setRequestTimeout(timeout);
+				}
 			}
-		}
-		else {
-			messageMapper.setSendTimeoutExpression(requestTimeout);
-		}
-		if (replyTimeout == null) {
-			gateway.setReplyTimeout(-1);
-		}
-		else if (replyTimeout instanceof ValueExpression) {
-			Long timeout = replyTimeout.getValue(Long.class);
-			if (timeout != null) {
-				gateway.setReplyTimeout(timeout);
+			else {
+				messageMapper.setSendTimeoutExpression(requestTimeout);
 			}
 		}
-		else {
-			messageMapper.setReplyTimeoutExpression(replyTimeout);
-		}
 		if (replyTimeout != null) {
+			if (replyTimeout instanceof ValueExpression) {
+				Long timeout = replyTimeout.getValue(Long.class);
+				if (timeout != null) {
+					gateway.setReplyTimeout(timeout);
+				}
+			}
+			else {
+				messageMapper.setReplyTimeoutExpression(replyTimeout);
+			}
 			gateway.setReceiveTimeoutExpression(replyTimeout);
 		}
 	}