Fix double start for AbstractEndpoint (#3928)

* Fix double start for `AbstractEndpoint`

When we use POJO methods in the `IntegrationFlowAdapter`,
the `IntegrationFlowAdapter` is set as a `target` for the `MessagingMethodInvokerHelper`.
When endpoint is started by the application context, such a `start()` is propagated
down to the `MessagingMethodInvokerHelper`.
And in our case back into an `IntegrationFlowAdapter` instance.
This one, in turn, starts its `IntegrationFlow` internally which leads to the
start of the mentioned endpoint in the beginning.
Therefore, we cause a recursive `start()` call on this endpoint from itself.
The `running` flag is set when we are already done with the `doStart()` logic.
Therefore a recursive `start()` call leads to two concurrent polling tasks
in the `AbstractPollingEndpoint`.

* Check also for the `active` flag in the `AbstractEndpoint.start()`
and reset it in case of exception in the `doStart()`

**Cherry-pick to `5.5.x`**

* * Change assert for message timestamps to `isCloseTo()` with percentage

* * Change `catch` in the `AbstractEndpoint.start()` to `RuntimeException`:
the `doStart()` cannot throw unchecked exceptions by definition

* * Fix imports in `AbstractEndpoint`

spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java

@@ -149,9 +149,15 @@ public abstract class AbstractEndpoint extends IntegrationObjectSupport
 	public final void start() {
 		this.lifecycleLock.lock();
 		try {
-			if (!this.running) {
+			if (!this.running && !this.active) {
 				this.active = true;
-				doStart();
+				try {
+					doStart();
+				}
+				catch (RuntimeException ex) {
+					this.active = false;
+					throw ex;
+				}
 				this.running = true;
 				logger.info(() -> "started " + this);
 			}

spring-integration-core/src/test/java/org/springframework/integration/dsl/flowservices/FlowServiceTests.java

@@ -25,6 +25,7 @@ import java.util.List;
 import java.util.Optional;
 import java.util.concurrent.atomic.AtomicReference;
 
+import org.assertj.core.data.Percentage;
 import org.junit.jupiter.api.Test;
 
 import org.springframework.aop.framework.Advised;
@@ -55,6 +56,7 @@ import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.PollableChannel;
 import org.springframework.messaging.handler.annotation.Header;
+import org.springframework.messaging.support.GenericMessage;
 import org.springframework.scheduling.TriggerContext;
 import org.springframework.stereotype.Component;
 import org.springframework.test.annotation.DirtiesContext;
@@ -118,6 +120,34 @@ public class FlowServiceTests {
 		assertThat(message.getPayload()).isEqualTo("FOO");
 	}
 
+	@Autowired
+	@Qualifier("delaysBetweenPollsInput")
+	private MessageChannel delaysBetweenPollsInput;
+
+	@Autowired
+	@Qualifier("delaysBetweenPollsOutput")
+	private PollableChannel delaysBetweenPollsOutput;
+
+	@Test
+	public void noDoubleStartForEndpoints() {
+		this.delaysBetweenPollsInput.send(new GenericMessage<>("A,B"));
+
+		Message<?> receive1 = this.delaysBetweenPollsOutput.receive(10_000);
+
+		assertThat(receive1).isNotNull()
+				.extracting(Message::getPayload)
+				.isEqualTo("A");
+
+		Message<?> receive2 = this.delaysBetweenPollsOutput.receive(10_000);
+
+		assertThat(receive2).isNotNull()
+				.extracting(Message::getPayload)
+				.isEqualTo("B");
+
+		assertThat(receive2.getHeaders().getTimestamp() - receive1.getHeaders().getTimestamp())
+				.isCloseTo(500, Percentage.withPercentage(10));
+	}
+
 	@Configuration
 	@EnableIntegration
 	@ComponentScan
@@ -226,4 +256,23 @@ public class FlowServiceTests {
 
 	}
 
+	@Component
+	public static class DelaysBetweenPollsAdapter extends IntegrationFlowAdapter {
+
+		@ServiceActivator
+		public String handle(String payload) {
+			return payload;
+		}
+
+		@Override
+		protected IntegrationFlowDefinition<?> buildFlow() {
+			return from("delaysBetweenPollsInput")
+					.split(splitter -> splitter.delimiters(","))
+					.channel(MessageChannels.queue())
+					.handle(this, "handle", e -> e.poller(poller -> poller.fixedDelay(500).maxMessagesPerPoll(1)))
+					.channel(MessageChannels.queue("delaysBetweenPollsOutput"));
+		}
+
+	}
+
 }