Fix spring-integration context leak
MODIFY
instrumentation/spring/spring-integration-4.1/library/src/main/java/io/opentelemetry/instrumentation/spring/integration/TracingChannelInterceptor.java
instrumentation/spring/spring-integration-4.1/library/src/main/java/io/opentelemetry/instrumentation/spring/integration/TracingChannelInterceptor.java
@@ -37,6 +37,23 @@ final class TracingChannelInterceptor implements ExecutorChannelInterceptor {
@Override
public Message<?> preSend(Message<?> message, MessageChannel messageChannel) {
+
+ Map<MessageChannel, ContextAndScope> localMap = LOCAL_CONTEXT_AND_SCOPE.get();
+ if (localMap.get(messageChannel) != null) {
+ // GlobalChannelInterceptorProcessor.afterSingletonsInstantiated() adds the global
+ // interceptors for every bean name / channel pair, which means it's possible that this
+ // interceptor is added twice to the same channel if the channel is registered twice under
+ // different bean names
+ //
+ // there's an option for this class to implement VetoCapableInterceptor and prevent itself
+ // from being registered if it's already registered, but the VetoCapableInterceptor interface
+ // broke backwards compatibility in 5.2.0, and the version prior to 5.2.0 takes a parameter
+ // of type ChannelInterceptorAware which doesn't exist after 5.2.0, and while it's possible to
+ // implement both at the same time (since we compile using 4.1.0), muzzle doesn't like the
+ // missing class type when running testLatestDeps
+ return message;
+ }
+
Context parentContext = Context.current();
MessageWithChannel messageWithChannel = MessageWithChannel.create(message, messageChannel);
@@ -55,16 +72,12 @@ final class TracingChannelInterceptor implements ExecutorChannelInterceptor {
// instrumentation should not create another one
if (shouldStart(parentContext, messageWithChannel)) {
context = instrumenter.start(parentContext, messageWithChannel);
- LOCAL_CONTEXT_AND_SCOPE
- .get()
- .put(messageChannel, ContextAndScope.create(context, context.makeCurrent()));
+ localMap.put(messageChannel, ContextAndScope.create(context, context.makeCurrent()));
} else {
// in case there already was another span in the context: back off and just inject the current
// context into the message
context = parentContext;
- LOCAL_CONTEXT_AND_SCOPE
- .get()
- .put(messageChannel, ContextAndScope.create(null, context.makeCurrent()));
+ localMap.put(messageChannel, ContextAndScope.create(null, context.makeCurrent()));
}
propagators
@@ -113,6 +126,13 @@ final class TracingChannelInterceptor implements ExecutorChannelInterceptor {
@Override
public Message<?> beforeHandle(
Message<?> message, MessageChannel channel, MessageHandler handler) {
+
+ Map<MessageChannel, ContextAndScope> localMap = LOCAL_CONTEXT_AND_SCOPE.get();
+ if (localMap.get(channel) != null) {
+ // see comment explaining the same conditional in preSend()
+ return message;
+ }
+
MessageWithChannel messageWithChannel = MessageWithChannel.create(message, channel);
Context context =
propagators
@@ -120,7 +140,7 @@ final class TracingChannelInterceptor implements ExecutorChannelInterceptor {
.extract(Context.current(), messageWithChannel, MessageHeadersGetter.INSTANCE);
// beforeHandle()/afterMessageHandles() always execute in a different thread than send(), so
// there's no real risk of overwriting the send() context
- LOCAL_CONTEXT_AND_SCOPE.get().put(channel, ContextAndScope.create(null, context.makeCurrent()));
+ localMap.put(channel, ContextAndScope.create(null, context.makeCurrent()));
return message;
}

MODIFY
instrumentation/spring/spring-integration-4.1/testing/src/main/groovy/AbstractSpringIntegrationTracingTest.groovy
instrumentation/spring/spring-integration-4.1/testing/src/main/groovy/AbstractSpringIntegrationTracingTest.groovy
@@ -89,6 +89,41 @@ abstract class AbstractSpringIntegrationTracingTest extends InstrumentationSpeci
"executorChannel" | "executorChannel process"
}
+ def "should not add interceptor twice"() {
+ given:
+ def channel = applicationContext.getBean("directChannel1", SubscribableChannel)
+
+ def messageHandler = new CapturingMessageHandler()
+ channel.subscribe(messageHandler)
+
+ when:
+ channel.send(MessageBuilder.withPayload("test")
+ .build())
+
+ then:
+ def capturedMessage = messageHandler.join()
+
+ assertTraces(1) {
+ trace(0, 2) {
+ span(0) {
+ // the channel name is overwritten by the last bean registration
+ name "application.directChannel2 process"
+ kind CONSUMER
+ }
+ span(1) {
+ name "handler"
+ childOf span(0)
+ }
+
+ def interceptorSpan = span(0)
+ verifyCorrectSpanWasPropagated(capturedMessage, interceptorSpan)
+ }
+ }
+
+ cleanup:
+ channel.unsubscribe(messageHandler)
+ }
+
def "should not create a span when there is already a span in the context"() {
given:
def channel = applicationContext.getBean("directChannel", SubscribableChannel)
@@ -165,11 +200,24 @@ abstract class AbstractSpringIntegrationTracingTest extends InstrumentationSpeci
@SpringBootConfiguration
@EnableAutoConfiguration
static class MessageChannelsConfig {
+
+ SubscribableChannel problematicSharedChannel = new DirectChannel()
+
@Bean
SubscribableChannel directChannel() {
new DirectChannel()
}
+ @Bean
+ SubscribableChannel directChannel1() {
+ problematicSharedChannel
+ }
+
+ @Bean
+ SubscribableChannel directChannel2() {
+ problematicSharedChannel
+ }
+
@Bean
SubscribableChannel executorChannel(GlobalChannelInterceptorWrapper otelInterceptor) {
def channel = new ExecutorSubscribableChannel(Executors.newSingleThreadExecutor())

Fix spring-integration context leak (#4673)