From ad01c4498023b2b958b1a3048d1eb9a9cedf34b5 Mon Sep 17 00:00:00 2001
From: Artem Bilan <abilan@vmware.com>
Date: Wed, 4 Oct 2023 11:49:28 -0400
Subject: [PATCH] Fix maxMessagesPerPoll for SourcePollingChAdapter (#8747)

* Fix maxMessagesPerPoll for SourcePollingChAdapter

The `AbstractMethodAnnotationPostProcessor` does not check
for `PollerMetadata.MAX_MESSAGES_UNBOUNDED` before setting
`maxMessagesPerPoll` into a `SourcePollingChannelAdapter`
which in this case must be `1`

Also fix `SourcePollingChannelAdapterFactoryBean` to not mutate
the provided `PollerMetadata` (which might be global default)
with a new `maxMessagesPerPoll`

**Cherry-pick to `6.1.x` & `6.0.x`**

* * Fix `this.` prefix in `SourcePollingChannelAdapterFactoryBean`
---
 .../config/AbstractMethodAnnotationPostProcessor.java    | 9 ++++++++-
 .../config/SourcePollingChannelAdapterFactoryBean.java   | 7 ++++---
 .../configuration/EnableIntegrationTests.java            | 8 ++++++--
 3 files changed, 18 insertions(+), 6 deletions(-)

diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/AbstractMethodAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/AbstractMethodAnnotationPostProcessor.java
index dd452ba1a4..ee4cdda7a0 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/config/AbstractMethodAnnotationPostProcessor.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/config/AbstractMethodAnnotationPostProcessor.java
@@ -687,7 +687,14 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
 		pollingEndpoint.setTaskExecutor(pollerMetadata.getTaskExecutor());
 		pollingEndpoint.setTrigger(pollerMetadata.getTrigger());
 		pollingEndpoint.setAdviceChain(pollerMetadata.getAdviceChain());
-		pollingEndpoint.setMaxMessagesPerPoll(pollerMetadata.getMaxMessagesPerPoll());
+		long maxMessagesPerPoll = pollerMetadata.getMaxMessagesPerPoll();
+		if (maxMessagesPerPoll == PollerMetadata.MAX_MESSAGES_UNBOUNDED &&
+				pollingEndpoint instanceof SourcePollingChannelAdapter) {
+			// the default is 1 since a source might return
+			// a non-null and non-interruptible value every time it is invoked
+			maxMessagesPerPoll = 1;
+		}
+		pollingEndpoint.setMaxMessagesPerPoll(maxMessagesPerPoll);
 		pollingEndpoint.setErrorHandler(pollerMetadata.getErrorHandler());
 		if (pollingEndpoint instanceof PollingConsumer) {
 			((PollingConsumer) pollingEndpoint).setReceiveTimeout(pollerMetadata.getReceiveTimeout());
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java
index b273a893a7..365be7cbca 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java
@@ -186,12 +186,13 @@ public class SourcePollingChannelAdapterFactoryBean implements FactoryBean<Sourc
 				Assert.notNull(this.pollerMetadata, () -> "No poller has been defined for channel-adapter '"
 						+ this.beanName + "', and no default poller is available within the context.");
 			}
-			if (this.pollerMetadata.getMaxMessagesPerPoll() == Integer.MIN_VALUE) {
+			long maxMessagesPerPoll = this.pollerMetadata.getMaxMessagesPerPoll();
+			if (maxMessagesPerPoll == PollerMetadata.MAX_MESSAGES_UNBOUNDED) {
 				// the default is 1 since a source might return
 				// a non-null and non-interruptible value every time it is invoked
-				this.pollerMetadata.setMaxMessagesPerPoll(1);
+				maxMessagesPerPoll = 1;
 			}
-			spca.setMaxMessagesPerPoll(this.pollerMetadata.getMaxMessagesPerPoll());
+			spca.setMaxMessagesPerPoll(maxMessagesPerPoll);
 			if (this.sendTimeout != null) {
 				spca.setSendTimeout(this.sendTimeout);
 			}
diff --git a/spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java b/spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java
index 0af3d08404..9aacd3d41e 100644
--- a/spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java
+++ b/spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java
@@ -105,6 +105,7 @@ import org.springframework.integration.endpoint.EventDrivenConsumer;
 import org.springframework.integration.endpoint.MethodInvokingMessageSource;
 import org.springframework.integration.endpoint.PollingConsumer;
 import org.springframework.integration.endpoint.ReactiveStreamsConsumer;
+import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
 import org.springframework.integration.expression.SpelPropertyAccessorRegistrar;
 import org.springframework.integration.gateway.GatewayProxyFactoryBean;
 import org.springframework.integration.handler.ServiceActivatingHandler;
@@ -417,12 +418,15 @@ public class EnableIntegrationTests {
 
 		assertThat(this.counterChannel.receive(10)).isNull();
 
-		SmartLifecycle countSA = this.context.getBean("annotationTestService.count.inboundChannelAdapter",
-				SmartLifecycle.class);
+		SourcePollingChannelAdapter countSA =
+				this.context.getBean("annotationTestService.count.inboundChannelAdapter",
+						SourcePollingChannelAdapter.class);
 		assertThat(countSA.isAutoStartup()).isFalse();
 		assertThat(countSA.getPhase()).isEqualTo(23);
 		countSA.start();
 
+		assertThat(countSA.getMaxMessagesPerPoll()).isEqualTo(1);
+
 		for (int i = 0; i < 10; i++) {
 			Message<?> message = this.counterChannel.receive(10_000);
 			assertThat(message).isNotNull();