diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/ServiceActivatorAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/ServiceActivatorAnnotationPostProcessor.java index 9659c31e33..c7314446c9 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/ServiceActivatorAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/ServiceActivatorAnnotationPostProcessor.java @@ -46,7 +46,7 @@ import org.springframework.util.StringUtils; public class ServiceActivatorAnnotationPostProcessor extends AbstractMethodAnnotationPostProcessor { public ServiceActivatorAnnotationPostProcessor() { - this.messageHandlerAttributes.addAll(Arrays.asList("outputChannel", "requiresReply", "adviceChain")); + this.messageHandlerAttributes.addAll(Arrays.asList("outputChannel", "requiresReply")); } @Override diff --git a/spring-integration-core/src/test/java/org/springframework/integration/support/management/observation/IntegrationObservabilityZipkinTests.java b/spring-integration-core/src/test/java/org/springframework/integration/support/management/observation/IntegrationObservabilityZipkinTests.java index f008524833..f34b65652f 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/support/management/observation/IntegrationObservabilityZipkinTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/support/management/observation/IntegrationObservabilityZipkinTests.java @@ -18,17 +18,23 @@ package org.springframework.integration.support.management.observation; import static org.assertj.core.api.Assertions.assertThat; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + import org.springframework.context.annotation.AnnotationConfigApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.integration.annotation.BridgeTo; import org.springframework.integration.annotation.EndpointId; import org.springframework.integration.annotation.Poller; +import org.springframework.integration.annotation.ServiceActivator; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.channel.interceptor.ObservationPropagationChannelInterceptor; import org.springframework.integration.config.EnableIntegration; import org.springframework.integration.config.EnableIntegrationManagement; import org.springframework.integration.config.GlobalChannelInterceptor; +import org.springframework.integration.handler.BridgeHandler; +import org.springframework.integration.handler.advice.HandleMessageAdvice; import org.springframework.integration.support.MutableMessage; import org.springframework.integration.support.MutableMessageBuilder; import org.springframework.messaging.Message; @@ -79,6 +85,9 @@ public class IntegrationObservabilityZipkinTests extends SampleTestRunner { Message receive = replyChannel.receive(10_000); assertThat(receive).isNotNull() .extracting("payload").isEqualTo("test data"); + var configuration = applicationContext.getBean(ObservationIntegrationTestConfiguration.class); + + assertThat(configuration.observedHandlerLatch.await(10, TimeUnit.SECONDS)).isTrue(); } SpansAssert.assertThat(bb.getFinishedSpans()) @@ -105,6 +114,8 @@ public class IntegrationObservabilityZipkinTests extends SampleTestRunner { @EnableIntegrationManagement public static class ObservationIntegrationTestConfiguration { + CountDownLatch observedHandlerLatch = new CountDownLatch(1); + @Bean @GlobalChannelInterceptor public ChannelInterceptor observationPropagationInterceptor(ObservationRegistry observationRegistry) { @@ -112,12 +123,31 @@ public class IntegrationObservabilityZipkinTests extends SampleTestRunner { } @Bean - @BridgeTo(poller = @Poller(fixedDelay = "100")) - @EndpointId("observedEndpoint") public PollableChannel queueChannel() { return new QueueChannel(); } + @Bean + @EndpointId("observedEndpoint") + @ServiceActivator(inputChannel = "queueChannel", + poller = @Poller(fixedDelay = "100"), + adviceChain = "observedHandlerAdvice") + BridgeHandler bridgeHandler() { + return new BridgeHandler(); + } + + @Bean + HandleMessageAdvice observedHandlerAdvice() { + return invocation -> { + try { + return invocation.proceed(); + } + finally { + this.observedHandlerLatch.countDown(); + } + }; + } + } }