Introduce configuration property key "spring.integration.poller.use-application-task-executor"

Fix GH-44534

Signed-off-by: Yanming Zhou <[email protected]>

Author: quaff

spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/integration/IntegrationAutoConfiguration.java

@@ -24,6 +24,8 @@
 import io.rsocket.transport.netty.server.TcpServerTransport;
 
 import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.ObjectProvider;
+import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.boot.autoconfigure.AutoConfiguration;
 import org.springframework.boot.autoconfigure.AutoConfigureBefore;
 import org.springframework.boot.autoconfigure.condition.AnyNestedCondition;
@@ -40,6 +42,7 @@
 import org.springframework.boot.autoconfigure.jmx.JmxProperties;
 import org.springframework.boot.autoconfigure.rsocket.RSocketMessagingAutoConfiguration;
 import org.springframework.boot.autoconfigure.sql.init.OnDatabaseInitializationCondition;
+import org.springframework.boot.autoconfigure.task.TaskExecutionAutoConfiguration;
 import org.springframework.boot.autoconfigure.task.TaskSchedulingAutoConfiguration;
 import org.springframework.boot.autoconfigure.thread.Threading;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
@@ -51,6 +54,7 @@
 import org.springframework.context.annotation.Conditional;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
+import org.springframework.core.task.TaskExecutor;
 import org.springframework.integration.config.EnableIntegration;
 import org.springframework.integration.config.EnableIntegrationManagement;
 import org.springframework.integration.config.IntegrationComponentScanRegistrar;
@@ -85,6 +89,7 @@
  * @author Vedran Pavic
  * @author Madhura Bhave
  * @author Yong-Hyun Kim
+ * @author Yanming Zhou
  * @since 1.1.0
  */
 @AutoConfiguration(after = { DataSourceAutoConfiguration.class, JmxAutoConfiguration.class,
@@ -130,7 +135,8 @@ protected static class IntegrationConfiguration {
 
 		@Bean(PollerMetadata.DEFAULT_POLLER)
 		@ConditionalOnMissingBean(name = PollerMetadata.DEFAULT_POLLER)
-		public PollerMetadata defaultPollerMetadata(IntegrationProperties integrationProperties) {
+		public PollerMetadata defaultPollerMetadata(IntegrationProperties integrationProperties,
+				@Qualifier(TaskExecutionAutoConfiguration.APPLICATION_TASK_EXECUTOR_BEAN_NAME) ObjectProvider<TaskExecutor> taskExecutor) {
 			IntegrationProperties.Poller poller = integrationProperties.getPoller();
 			MutuallyExclusiveConfigurationPropertiesException.throwIfMultipleNonNullValuesIn((entries) -> {
 				entries.put("spring.integration.poller.cron",
@@ -143,6 +149,9 @@ public PollerMetadata defaultPollerMetadata(IntegrationProperties integrationPro
 			map.from(poller::getMaxMessagesPerPoll).to(pollerMetadata::setMaxMessagesPerPoll);
 			map.from(poller::getReceiveTimeout).as(Duration::toMillis).to(pollerMetadata::setReceiveTimeout);
 			map.from(poller).as(this::asTrigger).to(pollerMetadata::setTrigger);
+			if (poller.isUseApplicationTaskExecutor()) {
+				pollerMetadata.setTaskExecutor(taskExecutor.getObject());
+			}
 			return pollerMetadata;
 		}
 

spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/integration/IntegrationProperties.java

@@ -30,6 +30,7 @@
  * @author Vedran Pavic
  * @author Stephane Nicoll
  * @author Artem Bilan
+ * @author Yanming Zhou
  * @since 2.0.0
  */
 @ConfigurationProperties("spring.integration")
@@ -369,6 +370,11 @@ public static class Poller {
 		 */
 		private String cron;
 
+		/**
+		 * Whether to use the application TaskExecutor for default poller.
+		 */
+		private boolean useApplicationTaskExecutor;
+
 		public int getMaxMessagesPerPoll() {
 			return this.maxMessagesPerPoll;
 		}
@@ -417,6 +423,14 @@ public void setCron(String cron) {
 			this.cron = cron;
 		}
 
+		public boolean isUseApplicationTaskExecutor() {
+			return this.useApplicationTaskExecutor;
+		}
+
+		public void setUseApplicationTaskExecutor(boolean useApplicationTaskExecutor) {
+			this.useApplicationTaskExecutor = useApplicationTaskExecutor;
+		}
+
 	}
 
 	public static class Management {

spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/integration/IntegrationAutoConfigurationTests.java

@@ -50,6 +50,7 @@
 import org.springframework.boot.autoconfigure.rsocket.RSocketRequesterAutoConfiguration;
 import org.springframework.boot.autoconfigure.rsocket.RSocketServerAutoConfiguration;
 import org.springframework.boot.autoconfigure.rsocket.RSocketStrategiesAutoConfiguration;
+import org.springframework.boot.autoconfigure.task.TaskExecutionAutoConfiguration;
 import org.springframework.boot.autoconfigure.task.TaskSchedulingAutoConfiguration;
 import org.springframework.boot.context.properties.source.MutuallyExclusiveConfigurationPropertiesException;
 import org.springframework.boot.jdbc.init.DataSourceScriptDatabaseInitializer;
@@ -105,6 +106,7 @@
  * @author Stephane Nicoll
  * @author Vedran Pavic
  * @author Yong-Hyun Kim
+ * @author Yanming Zhou
  */
 class IntegrationAutoConfigurationTests {
 
@@ -498,6 +500,19 @@ void whenFixedRatePollerPropertyIsSetThenItIsReflectedAsFixedRatePropertyOfPerio
 			});
 	}
 
+	@Test
+	void useApplicationTaskExecutorWhenCorrespondingPollerPropertyAreSet() {
+		this.contextRunner.withConfiguration(AutoConfigurations.of(TaskExecutionAutoConfiguration.class))
+			.withUserConfiguration(PollingConsumerConfiguration.class)
+			.withPropertyValues("spring.integration.poller.use-application-task-executor=true")
+			.run((context) -> {
+				assertThat(context).hasSingleBean(PollerMetadata.class);
+				PollerMetadata metadata = context.getBean(PollerMetadata.DEFAULT_POLLER, PollerMetadata.class);
+				assertThat(metadata.getTaskExecutor())
+					.isSameAs(context.getBean(TaskExecutionAutoConfiguration.APPLICATION_TASK_EXECUTOR_BEAN_NAME));
+			});
+	}
+
 	@Test
 	void integrationManagementLoggingIsEnabledByDefault() {
 		this.contextRunner.withBean(DirectChannel.class, DirectChannel::new)