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`**

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) {

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<ErrorMessage> 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 -> {