INT-4513: Clean IntManageConf from removed beans

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

The `IntegrationManagementConfigurer` keeps metrics beans in its local
stores.
When we add beans at runtime, they are not removed from those caches
after their removal

* Implement `DestructionAwareBeanPostProcessor` in the `IntegrationManagementConfigurer`
and clean up caches according provided bean type and its name
* Improve `ManualFlowTests` and `FtpTests` to be sure that
`IntegrationManagementConfigurer` caches are cleared after destroying
 `IntegrationFlowRegistration`

**Cherry-pick to 5.0.x**

Author: artembilan

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

@@ -27,7 +27,7 @@
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.BeanNameAware;
 import org.springframework.beans.factory.SmartInitializingSingleton;
-import org.springframework.beans.factory.config.BeanPostProcessor;
+import org.springframework.beans.factory.config.DestructionAwareBeanPostProcessor;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
 import org.springframework.integration.core.MessageSource;
@@ -64,8 +64,9 @@
  * @since 4.2
  *
  */
-public class IntegrationManagementConfigurer implements SmartInitializingSingleton, ApplicationContextAware,
-		BeanNameAware, BeanPostProcessor {
+public class IntegrationManagementConfigurer
+		implements SmartInitializingSingleton, ApplicationContextAware, BeanNameAware,
+				DestructionAwareBeanPostProcessor {
 
 	private static final Log logger = LogFactory.getLog(IntegrationManagementConfigurer.class);
 
@@ -246,7 +247,8 @@ public void afterSingletonsInstantiated() {
 			this.metricsFactory = new DefaultMetricsFactory();
 		}
 		this.sourceConfigurers.putAll(this.applicationContext.getBeansOfType(MessageSourceMetricsConfigurer.class));
-		Map<String, IntegrationManagement> managed = this.applicationContext.getBeansOfType(IntegrationManagement.class);
+		Map<String, IntegrationManagement> managed = this.applicationContext
+				.getBeansOfType(IntegrationManagement.class);
 		for (Entry<String, IntegrationManagement> entry : managed.entrySet()) {
 			IntegrationManagement bean = entry.getValue();
 			if (!bean.getOverrides().loggingConfigured) {
@@ -259,7 +261,8 @@ public void afterSingletonsInstantiated() {
 	}
 
 	private void injectCaptor() {
-		Map<String, IntegrationManagement> managed = this.applicationContext.getBeansOfType(IntegrationManagement.class);
+		Map<String, IntegrationManagement> managed = this.applicationContext
+				.getBeansOfType(IntegrationManagement.class);
 		for (Entry<String, IntegrationManagement> entry : managed.entrySet()) {
 			IntegrationManagement bean = entry.getValue();
 			if (!bean.getOverrides().loggingConfigured) {
@@ -280,6 +283,30 @@ public Object postProcessAfterInitialization(Object bean, String beanName) throw
 		return bean;
 	}
 
+	@Override
+	public boolean requiresDestruction(Object bean) {
+		return bean instanceof MessageChannelMetrics ||
+				bean instanceof MessageHandlerMetrics ||
+				bean instanceof MessageSourceMetrics;
+	}
+
+	@Override
+	public void postProcessBeforeDestruction(Object bean, String beanName) throws BeansException {
+		if (bean instanceof MessageChannelMetrics) {
+			this.channelsByName.remove(beanName);
+		}
+		else if (bean instanceof MessageHandlerMetrics) {
+			if (this.handlersByName.remove(beanName) == null) {
+				this.handlersByName.remove(beanName + ".handler");
+			}
+		}
+		else if (bean instanceof MessageSourceMetrics) {
+			if (this.sourcesByName.remove(beanName) == null) {
+				this.sourcesByName.remove(beanName + ".source");
+			}
+		}
+	}
+
 	private Object doConfigureMetrics(Object bean, String name) {
 		if (bean instanceof MessageChannelMetrics) {
 			configureChannelMetrics(name, (MessageChannelMetrics) bean);

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

@@ -60,7 +60,9 @@
 import org.springframework.context.annotation.Scope;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.config.EnableIntegration;
+import org.springframework.integration.config.EnableIntegrationManagement;
 import org.springframework.integration.config.EnableMessageHistory;
+import org.springframework.integration.config.IntegrationManagementConfigurer;
 import org.springframework.integration.core.MessageProducer;
 import org.springframework.integration.core.MessagingTemplate;
 import org.springframework.integration.dsl.IntegrationFlow;
@@ -114,6 +116,9 @@ public class ManualFlowTests {
 	@Autowired
 	private SmartLifecycleRoleController roleController;
 
+	@Autowired
+	private IntegrationManagementConfigurer integrationManagementConfigurer;
+
 	@Test
 	@SuppressWarnings("unchecked")
 	public void testWithAnonymousMessageProducerStart() {
@@ -128,9 +133,11 @@ protected void doStart() {
 
 		};
 		QueueChannel channel = new QueueChannel();
+		channel.setBeanName("channel");
 		IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(producer);
 
 		BridgeHandler bridgeHandler = new BridgeHandler();
+		bridgeHandler.setBeanName("bridge");
 
 		IntegrationFlow flow =
 				flowBuilder.handle(bridgeHandler)
@@ -145,9 +152,15 @@ protected void doStart() {
 
 		assertTrue(replyProducers.contains(bridgeHandler));
 
+		assertNotNull(this.integrationManagementConfigurer.getChannelMetrics("channel"));
+		assertNotNull(this.integrationManagementConfigurer.getHandlerMetrics("bridge"));
+
 		flowRegistration.destroy();
 
 		assertFalse(replyProducers.contains(bridgeHandler));
+
+		assertNull(this.integrationManagementConfigurer.getChannelMetrics("channel"));
+		assertNull(this.integrationManagementConfigurer.getHandlerMetrics("bridge"));
 	}
 
 	@Test
@@ -504,6 +517,7 @@ public void testConcurrentRegistration() throws InterruptedException {
 	@Configuration
 	@EnableIntegration
 	@EnableMessageHistory
+	@EnableIntegrationManagement
 	public static class RootConfiguration {
 
 		@Bean

spring-integration-ftp/src/test/java/org/springframework/integration/ftp/dsl/FtpTests.java

@@ -46,6 +46,8 @@
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.config.EnableIntegration;
+import org.springframework.integration.config.EnableIntegrationManagement;
+import org.springframework.integration.config.IntegrationManagementConfigurer;
 import org.springframework.integration.core.MessageSource;
 import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.dsl.IntegrationFlows;
@@ -85,6 +87,9 @@ public class FtpTests extends FtpTestSupport {
 	@Autowired
 	private ApplicationContext context;
 
+	@Autowired
+	private IntegrationManagementConfigurer integrationManagementConfigurer;
+
 	@Test
 	public void testFtpInboundFlow() throws IOException {
 		QueueChannel out = new QueueChannel();
@@ -132,7 +137,12 @@ public void testFtpInboundFlow() throws IOException {
 
 		MessageSource<?> source = context.getBean(FtpInboundFileSynchronizingMessageSource.class);
 		assertThat(TestUtils.getPropertyValue(source, "maxFetchSize"), equalTo(10));
+
+		assertNotNull(this.integrationManagementConfigurer.getSourceMetrics("ftpInboundAdapter.source"));
+
 		registration.destroy();
+
+		assertNull(this.integrationManagementConfigurer.getSourceMetrics("ftpInboundAdapter.source"));
 	}
 
 	@Test
@@ -221,6 +231,7 @@ public void testFtpMgetFlow() {
 
 	@Configuration
 	@EnableIntegration
+	@EnableIntegrationManagement
 	public static class ContextConfiguration {
 
 	}