INT-4569: Disallow beans override in DSL (#2664)

* INT-4569: Disallow beans override in DSL

JIRA: https://jira.spring.io/browse/INT-4569

* Thorw `BeanDefinitionOverrideException` from the
`IntegrationFlowBeanPostProcessor` when it detects existing bean and it
is not the same object we try to register from the DSL
* Document limitations about `prototype` beans
* Some polishing in the DSL chapter of the docs

* * Fix algorithm in the `IntegrationFlowBeanPostProcessor.noBeanPresentForComponent()`

* * Polishing dsl.adoc
* Call `BeanFactory.initializeBean()` for existing beans if they are
`prototype`

* * Code formatting in the `ManualFlowTests`

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

@@ -16,7 +16,6 @@
 
 package org.springframework.integration.dsl.context;
 
-import java.util.Collection;
 import java.util.LinkedHashMap;
 import java.util.Map;
 
@@ -34,10 +33,12 @@ import org.springframework.beans.factory.SmartInitializingSingleton;
 import org.springframework.beans.factory.config.BeanDefinition;
 import org.springframework.beans.factory.config.BeanDefinitionCustomizer;
 import org.springframework.beans.factory.config.BeanPostProcessor;
+import org.springframework.beans.factory.config.ConfigurableBeanFactory;
 import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
 import org.springframework.beans.factory.config.EmbeddedValueResolver;
 import org.springframework.beans.factory.support.AbstractBeanDefinition;
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.support.BeanDefinitionOverrideException;
 import org.springframework.beans.factory.support.BeanDefinitionRegistry;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
@@ -163,7 +164,7 @@ public class IntegrationFlowBeanPostProcessor
 					id = flowNamePrefix + id;
 				}
 
