From 4a55fa31885d55c3b3bbf0e853b3a872e39d6c52 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 7 Jul 2022 15:30:43 -0400 Subject: [PATCH] Fix IntegrationMBeanExporter logic for endpoints Related to https://stackoverflow.com/questions/72851234/error-in-startup-application-when-using-serviceactivator-in-spring-cloud-stream The `MessagingAnnotationPostProcessor` register an `endpoint` bean for Messaging Annotation on POJO methods. The `IntegrationMBeanExporter` post-process such a bean and registers respective MBean. It does this not in optimal way scanning all the `IntegrationConsumer` beans for requested `MessageHandler` which may cause a `BeanCurrentlyInCreationException`. * Rework the logic of the `IntegrationMBeanExporter.postProcessAbstractEndpoint()` to propagate provided endpoint for the monitor registration to bypass application context scanning for matched name. * Swap `equals()` for `monitor` since an `extractTarget()` may return `null` * Some other code clean in the `IntegrationMBeanExporter` * Change one of the `@ServiceActivator` in the configuration for the `ScatterGatherHandlerIntegrationTests` to POJO method. However, this didn't fail for me with original code unlike in the sample application provided in the mentioned SO thread. **Cherry-pick to `main`** --- .../monitor/IntegrationMBeanExporter.java | 66 ++++++++++--------- .../ScatterGatherHandlerIntegrationTests.java | 18 +---- 2 files changed, 39 insertions(+), 45 deletions(-) diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java index b4e3eb9780..841afa1056 100644 --- a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java @@ -334,7 +334,7 @@ public class IntegrationMBeanExporter extends MBeanExporter MessageHandler handler = integrationConsumer.getHandler(); MessageHandler monitor = (MessageHandler) extractTarget(handler); if (monitor instanceof IntegrationManagement) { - registerHandler((IntegrationManagement) monitor); + registerHandler((IntegrationManagement) monitor, integrationConsumer); this.handlers.put(((IntegrationManagement) monitor).getComponentName(), (IntegrationManagement) monitor); this.runtimeBeans.add(monitor); @@ -656,8 +656,12 @@ public class IntegrationMBeanExporter extends MBeanExporter } - private void registerHandler(IntegrationManagement monitor2) { - IntegrationManagement monitor = enhanceHandlerMonitor(monitor2); + private void registerHandler(IntegrationManagement monitor) { + registerHandler(monitor, null); + } + + private void registerHandler(IntegrationManagement monitor2, @Nullable IntegrationConsumer consumer) { + IntegrationManagement monitor = enhanceHandlerMonitor(monitor2, consumer); String name = monitor.getComponentName(); if (!this.objectNames.containsKey(monitor2) && matches(this.componentNamePatterns, name)) { String beanKey = getHandlerBeanKey(monitor); @@ -809,44 +813,47 @@ public class IntegrationMBeanExporter extends MBeanExporter .collect(Collectors.joining(",")); } - @SuppressWarnings("unlikely-arg-type") - private IntegrationManagement enhanceHandlerMonitor(IntegrationManagement monitor2) { + private IntegrationManagement enhanceHandlerMonitor(IntegrationManagement monitor, + @Nullable IntegrationConsumer consumer) { - if (monitor2.getManagedName() != null && monitor2.getManagedType() != null) { - return monitor2; + if (monitor.getManagedName() != null && monitor.getManagedType() != null) { + return monitor; } - // Assignment algorithm and bean id, with bean id pulled reflectively out of enclosing endpoint if possible - String[] names = this.applicationContext.getBeanNamesForType(IntegrationConsumer.class); - - String name = null; String endpointName = null; String source = "endpoint"; - IntegrationConsumer endpoint = null; + IntegrationConsumer endpoint = consumer; - for (String beanName : names) { - endpoint = this.applicationContext.getBean(beanName, IntegrationConsumer.class); - try { - MessageHandler handler = endpoint.getHandler(); - if (handler.equals(monitor2) || - extractTarget(handlerInAnonymousWrapper(handler)).equals(monitor2)) { - name = beanName; - endpointName = beanName; - break; + if (endpoint == null) { + // Assignment algorithm and bean id, with bean id pulled reflectively out of enclosing endpoint if possible + String[] names = this.applicationContext.getBeanNamesForType(IntegrationConsumer.class); + + for (String beanName : names) { + endpoint = this.applicationContext.getBean(beanName, IntegrationConsumer.class); + try { + MessageHandler handler = endpoint.getHandler(); + if (handler.equals(monitor) || monitor.equals(extractTarget(handlerInAnonymousWrapper(handler)))) { + endpointName = beanName; + break; + } + } + catch (Exception ex) { + logger.trace("Could not get handler from bean = " + beanName, ex); + endpoint = null; } } - catch (Exception e) { - logger.trace("Could not get handler from bean = " + beanName, e); - endpoint = null; - } + } + else { + endpointName = endpoint.getBeanName(); } IntegrationManagement messageHandlerMetrics = - buildMessageHandlerMetrics(monitor2, name, source, endpoint); + buildMessageHandlerMetrics(monitor, endpointName, source, endpoint); if (endpointName != null) { this.endpointsByMonitor.put(messageHandlerMetrics, endpointName); } return messageHandlerMetrics; + } private IntegrationManagement buildMessageHandlerMetrics( @@ -869,7 +876,7 @@ public class IntegrationMBeanExporter extends MBeanExporter } if (managedName == null) { - managedName = ((NamedComponent) monitor2).getComponentName(); + managedName = monitor2.getComponentName(); if (managedName == null) { managedName = monitor2.toString(); } @@ -902,7 +909,6 @@ public class IntegrationMBeanExporter extends MBeanExporter } private IntegrationInboundManagement enhanceSourceMonitor(IntegrationInboundManagement source2) { - if (source2.getManagedName() != null) { return source2; } @@ -958,8 +964,8 @@ public class IntegrationMBeanExporter extends MBeanExporter try { target = targetSource.getTarget(); } - catch (Exception e) { - logger.error("Could not get handler from bean = " + managedName); + catch (Exception ex) { + logger.error("Could not get handler from bean = " + managedName, ex); } } diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ScatterGatherHandlerIntegrationTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ScatterGatherHandlerIntegrationTests.java index 6af6141a32..0d50757b20 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ScatterGatherHandlerIntegrationTests.java +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ScatterGatherHandlerIntegrationTests.java @@ -333,21 +333,9 @@ public class ScatterGatherHandlerIntegrationTests { return new DirectChannel(); } - @Bean - @ServiceActivator(inputChannel = "serviceChannel2") - public MessageHandler service2() { - return new AbstractReplyProducingMessageHandler() { - - { - setOutputChannel(gatherChannel()); - } - - @Override - protected Object handleRequestMessage(Message requestMessage) { - return Math.random(); - } - - }; + @ServiceActivator(inputChannel = "serviceChannel2", outputChannel = "gatherChannel") + public double service2() { + return Math.random(); } }