From b0cbacfafbbb37a32cbd3684215652b464bf99aa Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 3 Jun 2024 14:49:33 -0400 Subject: [PATCH] GH-9198: Fix `MessageChannel` observation for `ErrorMessage` Fixes: #9198 When observation is enabled on the `MessageChannel`, the message to send is converted to a `MutableMessage`. In case of `ErrorMessage` this causes a loss of `originalMessage` and may lead to `ClassCastException` in the target error handler. * Check for `ErrorMessage` in the `AbstractMessageChannel.sendWithObservation()` and create a new one as a copy of original request, but including observation headers before performing `sendInternal()` * Modify `IntegrationObservabilityZipkinTests` to verify an observation with an `ErrorMessage` and its handler. **Auto-cherry-pick to `6.3.x` & `6.2.x`** --- .../channel/AbstractMessageChannel.java | 24 +++++++++++---- .../IntegrationObservabilityZipkinTests.java | 29 +++++++++++++++++-- 2 files changed, 45 insertions(+), 8 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java index 2ddc0c0d24..a46824b2de 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java @@ -30,6 +30,7 @@ import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; +import io.micrometer.observation.Observation; import io.micrometer.observation.ObservationRegistry; import org.springframework.beans.factory.BeanFactory; @@ -59,6 +60,7 @@ import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageDeliveryException; import org.springframework.messaging.converter.MessageConverter; import org.springframework.messaging.support.ChannelInterceptor; +import org.springframework.messaging.support.ErrorMessage; import org.springframework.messaging.support.InterceptableChannel; import org.springframework.util.Assert; import org.springframework.util.StringUtils; @@ -335,12 +337,22 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport private boolean sendWithObservation(Message message, long timeout) { MutableMessage messageToSend = MutableMessage.of(message); - return IntegrationObservation.PRODUCER.observation( - this.observationConvention, - DefaultMessageSenderObservationConvention.INSTANCE, - () -> new MessageSenderContext(messageToSend, getComponentName()), - this.observationRegistry) - .observe(() -> sendInternal(messageToSend, timeout)); // NOSONAR - never null + Observation observation = IntegrationObservation.PRODUCER.observation( + this.observationConvention, + DefaultMessageSenderObservationConvention.INSTANCE, + () -> new MessageSenderContext(messageToSend, getComponentName()), + this.observationRegistry); + Boolean observe = observation.observe(() -> { + Message messageToSendInternal = messageToSend; + if (message instanceof ErrorMessage errorMessage) { + messageToSendInternal = + new ErrorMessage(errorMessage.getPayload(), + messageToSend.getHeaders(), + errorMessage.getOriginalMessage()); + } + return sendInternal(messageToSendInternal, timeout); + }); + return Boolean.TRUE.equals(observe); } private boolean sendWithMetrics(Message message, long timeout) { 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 46ede28b3d..16d7e0d9ee 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 @@ -48,7 +48,9 @@ import org.springframework.integration.handler.BridgeHandler; import org.springframework.integration.handler.advice.HandleMessageAdvice; import org.springframework.lang.Nullable; import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.support.ErrorMessage; import org.springframework.messaging.support.GenericMessage; import static org.assertj.core.api.Assertions.assertThat; @@ -102,6 +104,18 @@ public class IntegrationObservabilityZipkinTests extends SampleTestRunner { assertThat(configuration.observedHandlerLatch.await(10, TimeUnit.SECONDS)).isTrue(); await().untilAsserted(() -> assertThat(configuration.observationReference.get()).isNotNull()); + + MessageChannel errorChannel = applicationContext.getBean("myErrorChannel", MessageChannel.class); + ErrorMessage errorMessage = + new ErrorMessage(new RuntimeException("some error"), new GenericMessage<>("some original")); + errorChannel.send(errorMessage); + + assertThat(configuration.errorMessageReceivedLatch.await(10, TimeUnit.SECONDS)).isTrue(); + + ErrorMessage receivedErrorMessage = configuration.errorMessageReference.get(); + assertThat(receivedErrorMessage.getOriginalMessage()).isEqualTo(errorMessage.getOriginalMessage()); + assertThat(receivedErrorMessage.getPayload()).isEqualTo(errorMessage.getPayload()); + assertThat(receivedErrorMessage.getHeaders()).containsKeys("X-B3-TraceId", "X-B3-SpanId"); } SpansAssert.assertThat(bb.getFinishedSpans()) @@ -119,7 +133,7 @@ public class IntegrationObservabilityZipkinTests extends SampleTestRunner { .hasTag(IntegrationObservation.ProducerTags.COMPONENT_NAME.asString(), "queueChannel") .hasTag(IntegrationObservation.ProducerTags.COMPONENT_TYPE.asString(), "producer") .hasKindEqualTo(Span.Kind.PRODUCER)) - .hasSize(4); + .hasSize(5); MeterRegistryAssert.assertThat(getMeterRegistry()) .hasTimerWithNameAndTags("spring.integration.handler", @@ -134,7 +148,7 @@ public class IntegrationObservabilityZipkinTests extends SampleTestRunner { @EnableIntegration @EnableIntegrationManagement( observationPatterns = { - "${spring.integration.management.observation-patterns:testInboundGateway,skippedObservationInboundGateway,queueChannel,observedEndpoint,publishSubscribeChannel}", + "${spring.integration.management.observation-patterns:testInboundGateway,skippedObservationInboundGateway,queueChannel,myErrorChannel,observedEndpoint,publishSubscribeChannel}", "${spring.integration.management.observation-patterns:}" }) public static class ObservationIntegrationTestConfiguration { @@ -181,6 +195,17 @@ public class IntegrationObservabilityZipkinTests extends SampleTestRunner { return bridgeHandler; } + + AtomicReference errorMessageReference = new AtomicReference<>(); + + CountDownLatch errorMessageReceivedLatch = new CountDownLatch(1); + + @ServiceActivator(inputChannel = "myErrorChannel") + void handleError(ErrorMessage errorMessage) { + this.errorMessageReference.set(errorMessage); + this.errorMessageReceivedLatch.countDown(); + } + @Bean HandleMessageAdvice observedHandlerAdvice() { return invocation -> {