GH-3464: Treat 0 as special for polling endpoint (#3487)

* GH-3464: Treat 0 as special for polling endpoint

Fixes https://github.com/spring-projects/spring-integration/issues/3464

The `maxMessagePerPoll <= 0` is considered as an unbound `receive()` call.
End-users would like to have a special treatment for `0` value -
skip the `receive()` call altogether for the current polling cycle.

* Change the logic for a scheduled poller to not call `pollForMessage()`
and just log an INFO when `maxMessagePerPoll == 0`
* Fix reactive poller to deal with `maxMessagePerPoll == 0` properly
* Expose `maxMessagesPerPoll` as a `@ManagedAttribute` to let it to be
modified via Control Bus and JMX
* Test and document a new behavior

* * Fix unused imports in the test class

* Fix language in docs accoridng PR review

Co-authored-by: Gary Russell <grussell@vmware.com>

Co-authored-by: Gary Russell <grussell@vmware.com>

spring-integration-core/src/test/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBeanTests.java

@@ -18,7 +18,9 @@ package org.springframework.integration.config;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.BDDMockito.willAnswer;
+import static org.mockito.Mockito.atLeastOnce;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.verify;
@@ -222,6 +224,47 @@ public class SourcePollingChannelAdapterFactoryBeanTests {
 		pollingChannelAdapter.stop();
 	}
 
+	@Test
+	public void testZeroForMaxMessagesPerPoll() throws InterruptedException {
+		ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
+		taskScheduler.afterPropertiesSet();
+
+		SourcePollingChannelAdapter pollingChannelAdapter = new SourcePollingChannelAdapter();
+		pollingChannelAdapter.setTaskScheduler(taskScheduler);
+		pollingChannelAdapter.setSource(() -> new GenericMessage<>("test"));
+		pollingChannelAdapter.setTrigger(new PeriodicTrigger(1));
+		pollingChannelAdapter.setMaxMessagesPerPoll(0);
+		QueueChannel outputChannel = new QueueChannel();
+		pollingChannelAdapter.setOutputChannel(outputChannel);
+		pollingChannelAdapter.setBeanFactory(mock(BeanFactory.class));
+
+		LogAccessor logger = spy(TestUtils.getPropertyValue(pollingChannelAdapter, "logger", LogAccessor.class));
+		new DirectFieldAccessor(pollingChannelAdapter).setPropertyValue("logger", logger);
+
+		CountDownLatch logCalledLatch = new CountDownLatch(1);
+
+		willAnswer(invocation -> {
+			logCalledLatch.countDown();
+			return invocation.callRealMethod();
+		})
+				.given(logger)
+				.info(anyString());
+
+		pollingChannelAdapter.afterPropertiesSet();
+		pollingChannelAdapter.start();
+
+		assertThat(logCalledLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		verify(logger, atLeastOnce()).info("Polling disabled while 'maxMessagesPerPoll == 0'");
+
+		pollingChannelAdapter.setMaxMessagesPerPoll(1);
+
+		Message<?> receive = outputChannel.receive(10_000);
+		assertThat(receive).isNotNull();
+		assertThat(receive.getPayload()).isEqualTo("test");
+		pollingChannelAdapter.stop();
+	}
+
+
 	private static class LifecycleMessageSource implements MessageSource<Boolean>, Lifecycle {
 
 		private volatile boolean running;

spring-integration-core/src/test/java/org/springframework/integration/endpoint/ReactiveInboundChannelAdapterTests.java

@@ -28,10 +28,12 @@ import java.util.function.Supplier;
 import org.junit.jupiter.api.Test;
 
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.core.task.SimpleAsyncTaskExecutor;
 import org.springframework.core.task.TaskExecutor;
+import org.springframework.integration.annotation.EndpointId;
 import org.springframework.integration.annotation.InboundChannelAdapter;
 import org.springframework.integration.annotation.Poller;
 import org.springframework.integration.channel.FluxMessageChannel;
@@ -56,6 +58,10 @@ public class ReactiveInboundChannelAdapterTests {
 	@Autowired
 	private FluxMessageChannel fluxChannel;
 
+	@Autowired
+	@Qualifier("counterEndpoint")
+	private AbstractPollingEndpoint abstractPollingEndpoint;
+
 	@Test
 	public void testReactiveInboundChannelAdapter() {
 		Flux<Integer> testFlux =
@@ -64,6 +70,9 @@ public class ReactiveInboundChannelAdapterTests {
 						.cast(Integer.class);
 
 		StepVerifier.create(testFlux)
+				.expectSubscription()
+				.expectNoEvent(Duration.ofSeconds(1))
+				.then(() -> abstractPollingEndpoint.setMaxMessagesPerPoll(-1))
 				.expectNext(2, 4, 6, 8, 10, 12, 14, 16)
 				.thenCancel()
 				.verify(Duration.ofSeconds(10));
@@ -109,7 +118,8 @@ public class ReactiveInboundChannelAdapterTests {
 
 		@Bean
 		@InboundChannelAdapter(value = "fluxChannel",
-				poller = @Poller(fixedDelay = "100", maxMessagesPerPoll = "-1", taskExecutor = "taskExecutor"))
+				poller = @Poller(fixedDelay = "100", maxMessagesPerPoll = "0", taskExecutor = "taskExecutor"))
+		@EndpointId("counterEndpoint")
 		public Supplier<Integer> counterMessageSupplier() {
 			return () -> {
 				int i = counter().incrementAndGet();