Remove dependant beans recursively

https://build.spring.io/browse/INT-MASTER-1058/

When we destroy manually registered `IntegrationFlow`, we need to
count with sub-flows and iterate all the dependant beans recursively
for full flow removal

**Cherry-pick to 5.0.x**

spring-integration-core/src/main/java/org/springframework/integration/dsl/context/StandardIntegrationFlowContext.java

@@ -16,7 +16,6 @@
 
 package org.springframework.integration.dsl.context;
 
-import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
@@ -58,6 +57,8 @@ public final class StandardIntegrationFlowContext implements IntegrationFlowCont
 
 	private ConfigurableListableBeanFactory beanFactory;
 
+	private BeanDefinitionRegistry beanDefinitionRegistry;
+
 	private StandardIntegrationFlowContext() {
 	}
 
@@ -68,6 +69,7 @@ public final class StandardIntegrationFlowContext implements IntegrationFlowCont
 						"'ConfigurableListableBeanFactory'. " +
 						"Consider using 'GenericApplicationContext' implementation.");
 		this.beanFactory = (ConfigurableListableBeanFactory) beanFactory;
+		this.beanDefinitionRegistry = (BeanDefinitionRegistry) this.beanFactory;
 	}
 
 	/**
@@ -163,17 +165,9 @@ public final class StandardIntegrationFlowContext implements IntegrationFlowCont
 			IntegrationFlowRegistration flowRegistration = this.registry.remove(flowId);
 			flowRegistration.stop();
 
-			BeanDefinitionRegistry beanDefinitionRegistry = (BeanDefinitionRegistry) this.beanFactory;
+			removeDependantBeans(flowId);
 
-			Arrays.stream(this.beanFactory.getDependentBeans(flowId))
-					.forEach(beanName -> {
-						beanDefinitionRegistry.removeBeanDefinition(beanName);
-						// TODO until https://jira.spring.io/browse/SPR-16837
-						Arrays.asList(beanDefinitionRegistry.getAliases(beanName))
-								.forEach(beanDefinitionRegistry::removeAlias);
-					});
-
-			beanDefinitionRegistry.removeBeanDefinition(flowId);
+			this.beanDefinitionRegistry.removeBeanDefinition(flowId);
 		}
 		else {
 			throw new IllegalStateException("An IntegrationFlow with the id "
@@ -181,6 +175,19 @@ public final class StandardIntegrationFlowContext implements IntegrationFlowCont
 		}
 	}
 
+	private void removeDependantBeans(String parentName) {
+		String[] dependentBeans = this.beanFactory.getDependentBeans(parentName);
+		for (String beanName : dependentBeans) {
+			removeDependantBeans(beanName);
+			this.beanDefinitionRegistry.removeBeanDefinition(beanName);
+			// TODO until https://jira.spring.io/browse/SPR-16837
+			String[] aliases = this.beanDefinitionRegistry.getAliases(beanName);
+			for (String alias : aliases) {
+				this.beanDefinitionRegistry.removeAlias(alias);
+			}
+		}
+	}
+
 	/**
 	 * Obtain a {@link MessagingTemplate} with its default destination set to the input channel
 	 * of the {@link IntegrationFlow} for provided {@code flowId}.

spring-integration-core/src/test/java/org/springframework/integration/dsl/manualflow/ManualFlowTests.java

@@ -29,11 +29,11 @@ import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
-import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Date;
 import java.util.List;
 import java.util.Objects;
+import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
@@ -371,7 +371,7 @@ public class ManualFlowTests {
 		assertTrue(this.roleController.getEndpointsRunningStatus(testRole).isEmpty());
 	}
 
-	//	@Test
+	@Test
 	public void testDynamicSubFlowCreation() {
 		Flux<Message<?>> messageFlux =
 				Flux.just("1,2,3,4")
@@ -393,7 +393,9 @@ public class ManualFlowTests {
 				.get();
 
 		IntegrationFlowRegistration flowRegistration =
-				this.integrationFlowContext.registration(integrationFlow).register();
+				this.integrationFlowContext.registration(integrationFlow)
+						.id("dynamicSubFlows")
+						.register();
 
 		for (int i = 0; i < 4; i++) {
 			Message<?> receive = resultChannel.receive(10_000);
@@ -437,7 +439,7 @@ public class ManualFlowTests {
 	public void testConcurrentRegistration() throws InterruptedException {
 		ExecutorService executorService = Executors.newCachedThreadPool();
 
-		List<IntegrationFlowRegistration> flowRegistrations = new ArrayList<>();
+		List<IntegrationFlowRegistration> flowRegistrations = new CopyOnWriteArrayList<>();
 
 		AtomicBoolean exceptionHappened = new AtomicBoolean();