diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java
index 8e4653f5d9..66b1c3cab0 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java
@@ -416,17 +416,9 @@ public class ConsumerEndpointFactoryBean
 		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);
 	}
 
 
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/EventDrivenConsumer.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/EventDrivenConsumer.java
index d953e09691..f854d7c952 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/EventDrivenConsumer.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/EventDrivenConsumer.java
@@ -45,7 +45,12 @@ public class EventDrivenConsumer extends AbstractEndpoint implements Integration
 		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
diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/ServiceActivatorParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/ServiceActivatorParserTests.java
index 528a458a34..22de5342e2 100644
--- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/ServiceActivatorParserTests.java
+++ b/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.ServiceActivatingHandler;
 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 @@ import static org.assertj.core.api.Assertions.fail;
  * @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 class ServiceActivatorParserTests {
 
 	@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) {
diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/correlation/CorrelationHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/correlation/CorrelationHandlerTests.java
index 4796f3296a..04100e0ba5 100644
--- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/correlation/CorrelationHandlerTests.java
+++ b/spring-integration-core/src/test/java/org/springframework/integration/dsl/correlation/CorrelationHandlerTests.java
@@ -38,6 +38,7 @@ import org.springframework.core.task.TaskExecutor;
 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.Transformers;
 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 @@ import static org.assertj.core.api.Assertions.assertThat;
 /**
  * @author Artem Bilan
  * @author Gary Russell
- *
  * @since 5.0
  */
 @SpringJUnitConfig
@@ -320,6 +322,23 @@ public class CorrelationHandlerTests {
 					.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) {