From 44c67d9e1c5706670e3efce004d11cd70a454155 Mon Sep 17 00:00:00 2001 From: Claude Meyer Date: Thu, 23 Jul 2026 02:02:31 +0000 Subject: [PATCH] Preserve MessageHeaderAccessor binding in spring-integration TracingChannelInterceptor preSend() rebuilds the message so it can carry the injected trace context. It did so with MessageBuilder.fromMessage(..).copyHeaders(..).build(), which produces plain MessageHeaders. That drops the MutableMessageHeaders binding between the message and its MessageHeaderAccessor, so components that recover the accessor via MessageHeaderAccessor.getAccessor() no longer find it. StompBrokerRelayMessageHandler is one such component: it requires the accessor and throws "No header accessor (not using the SimpMessagingTemplate?)" when it is missing, which makes STOMP broker relaying fail while this instrumentation is enabled. Rebuilding now distinguishes three cases. A message with no bound accessor, and any ErrorMessage, keep the existing MessageBuilder path; ErrorMessage does so because from 5.3 on MessageBuilder carries its originalMessage over into the rebuilt message and createMessage cannot. When the message's headers are mutable, getMutableAccessor() returns the message's own accessor and the context has already been written in place, so the message is returned unchanged; rebuilding it would seal the caller's message. Otherwise the accessor copy is sealed and its own MessageHeaders are handed to the new message, which preserves the binding while still giving the message an id and preventing it from sharing mutable headers with the accessor. Adds AbstractTracingChannelInterceptorTest covering context propagation, payload and header preservation, message id, header immutability, ErrorMessage type and originalMessage, STOMP native headers, and accessor preservation for mutable and immutable messages, over both a direct and an executor channel, from the library and javaagent test modules. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../v4_1/TracingChannelInterceptorTest.java | 20 ++ .../v4_1/TracingChannelInterceptor.java | 27 +- .../v4_1/TracingChannelInterceptorTest.java | 20 ++ ...AbstractTracingChannelInterceptorTest.java | 256 ++++++++++++++++++ 4 files changed, 320 insertions(+), 3 deletions(-) create mode 100644 instrumentation/spring/spring-integration-4.1/javaagent/src/test/java/io/opentelemetry/javaagent/instrumentation/spring/integration/v4_1/TracingChannelInterceptorTest.java create mode 100644 instrumentation/spring/spring-integration-4.1/library/src/test/java/io/opentelemetry/javaagent/instrumentation/spring/integration/v4_1/TracingChannelInterceptorTest.java create mode 100644 instrumentation/spring/spring-integration-4.1/testing/src/main/java/io/opentelemetry/javaagent/instrumentation/spring/integration/v4_1/AbstractTracingChannelInterceptorTest.java diff --git a/instrumentation/spring/spring-integration-4.1/javaagent/src/test/java/io/opentelemetry/javaagent/instrumentation/spring/integration/v4_1/TracingChannelInterceptorTest.java b/instrumentation/spring/spring-integration-4.1/javaagent/src/test/java/io/opentelemetry/javaagent/instrumentation/spring/integration/v4_1/TracingChannelInterceptorTest.java new file mode 100644 index 000000000000..8b3d8411a536 --- /dev/null +++ b/instrumentation/spring/spring-integration-4.1/javaagent/src/test/java/io/opentelemetry/javaagent/instrumentation/spring/integration/v4_1/TracingChannelInterceptorTest.java @@ -0,0 +1,20 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.javaagent.instrumentation.spring.integration.v4_1; + +import io.opentelemetry.instrumentation.testing.junit.AgentInstrumentationExtension; +import io.opentelemetry.instrumentation.testing.junit.InstrumentationExtension; +import org.junit.jupiter.api.extension.RegisterExtension; + +class TracingChannelInterceptorTest extends AbstractTracingChannelInterceptorTest { + + @RegisterExtension + private static final InstrumentationExtension testing = AgentInstrumentationExtension.create(); + + TracingChannelInterceptorTest() { + super(testing, null); + } +} diff --git a/instrumentation/spring/spring-integration-4.1/library/src/main/java/io/opentelemetry/instrumentation/spring/integration/v4_1/TracingChannelInterceptor.java b/instrumentation/spring/spring-integration-4.1/library/src/main/java/io/opentelemetry/instrumentation/spring/integration/v4_1/TracingChannelInterceptor.java index 9c595d006b96..5b1543d7f526 100644 --- a/instrumentation/spring/spring-integration-4.1/library/src/main/java/io/opentelemetry/instrumentation/spring/integration/v4_1/TracingChannelInterceptor.java +++ b/instrumentation/spring/spring-integration-4.1/library/src/main/java/io/opentelemetry/instrumentation/spring/integration/v4_1/TracingChannelInterceptor.java @@ -22,6 +22,7 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.support.ErrorMessage; import org.springframework.messaging.support.ExecutorChannelInterceptor; import org.springframework.messaging.support.MessageBuilder; import org.springframework.messaging.support.MessageHeaderAccessor; @@ -234,9 +235,29 @@ private static void ensureNativeHeadersAreMutable(MessageHeaderAccessor headerAc private static Message createMessageWithHeaders( Message message, MessageHeaderAccessor messageHeaderAccessor) { - return MessageBuilder.fromMessage(message) - .copyHeaders(messageHeaderAccessor.toMessageHeaders()) - .build(); + MessageHeaderAccessor boundAccessor = + MessageHeaderAccessor.getAccessor(message, MessageHeaderAccessor.class); + + // an ErrorMessage carries an originalMessage that only MessageBuilder propagates + if (boundAccessor == null || message instanceof ErrorMessage) { + return MessageBuilder.fromMessage(message) + .copyHeaders(messageHeaderAccessor.toMessageHeaders()) + .build(); + } + + if (boundAccessor == messageHeaderAccessor) { + // getMutableAccessor() hands back the message's own accessor when its headers are mutable, + // so the context has already been written into this message + return message; + } + + // Sealing the copy gives the new message an id and keeps it from sharing mutable headers with + // the accessor. Passing the accessor's own MessageHeaders preserves the binding that Spring + // components such as StompBrokerRelayMessageHandler recover through + // MessageHeaderAccessor.getAccessor(). + messageHeaderAccessor.setImmutable(); + return MessageBuilder.createMessage( + message.getPayload(), messageHeaderAccessor.getMessageHeaders()); } private boolean createProducerSpan(MessageChannel messageChannel) { diff --git a/instrumentation/spring/spring-integration-4.1/library/src/test/java/io/opentelemetry/javaagent/instrumentation/spring/integration/v4_1/TracingChannelInterceptorTest.java b/instrumentation/spring/spring-integration-4.1/library/src/test/java/io/opentelemetry/javaagent/instrumentation/spring/integration/v4_1/TracingChannelInterceptorTest.java new file mode 100644 index 000000000000..e6a2392d5c66 --- /dev/null +++ b/instrumentation/spring/spring-integration-4.1/library/src/test/java/io/opentelemetry/javaagent/instrumentation/spring/integration/v4_1/TracingChannelInterceptorTest.java @@ -0,0 +1,20 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.javaagent.instrumentation.spring.integration.v4_1; + +import io.opentelemetry.instrumentation.testing.junit.InstrumentationExtension; +import io.opentelemetry.instrumentation.testing.junit.LibraryInstrumentationExtension; +import org.junit.jupiter.api.extension.RegisterExtension; + +class TracingChannelInterceptorTest extends AbstractTracingChannelInterceptorTest { + + @RegisterExtension + static final InstrumentationExtension testing = LibraryInstrumentationExtension.create(); + + TracingChannelInterceptorTest() { + super(testing, GlobalInterceptorSpringConfig.class); + } +} diff --git a/instrumentation/spring/spring-integration-4.1/testing/src/main/java/io/opentelemetry/javaagent/instrumentation/spring/integration/v4_1/AbstractTracingChannelInterceptorTest.java b/instrumentation/spring/spring-integration-4.1/testing/src/main/java/io/opentelemetry/javaagent/instrumentation/spring/integration/v4_1/AbstractTracingChannelInterceptorTest.java new file mode 100644 index 000000000000..ed3c1f5168c0 --- /dev/null +++ b/instrumentation/spring/spring-integration-4.1/testing/src/main/java/io/opentelemetry/javaagent/instrumentation/spring/integration/v4_1/AbstractTracingChannelInterceptorTest.java @@ -0,0 +1,256 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.javaagent.instrumentation.spring.integration.v4_1; + +import static io.opentelemetry.instrumentation.testing.util.TestLatestDeps.testLatestDeps; +import static io.opentelemetry.javaagent.instrumentation.spring.integration.v4_1.AbstractSpringIntegrationTracingTest.verifyCorrectSpanWasPropagated; +import static java.util.Collections.singletonMap; +import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.jupiter.api.Assumptions.assumeTrue; + +import io.opentelemetry.api.trace.SpanKind; +import io.opentelemetry.instrumentation.testing.internal.AutoCleanupExtension; +import io.opentelemetry.instrumentation.testing.junit.InstrumentationExtension; +import java.lang.reflect.Constructor; +import java.util.ArrayList; +import java.util.List; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; +import org.springframework.boot.SpringApplication; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.SubscribableChannel; +import org.springframework.messaging.simp.stomp.StompCommand; +import org.springframework.messaging.simp.stomp.StompHeaderAccessor; +import org.springframework.messaging.support.ErrorMessage; +import org.springframework.messaging.support.GenericMessage; +import org.springframework.messaging.support.MessageBuilder; +import org.springframework.messaging.support.MessageHeaderAccessor; + +/** + * Covers what the interceptor must preserve while rebuilding a message to carry trace context: the + * payload, the existing headers, the message id, the message type, and the {@link + * MessageHeaderAccessor} binding, which Spring components such as {@code + * StompBrokerRelayMessageHandler} recover via {@link MessageHeaderAccessor#getAccessor(Message, + * Class)}. + */ +abstract class AbstractTracingChannelInterceptorTest { + + @RegisterExtension static final AutoCleanupExtension cleanup = AutoCleanupExtension.create(); + + private final InstrumentationExtension testing; + + private final Class additionalContextClass; + + private ConfigurableApplicationContext applicationContext; + + AbstractTracingChannelInterceptorTest( + InstrumentationExtension testing, Class additionalContextClass) { + this.testing = testing; + this.additionalContextClass = additionalContextClass; + } + + @BeforeEach + void setUp() { + List> contextClasses = new ArrayList<>(); + contextClasses.add(AbstractSpringIntegrationTracingTest.MessageChannelsConfig.class); + if (additionalContextClass != null) { + contextClasses.add(additionalContextClass); + } + SpringApplication springApplication = + new SpringApplication(contextClasses.toArray(new Class[0])); + springApplication.setDefaultProperties( + singletonMap("spring.main.web-application-type", "none")); + applicationContext = springApplication.run(); + cleanup.deferCleanup(applicationContext); + } + + @ParameterizedTest + @ValueSource(strings = {"directChannel", "executorChannel"}) + void preservesHeaderAccessorOfImmutableMessage(String channelName) { + // messages produced by SimpMessagingTemplate carry immutable headers, so the interceptor has to + // take a mutable copy of the accessor rather than mutating the message in place + TestHeaderAccessor accessor = new TestHeaderAccessor(); + accessor.setHeader("testKey", "testValue"); + Message message = MessageBuilder.createMessage("payload", accessor.getMessageHeaders()); + + Message received = sendAndCapture(channelName, message); + + assertThat(received.getPayload()).isEqualTo("payload"); + assertThat(received.getHeaders()).containsEntry("testKey", "testValue"); + assertThat(MessageHeaderAccessor.getAccessor(received, TestHeaderAccessor.class)).isNotNull(); + // a rebuilt message still needs an id, and must not go on sharing mutable headers with the + // accessor the interceptor injected into + assertThat(received.getHeaders().getId()).isNotNull(); + assertThat(MessageHeaderAccessor.getAccessor(received, MessageHeaderAccessor.class).isMutable()) + .isFalse(); + + assertContextWasPropagated(received); + } + + @Test + void preservesHeaderAccessorOfMutableMessage() { + TestHeaderAccessor accessor = new TestHeaderAccessor(); + accessor.setLeaveMutable(true); + accessor.setHeader("testKey", "testValue"); + + Message received = + sendAndCapture( + "directChannel", MessageBuilder.createMessage("payload", accessor.getMessageHeaders())); + + assertThat(received.getHeaders()).containsEntry("testKey", "testValue"); + assertThat(MessageHeaderAccessor.getAccessor(received, TestHeaderAccessor.class)).isNotNull(); + // sealing the accessor here would seal the caller's message too + assertThat(accessor.isMutable()).isTrue(); + + assertContextWasPropagated(received); + } + + @Test + void doesNotShareHeadersBetweenSends() { + TestHeaderAccessor accessor = new TestHeaderAccessor(); + Message message = MessageBuilder.createMessage("payload", accessor.getMessageHeaders()); + + Message first = sendAndCapture("directChannel", message); + Object firstTraceParent = first.getHeaders().get("traceparent"); + + Message second = sendAndCapture("directChannel", first); + + assertThat(second.getHeaders()).isNotSameAs(first.getHeaders()); + assertThat(first.getHeaders().get("traceparent")).isEqualTo(firstTraceParent); + } + + @Test + void preservesStompHeaderAccessor() { + // the shape SimpMessagingTemplate produces and StompBrokerRelayMessageHandler consumes + StompHeaderAccessor accessor = StompHeaderAccessor.create(StompCommand.SEND); + accessor.setDestination("/topic/test"); + + Message received = + sendAndCapture( + "directChannel", MessageBuilder.createMessage("payload", accessor.getMessageHeaders())); + + StompHeaderAccessor recovered = + MessageHeaderAccessor.getAccessor(received, StompHeaderAccessor.class); + assertThat(recovered).isNotNull(); + assertThat(recovered.getDestination()).isEqualTo("/topic/test"); + // a STOMP frame only carries native headers, so the context has to reach them too + assertThat(recovered.getNativeHeader("traceparent")).isNotNull(); + + assertContextWasPropagated(received); + } + + @Test + void preservesErrorMessage() { + // ErrorMessage keeps the MessageBuilder path so that its originalMessage survives + IllegalStateException payload = new IllegalStateException("boom"); + TestHeaderAccessor accessor = new TestHeaderAccessor(); + + Message received = + sendAndCapture( + "directChannel", MessageBuilder.createMessage(payload, accessor.getMessageHeaders())); + + assertThat(received).isInstanceOf(ErrorMessage.class); + assertThat(received.getPayload()).isSameAs(payload); + assertThat(received.getHeaders().getId()).isNotNull(); + + assertContextWasPropagated(received); + } + + @Test + void preservesErrorMessageOriginalMessage() throws ReflectiveOperationException { + // MessageBuilder only carries originalMessage over into a rebuilt ErrorMessage from 5.3 on; + // reflection is needed either way because neither the constructor that populates it nor + // getOriginalMessage() exists in 4.1, which is the version this module compiles against + assumeTrue(testLatestDeps()); + Constructor constructor = + ErrorMessage.class.getConstructor(Throwable.class, MessageHeaders.class, Message.class); + Message original = new GenericMessage<>("original"); + TestHeaderAccessor accessor = new TestHeaderAccessor(); + Message errorMessage = + constructor.newInstance( + new IllegalStateException("boom"), accessor.getMessageHeaders(), original); + + Message received = sendAndCapture("directChannel", errorMessage); + + assertThat(ErrorMessage.class.getMethod("getOriginalMessage").invoke(received)) + .isSameAs(original); + } + + @Test + void preservesPayloadAndHeadersOfPlainMessage() { + Message received = + sendAndCapture( + "directChannel", + MessageBuilder.withPayload("payload").setHeader("testKey", "testValue").build()); + + assertThat(received.getPayload()).isEqualTo("payload"); + assertThat(received.getHeaders()).containsEntry("testKey", "testValue"); + assertThat(received.getHeaders().getId()).isNotNull(); + + assertContextWasPropagated(received); + } + + @Test + void injectsTraceContextIntoPlainMessage() { + Message received = sendAndCapture("directChannel", new GenericMessage<>("payload")); + + assertThat(received.getHeaders()).containsKey("traceparent"); + + assertContextWasPropagated(received); + } + + /** Sends {@code message} through an intercepted channel and returns what the handler received. */ + private Message sendAndCapture(String channelName, Message message) { + SubscribableChannel channel = + applicationContext.getBean(channelName, SubscribableChannel.class); + CapturingMessageHandler messageHandler = new CapturingMessageHandler(); + channel.subscribe(messageHandler); + try { + channel.send(message); + return messageHandler.join(); + } finally { + channel.unsubscribe(messageHandler); + } + } + + /** Asserts that the delivered message carries the trace context of the interceptor's span. */ + private void assertContextWasPropagated(Message received) { + testing.waitAndAssertTraces( + trace -> + trace.hasSpansSatisfyingExactly( + span -> { + span.hasKind(SpanKind.CONSUMER); + verifyCorrectSpanWasPropagated(received, trace.getSpan(0)); + }, + span -> span.hasName("handler").hasParent(trace.getSpan(0)))); + } + + /** + * Stands in for accessor subclasses such as {@code SimpMessageHeaderAccessor}, which override + * {@link MessageHeaderAccessor#createAccessor(Message)} so that the subclass survives when a + * mutable copy is taken of an immutable message. + */ + static class TestHeaderAccessor extends MessageHeaderAccessor { + + TestHeaderAccessor() { + super(); + } + + TestHeaderAccessor(Message message) { + super(message); + } + + @Override + protected MessageHeaderAccessor createAccessor(Message message) { + return new TestHeaderAccessor(message); + } + } +}