GH-9521: Make middle-flow endpoints started later

Fixes: #9521

Author: artembilan

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

@@ -416,17 +416,9 @@ private void smartLifecycle() {
 		if (this.autoStartup != null) {
 			this.endpoint.setAutoStartup(this.autoStartup);
 		}
-		int phaseToSet = this.phase;
-		if (!this.isPhaseSet) {
-			if (this.endpoint instanceof PollingConsumer) {
-				phaseToSet = Integer.MAX_VALUE / 2;
-			}
-			else {
-				phaseToSet = Integer.MIN_VALUE;
-			}
+		if (this.isPhaseSet) {
+			this.endpoint.setPhase(this.phase);
 		}
-
-		this.endpoint.setPhase(phaseToSet);
 	}
 
 

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

@@ -45,7 +45,12 @@ public EventDrivenConsumer(SubscribableChannel inputChannel, MessageHandler hand
 		Assert.notNull(handler, "handler must not be null");
 		this.inputChannel = inputChannel;
 		this.handler = handler;
-		setPhase(Integer.MIN_VALUE);
+		if (this.handler instanceof MessageProducer) {
+			setPhase(Integer.MIN_VALUE + 1000);
+		}
+		else {
+			setPhase(Integer.MIN_VALUE);
+		}
 	}
 
 	@Override

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

@@ -16,8 +16,7 @@
 
 package org.springframework.integration.config.xml;
 
-import org.junit.Test;
-import org.junit.runner.RunWith;
+import org.junit.jupiter.api.Test;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
@@ -29,8 +28,8 @@
 import org.springframework.integration.handler.advice.AbstractRequestHandlerAdvice;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
-import org.springframework.test.context.ContextConfiguration;
-import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.fail;
@@ -39,11 +38,10 @@
  * @author Mark Fisher
  * @author Gary Russell
  * @author Artem Bilan
- *
  * @since 2.0
  */
-@ContextConfiguration
-@RunWith(SpringJUnit4ClassRunner.class)
+@SpringJUnitConfig
+@DirtiesContext
 public class ServiceActivatorParserTests {
 
 	@Autowired
@@ -211,7 +209,7 @@ public void failMethodAndExpressionElement() {
 
 	@Test
 	public void testConsumerEndpointFactoryBeanDefaultPhase() {
-		assertThat(this.testAliasEndpoint.getPhase()).isEqualTo(Integer.MIN_VALUE);
+		assertThat(this.testAliasEndpoint.getPhase()).isEqualTo(Integer.MIN_VALUE + 1000);
 	}
 
 	private Object sendAndReceive(MessageChannel channel, Object payload) {

spring-integration-core/src/test/java/org/springframework/integration/dsl/correlation/CorrelationHandlerTests.java

@@ -38,6 +38,7 @@
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
 import org.springframework.integration.aggregator.FluxAggregatorMessageHandler;
 import org.springframework.integration.aggregator.HeaderAttributeCorrelationStrategy;
+import org.springframework.integration.aggregator.MessageCountReleaseStrategy;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.config.EnableIntegration;
 import org.springframework.integration.dsl.IntegrationFlow;
@@ -47,6 +48,8 @@
 import org.springframework.integration.dsl.context.IntegrationFlowContext;
 import org.springframework.integration.handler.MessageTriggerAction;
 import org.springframework.integration.json.ObjectToJsonTransformer;
+import org.springframework.integration.store.MessageGroupStore;
+import org.springframework.integration.store.SimpleMessageStore;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
@@ -60,7 +63,6 @@
 /**
  * @author Artem Bilan
  * @author Gary Russell
- *
  * @since 5.0
  */
 @SpringJUnitConfig
@@ -320,6 +322,23 @@ public IntegrationFlow releaseBarrierFlow(MessageTriggerAction barrierTriggerAct
 					.get();
 		}
 
+		@Bean
+		IntegrationFlow purgeOrphanedGroupsDoesNotFailStartupFlow() {
+			MessageGroupStore messageStore = new SimpleMessageStore();
+			// Add two messages that 'complete' the aggregation immediately on startup
+			messageStore.addMessagesToGroup("test", new GenericMessage<>("1"));
+			messageStore.addMessagesToGroup("test", new GenericMessage<>("2"));
+
+			return f -> f.aggregate((a) -> a
+							.messageStore(messageStore)
+							.id("purgeOrphanedGroups")
+							.expireTimeout(1) // Expire immediately
+							// The group above is actually 'complete' on startup
+							.releaseStrategy(new MessageCountReleaseStrategy(2)))
+					.channel("purgeOrphanedGroupsChannel")
+					.handle(message -> { }, e -> e.id("endOfFlowEndpoint"));
+		}
+
 	}
 
 	record TestSplitterPojo(List<String> first, List<String> second) {