-				if (noBeanPresentForComponent(messageHandler)) {
+				if (noBeanPresentForComponent(messageHandler, flowBeanName)) {
 					String handlerBeanName = generateBeanName(messageHandler, flowNamePrefix);
 
 					registerComponent(messageHandler, handlerBeanName, flowBeanName);
@@ -174,7 +175,7 @@ public class IntegrationFlowBeanPostProcessor
 				targetIntegrationComponents.put(endpoint, id);
 			}
 			else {
-				if (noBeanPresentForComponent(component)) {
+				if (noBeanPresentForComponent(component, flowBeanName)) {
 					if (component instanceof AbstractMessageChannel) {
 						String channelBeanName = ((AbstractMessageChannel) component).getComponentName();
 						if (channelBeanName == null) {
@@ -211,7 +212,7 @@ public class IntegrationFlowBeanPostProcessor
 						if (!CollectionUtils.isEmpty(componentsToRegister)) {
 							componentsToRegister.entrySet()
 									.stream()
-									.filter(o -> noBeanPresentForComponent(o.getKey()))
+									.filter(o -> noBeanPresentForComponent(o.getKey(), flowBeanName))
 									.forEach(o ->
 											registerComponent(o.getKey(),
 													generateBeanName(o.getKey(), flowNamePrefix, o.getValue(),
@@ -231,7 +232,7 @@ public class IntegrationFlowBeanPostProcessor
 						targetIntegrationComponents.put(pollingChannelAdapterFactoryBean, id);
 
 						MessageSource<?> messageSource = spec.get().getT2();
-						if (noBeanPresentForComponent(messageSource)) {
+						if (noBeanPresentForComponent(messageSource, flowBeanName)) {
 							String messageSourceId = id + ".source";
 							if (messageSource instanceof NamedComponent
 									&& ((NamedComponent) messageSource).getComponentName() != null) {
@@ -277,7 +278,14 @@ public class IntegrationFlowBeanPostProcessor
 					}
 				}
 				else {
-					targetIntegrationComponents.put(entry.getKey(), entry.getValue());
+					Object componentToUse = entry.getKey();
+					String beanNameToUse = entry.getValue();
+					if (StringUtils.hasText(beanNameToUse) &&
+							ConfigurableBeanFactory.SCOPE_PROTOTYPE.equals(
+									this.beanFactory.getBeanDefinition(beanNameToUse).getScope())) {
+						this.beanFactory.initializeBean(componentToUse, beanNameToUse);
+					}
+					targetIntegrationComponents.put(component, beanNameToUse);
 				}
 			}
 		}
@@ -337,7 +345,7 @@ public class IntegrationFlowBeanPostProcessor
 
 				componentsToRegister.entrySet()
 						.stream()
-						.filter(component -> noBeanPresentForComponent(component.getKey()))
+						.filter(component -> noBeanPresentForComponent(component.getKey(), beanName))
 						.forEach(component ->
 								registerComponent(component.getKey(),
 										generateBeanName(component.getKey(), component.getValue())));
@@ -378,17 +386,36 @@ public class IntegrationFlowBeanPostProcessor
 		}
 	}
 
-	private boolean noBeanPresentForComponent(Object instance) {
+	@SuppressWarnings("unchecked")
+	private boolean noBeanPresentForComponent(Object instance, String parentBeanName) {
 		if (instance instanceof NamedComponent) {
 			String beanName = ((NamedComponent) instance).getComponentName();
 			if (beanName != null) {
-				return !this.beanFactory.containsBean(beanName);
+				if (this.beanFactory.containsBean(beanName)) {
+					BeanDefinition existingBeanDefinition = this.beanFactory.getBeanDefinition(beanName);
+					if (!ConfigurableBeanFactory.SCOPE_PROTOTYPE.equals(existingBeanDefinition.getScope())
+							&& !instance.equals(this.beanFactory.getBean(beanName))) {
+
+						AbstractBeanDefinition beanDefinition =
+								BeanDefinitionBuilder.genericBeanDefinition((Class<Object>) instance.getClass(),
+										() -> instance)
+										.getBeanDefinition();
+						beanDefinition.setResourceDescription("the '" + parentBeanName + "' bean definition");
+						throw new BeanDefinitionOverrideException(beanName, beanDefinition, existingBeanDefinition);
+					}
+					else {
+						return false;
+					}
+				}
+				else {
+					return true;
+				}
 			}
 		}
 
-		Collection<?> beans = this.beanFactory.getBeansOfType(instance.getClass(), false, false).values();
-
-		return !beans.contains(instance);
+		return !this.beanFactory.getBeansOfType(instance.getClass(), false, false)
+				.values()
+				.contains(instance);
 	}
 
 	private void registerComponent(Object component, String beanName) {

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

@@ -16,6 +16,7 @@
 
 package org.springframework.integration.dsl.manualflow;
 
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.lessThan;
@@ -46,6 +47,7 @@ import org.hamcrest.Matchers;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 
+import org.springframework.beans.factory.BeanCreationException;
 import org.springframework.beans.factory.BeanCreationNotAllowedException;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.DisposableBean;
@@ -53,11 +55,13 @@ import org.springframework.beans.factory.ListableBeanFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.config.BeanFactoryPostProcessor;
 import org.springframework.beans.factory.config.ConfigurableBeanFactory;
+import org.springframework.beans.factory.support.BeanDefinitionOverrideException;
 import org.springframework.beans.factory.support.DefaultListableBeanFactory;
 import org.springframework.context.annotation.AnnotationConfigApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Scope;
+import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.config.EnableIntegration;
 import org.springframework.integration.config.EnableIntegrationManagement;
@@ -241,7 +245,8 @@ public class ManualFlowTests {
 			assertThat(e.getMessage(), containsString("The 'receive()/receiveAndConvert()' isn't supported"));
 		}
 
-		assertThat(this.beanFactory.getBeanNamesForType(MessageTransformingHandler.class)[0], startsWith(flowId + "."));
+		assertThat(this.beanFactory.getBeanNamesForType(MessageTransformingHandler.class)[0],
+				startsWith(flowId + "."));
 
 		flowRegistration.destroy();
 
@@ -516,6 +521,18 @@ public class ManualFlowTests {
 		flowRegistrations.forEach(IntegrationFlowRegistration::destroy);
 	}
 
+
+	@Test
+	public void testDisabledBeansOverride() {
+		assertThatThrownBy(
+				() -> this.integrationFlowContext
+						.registration(f -> f.channel(c -> c.direct("doNotOverrideChannel")))
+						.register())
+				.isExactlyInstanceOf(BeanCreationException.class)
+				.hasCauseExactlyInstanceOf(BeanDefinitionOverrideException.class)
+				.hasMessageContaining("Invalid bean definition with name 'doNotOverrideChannel'");
+	}
+
 	@Configuration
 	@EnableIntegration
 	@EnableMessageHistory
@@ -533,6 +550,12 @@ public class ManualFlowTests {
 			return new Date();
 		}
 
+
+		@Bean
+		public MessageChannel doNotOverrideChannel() {
+			return new DirectChannel();
+		}
+
 	}
 
 	private static class MyFlowAdapter extends IntegrationFlowAdapter {
@@ -582,7 +605,7 @@ public class ManualFlowTests {
 		}
 
 		@Override
-		public void destroy() throws Exception {
+		public void destroy() {
 			this.destroyed = true;
 		}