Fix SourcePollingChAdFB autoStartup propagation

The `spring.integration.properties` can come with the `noAutoStartup`
property where we can specify a source polling channel adapter endpoint to
not start automatically.
Turns out the `SourcePollingChannelAdapterFactoryBean` propagates its
`autoStartup` property unconditionally which will skip the `noAutoStartup` value
because an `AbstractEndpoint.setAutoStartup()` sets an `autoStartupSetExplicitly` state

* Fix `SourcePollingChannelAdapterFactoryBean` to rely on a `Boolean` object state
and don't call target endpoint `setAutoStartup()` if it was not set
* Adjust `spring.integration.properties` in tests to use `noAutoStartup`
for some `SourcePollingChannelAdapterFactoryBean`
* Verify that property was applied in the `IntegrationFlowTests.testWithSupplierMessageSourceImpliedPoller()`

**Cherry-pick to `5.4.x` & `5.3.x`**

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2021 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.
@@ -55,7 +55,7 @@ public class SourcePollingChannelAdapterFactoryBean implements FactoryBean<Sourc
 
 	private PollerMetadata pollerMetadata;
 
-	private boolean autoStartup = true;
+	private Boolean autoStartup;
 
 	private int phase = Integer.MAX_VALUE / 2;
 
@@ -96,7 +96,7 @@ public class SourcePollingChannelAdapterFactoryBean implements FactoryBean<Sourc
 		this.pollerMetadata = pollerMetadata;
 	}
 
-	public void setAutoStartup(boolean autoStartup) {
+	public void setAutoStartup(Boolean autoStartup) {
 		this.autoStartup = autoStartup;
 	}
 
@@ -178,7 +178,7 @@ public class SourcePollingChannelAdapterFactoryBean implements FactoryBean<Sourc
 
 			if (this.pollerMetadata == null) {
 				this.pollerMetadata = PollerMetadata.getDefaultPollerMetadata(this.beanFactory);
-				Assert.notNull(this.pollerMetadata, "No poller has been defined for channel-adapter '"
+				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) {
@@ -195,7 +195,9 @@ public class SourcePollingChannelAdapterFactoryBean implements FactoryBean<Sourc
 			spca.setTrigger(this.pollerMetadata.getTrigger());
 			spca.setErrorHandler(this.pollerMetadata.getErrorHandler());
 			spca.setBeanClassLoader(this.beanClassLoader);
-			spca.setAutoStartup(this.autoStartup);
+			if (this.autoStartup != null) {
+				spca.setAutoStartup(this.autoStartup);
+			}
 			spca.setPhase(this.phase);
 			spca.setRole(this.role);
 			spca.setBeanName(this.beanName);

spring-integration-core/src/test/java/org/springframework/integration/dsl/flows/IntegrationFlowTests.java

@@ -64,6 +64,7 @@ import org.springframework.integration.dsl.IntegrationFlows;
 import org.springframework.integration.dsl.MessageChannels;
 import org.springframework.integration.dsl.Pollers;
 import org.springframework.integration.dsl.Transformers;
+import org.springframework.integration.endpoint.AbstractEndpoint;
 import org.springframework.integration.endpoint.EventDrivenConsumer;
 import org.springframework.integration.gateway.GatewayProxyFactoryBean;
 import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
@@ -180,8 +181,14 @@ public class IntegrationFlowTests {
 	@Qualifier("lambdasInput")
 	private MessageChannel lambdasInput;
 
+	@Autowired
+	AbstractEndpoint stringSupplierEndpoint;
+
 	@Test
 	public void testWithSupplierMessageSourceImpliedPoller() {
+		assertThat(this.stringSupplierEndpoint.isAutoStartup()).isFalse();
+		assertThat(this.stringSupplierEndpoint.isRunning()).isFalse();
+		this.stringSupplierEndpoint.start();
 		assertThat(this.suppliedChannel.receive(10000).getPayload()).isEqualTo("FOO");
 	}
 
@@ -563,7 +570,7 @@ public class IntegrationFlowTests {
 
 		@Bean
 		public IntegrationFlow supplierFlow() {
-			return IntegrationFlows.fromSupplier(stringSupplier())
+			return IntegrationFlows.fromSupplier(stringSupplier(), c -> c.id("stringSupplierEndpoint"))
 					.transform(toUpperCaseFunction())
 					.channel("suppliedChannel")
 					.get();

spring-integration-core/src/test/resources/META-INF/spring.integration.properties

@@ -3,4 +3,4 @@
 #spring.integration.channels.maxBroadcastSubscribers=1
 spring.integration.taskScheduler.poolSize=20
 spring.integration.messagingTemplate.throwExceptionOnLateReply=true
-spring.integration.endpoints.noAutoStartup=fooService*
+spring.integration.endpoints.noAutoStartup=fooService*,stringSupplierEndpoint