Added better remote name resolution for messaging; fixes gh-1481

2020-02-05 13:36:15 +01:00
parent ac43895b75
commit 415c809fd0
2 changed files with 152 additions and 153 deletions
--- a/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptor.java
+++ b/spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptor.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2019 the original author or authors.
+ * Copyright 2013-2020 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
@@ -59,8 +59,7 @@ import org.springframework.util.ClassUtils;
 *
 * @author Marcin Grzejszczak
 */
-public final class TracingChannelInterceptor extends ChannelInterceptorAdapter
-		implements ExecutorChannelInterceptor {
+public final class TracingChannelInterceptor extends ChannelInterceptorAdapter implements ExecutorChannelInterceptor {

 	/**
 	 * Name of the class in Spring Cloud Stream that is a direct channel.
@@ -109,25 +108,22 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter

 	@Autowired
 	TracingChannelInterceptor(Tracing tracing) {
-		this(tracing, MessageHeaderPropagation.INSTANCE,
-				MessageHeaderPropagation.INSTANCE);
+		this(tracing, MessageHeaderPropagation.INSTANCE, MessageHeaderPropagation.INSTANCE);
 	}

-	TracingChannelInterceptor(Tracing tracing,
-			Propagation.Setter<MessageHeaderAccessor, String> setter,
+	TracingChannelInterceptor(Tracing tracing, Propagation.Setter<MessageHeaderAccessor, String> setter,
 			Propagation.Getter<MessageHeaderAccessor, String> getter) {
 		this.tracing = tracing;
 		this.tracer = tracing.tracer();
 		this.threadLocalSpan = ThreadLocalSpan.create(this.tracer);
 		this.injector = tracing.propagation().injector(setter);
 		this.extractor = tracing.propagation().extractor(getter);
-		this.integrationObjectSupportPresent = ClassUtils.isPresent(
-				"org.springframework.integration.context.IntegrationObjectSupport", null);
-		this.hasDirectChannelClass = ClassUtils
-				.isPresent("org.springframework.integration.channel.DirectChannel", null);
-		this.directWithAttributesChannelClass = ClassUtils
-				.isPresent(STREAM_DIRECT_CHANNEL, null)
-						? ClassUtils.resolveClassName(STREAM_DIRECT_CHANNEL, null) : null;
+		this.integrationObjectSupportPresent = ClassUtils
+				.isPresent("org.springframework.integration.context.IntegrationObjectSupport", null);
+		this.hasDirectChannelClass = ClassUtils.isPresent("org.springframework.integration.channel.DirectChannel",
+				null);
+		this.directWithAttributesChannelClass = ClassUtils.isPresent(STREAM_DIRECT_CHANNEL, null)
+				? ClassUtils.resolveClassName(STREAM_DIRECT_CHANNEL, null) : null;
 	}

 	public static TracingChannelInterceptor create(Tracing tracing) {
@@ -170,12 +166,11 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter
 		MessageHeaderAccessor headers = mutableHeaderAccessor(retrievedMessage);
 		TraceContextOrSamplingFlags extracted = this.extractor.extract(headers);
 		Span span = this.threadLocalSpan.next(extracted);
-		MessageHeaderPropagation.removeAnyTraceHeaders(headers,
-				this.tracing.propagation().keys());
+		MessageHeaderPropagation.removeAnyTraceHeaders(headers, this.tracing.propagation().keys());
 		this.injector.inject(span.context(), headers);
 		if (!span.isNoop()) {
 			span.kind(Span.Kind.PRODUCER).name("send").start();
-			span.remoteServiceName(REMOTE_SERVICE_NAME);
+			span.remoteServiceName(toRemoteServiceName(headers));
 			addTags(message, span, channel);
 		}
 		if (log.isDebugEnabled()) {
@@ -188,24 +183,31 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter
 		return outputMessage;
 	}

-	private Message<?> outputMessage(Message<?> originalMessage,
-			Message<?> retrievedMessage, MessageHeaderAccessor additionalHeaders) {
-		MessageHeaderAccessor headers = MessageHeaderAccessor
-				.getMutableAccessor(originalMessage);
+	private String toRemoteServiceName(MessageHeaderAccessor headers) {
+		for (String key : headers.getMessageHeaders().keySet()) {
+			if (key.startsWith("kafka_")) {
+				return "kafka";
+			}
+			else if (key.startsWith("amqp_")) {
+				return "rabbitmq";
+			}
+		}
+		return REMOTE_SERVICE_NAME;
+	}
+
+	private Message<?> outputMessage(Message<?> originalMessage, Message<?> retrievedMessage,
+			MessageHeaderAccessor additionalHeaders) {
+		MessageHeaderAccessor headers = MessageHeaderAccessor.getMutableAccessor(originalMessage);
 		if (originalMessage instanceof ErrorMessage) {
 			ErrorMessage errorMessage = (ErrorMessage) originalMessage;
-			headers.copyHeaders(MessageHeaderPropagation.propagationHeaders(
-					additionalHeaders.getMessageHeaders(),
+			headers.copyHeaders(MessageHeaderPropagation.propagationHeaders(additionalHeaders.getMessageHeaders(),
 					this.tracing.propagation().keys()));
-			return new ErrorMessage(errorMessage.getPayload(),
-					isWebSockets(headers) ? headers.getMessageHeaders()
-							: new MessageHeaders(headers.getMessageHeaders()),
-					errorMessage.getOriginalMessage());
+			return new ErrorMessage(errorMessage.getPayload(), isWebSockets(headers) ? headers.getMessageHeaders()
+					: new MessageHeaders(headers.getMessageHeaders()), errorMessage.getOriginalMessage());
 		}
 		headers.copyHeaders(additionalHeaders.getMessageHeaders());
 		return new GenericMessage<>(retrievedMessage.getPayload(),
-				isWebSockets(headers) ? headers.getMessageHeaders()
-						: new MessageHeaders(headers.getMessageHeaders()));
+				isWebSockets(headers) ? headers.getMessageHeaders() : new MessageHeaders(headers.getMessageHeaders()));
 	}

 	private boolean isWebSockets(MessageHeaderAccessor headerAccessor) {
@@ -215,8 +217,7 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter

 	private boolean isDirectChannel(MessageChannel channel) {
 		Class<?> targetClass = AopUtils.getTargetClass(channel);
-		boolean directChannel = this.hasDirectChannelClass
-				&& DirectChannel.class.isAssignableFrom(targetClass);
+		boolean directChannel = this.hasDirectChannelClass && DirectChannel.class.isAssignableFrom(targetClass);
 		if (!directChannel) {
 			return false;
 		}
@@ -231,8 +232,7 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter
 	}

 	@Override
-	public void afterSendCompletion(Message<?> message, MessageChannel channel,
-			boolean sent, Exception ex) {
+	public void afterSendCompletion(Message<?> message, MessageChannel channel, boolean sent, Exception ex) {
 		if (emptyMessage(message)) {
 			return;
 		}
@@ -240,8 +240,7 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter
 			afterMessageHandled(message, channel, null, ex);
 		}
 		if (log.isDebugEnabled()) {
-			log.debug("Will finish the current span after completion "
-					+ this.tracer.currentSpan());
+			log.debug("Will finish the current span after completion " + this.tracer.currentSpan());
 		}
 		finishSpan(ex);
 	}
@@ -258,12 +257,11 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter
 		MessageHeaderAccessor headers = mutableHeaderAccessor(message);
 		TraceContextOrSamplingFlags extracted = this.extractor.extract(headers);
 		Span span = this.threadLocalSpan.next(extracted);
-		MessageHeaderPropagation.removeAnyTraceHeaders(headers,
-				this.tracing.propagation().keys());
+		MessageHeaderPropagation.removeAnyTraceHeaders(headers, this.tracing.propagation().keys());
 		this.injector.inject(span.context(), headers);
 		if (!span.isNoop()) {
 			span.kind(Span.Kind.CONSUMER).name("receive").start();
-			span.remoteServiceName(REMOTE_SERVICE_NAME);
+			span.remoteServiceName(toRemoteServiceName(headers));
 			addTags(message, span, channel);
 		}
 		if (log.isDebugEnabled()) {
@@ -272,21 +270,19 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter
 		headers.setImmutable();
 		if (message instanceof ErrorMessage) {
 			ErrorMessage errorMessage = (ErrorMessage) message;
-			return new ErrorMessage(errorMessage.getPayload(),
-					headers.getMessageHeaders(), errorMessage.getOriginalMessage());
+			return new ErrorMessage(errorMessage.getPayload(), headers.getMessageHeaders(),
+					errorMessage.getOriginalMessage());
 		}
 		return new GenericMessage<>(message.getPayload(), headers.getMessageHeaders());
 	}

 	@Override
-	public void afterReceiveCompletion(Message<?> message, MessageChannel channel,
-			Exception ex) {
+	public void afterReceiveCompletion(Message<?> message, MessageChannel channel, Exception ex) {
 		if (emptyMessage(message)) {
 			return;
 		}
 		if (log.isDebugEnabled()) {
-			log.debug("Will finish the current span after receive completion "
-					+ this.tracer.currentSpan());
+			log.debug("Will finish the current span after receive completion " + this.tracer.currentSpan());
 		}
 		finishSpan(ex);
 	}
@@ -296,8 +292,7 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter
 	 * context. It then creates a span for the handler, placing it in scope.
 	 */
 	@Override
-	public Message<?> beforeHandle(Message<?> message, MessageChannel channel,
-			MessageHandler handler) {
+	public Message<?> beforeHandle(Message<?> message, MessageChannel channel, MessageHandler handler) {
 		if (emptyMessage(message)) {
 			return message;
 		}
@@ -312,34 +307,28 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter
 			consumerSpan.finish();
 		}
 		// create and scope a span for the message processor
-		this.threadLocalSpan
-				.next(TraceContextOrSamplingFlags.create(consumerSpan.context()))
-				.name("handle").start();
+		this.threadLocalSpan.next(TraceContextOrSamplingFlags.create(consumerSpan.context())).name("handle").start();
 		// remove any trace headers, but don't re-inject as we are synchronously
 		// processing the
 		// message and can rely on scoping to access this span later.
-		MessageHeaderPropagation.removeAnyTraceHeaders(headers,
-				this.tracing.propagation().keys());
+		MessageHeaderPropagation.removeAnyTraceHeaders(headers, this.tracing.propagation().keys());
 		if (log.isDebugEnabled()) {
 			log.debug("Created a new span in before handle" + consumerSpan);
 		}
 		if (message instanceof ErrorMessage) {
-			return new ErrorMessage((Throwable) message.getPayload(),
-					headers.getMessageHeaders());
+			return new ErrorMessage((Throwable) message.getPayload(), headers.getMessageHeaders());
 		}
 		headers.setImmutable();
 		return new GenericMessage<>(message.getPayload(), headers.getMessageHeaders());
 	}

 	@Override
-	public void afterMessageHandled(Message<?> message, MessageChannel channel,
-			MessageHandler handler, Exception ex) {
+	public void afterMessageHandled(Message<?> message, MessageChannel channel, MessageHandler handler, Exception ex) {
 		if (emptyMessage(message)) {
 			return;
 		}
 		if (log.isDebugEnabled()) {
-			log.debug("Will finish the current span after message handled "
-					+ this.tracer.currentSpan());
+			log.debug("Will finish the current span after message handled " + this.tracer.currentSpan());
 		}
 		finishSpan(ex);
 	}
--- a/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptorTest.java
+++ b/spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptorTest.java
@@ -29,8 +29,10 @@ import org.junit.After;
 import org.junit.Test;
 import zipkin2.Span;

+import org.springframework.amqp.support.AmqpHeaders;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.channel.QueueChannel;
+import org.springframework.kafka.support.KafkaHeaders;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
@@ -51,10 +53,11 @@ public class TracingChannelInterceptorTest {

 	List<Span> spans = new ArrayList<>();

-	ChannelInterceptor interceptor = TracingChannelInterceptor.create(Tracing.newBuilder()
-			.currentTraceContext(ThreadLocalCurrentTraceContext.newBuilder()
-					.addScopeDecorator(StrictScopeDecorator.create()).build())
-			.spanReporter(this.spans::add).build());
+	ChannelInterceptor interceptor = TracingChannelInterceptor
+			.create(Tracing.newBuilder()
+					.currentTraceContext(ThreadLocalCurrentTraceContext.newBuilder()
+							.addScopeDecorator(StrictScopeDecorator.create()).build())
+					.spanReporter(this.spans::add).build());

 	QueueChannel channel = new QueueChannel();

@@ -83,10 +86,9 @@ public class TracingChannelInterceptorTest {

 		this.channel.send(MessageBuilder.withPayload("foo").build());

-		assertThat(this.channel.receive().getHeaders()).containsKeys("X-B3-TraceId",
-				"X-B3-SpanId", "X-B3-Sampled", "nativeHeaders");
-		assertThat(this.spans).hasSize(1).flatExtracting(Span::kind)
-				.containsExactly(Span.Kind.PRODUCER);
+		assertThat(this.channel.receive().getHeaders()).containsKeys("X-B3-TraceId", "X-B3-SpanId", "X-B3-Sampled",
+				"nativeHeaders");
+		assertThat(this.spans).hasSize(1).flatExtracting(Span::kind).containsExactly(Span.Kind.PRODUCER);
 	}

 	@Test
@@ -96,10 +98,9 @@ public class TracingChannelInterceptorTest {
 		this.directChannel.send(MessageBuilder.withPayload("foo").build());

 		assertThat(this.message).isNotNull();
-		assertThat(this.message.getHeaders()).containsKeys("X-B3-TraceId", "X-B3-SpanId",
-				"X-B3-Sampled", "nativeHeaders");
-		assertThat(this.spans).flatExtracting(Span::kind).contains(Span.Kind.CONSUMER,
-				Span.Kind.PRODUCER);
+		assertThat(this.message.getHeaders()).containsKeys("X-B3-TraceId", "X-B3-SpanId", "X-B3-Sampled",
+				"nativeHeaders");
+		assertThat(this.spans).flatExtracting(Span::kind).contains(Span.Kind.CONSUMER, Span.Kind.PRODUCER);
 	}

 	@Test
@@ -108,9 +109,8 @@ public class TracingChannelInterceptorTest {

 		this.channel.send(MessageBuilder.withPayload("foo").build());

-		assertThat((Map) this.channel.receive().getHeaders().get(NATIVE_HEADERS))
-				.containsOnlyKeys("X-B3-TraceId", "X-B3-SpanId", "X-B3-Sampled",
-						"spanTraceId", "spanId", "spanSampled");
+		assertThat((Map) this.channel.receive().getHeaders().get(NATIVE_HEADERS)).containsOnlyKeys("X-B3-TraceId",
+				"X-B3-SpanId", "X-B3-Sampled", "spanTraceId", "spanId", "spanSampled");
 	}

 	/**
@@ -122,13 +122,11 @@ public class TracingChannelInterceptorTest {
 	public void producerConsidersOldSpanIds() {
 		this.channel.addInterceptor(producerSideOnly(this.interceptor));

-		this.channel.send(MessageBuilder.withPayload("foo")
-				.setHeader("X-B3-TraceId", "000000000000000a")
-				.setHeader("X-B3-ParentSpanId", "000000000000000a")
-				.setHeader("X-B3-SpanId", "000000000000000b").build());
+		this.channel.send(MessageBuilder.withPayload("foo").setHeader("X-B3-TraceId", "000000000000000a")
+				.setHeader("X-B3-ParentSpanId", "000000000000000a").setHeader("X-B3-SpanId", "000000000000000b")
+				.build());

-		assertThat(this.channel.receive().getHeaders()).containsEntry("X-B3-ParentSpanId",
-				"000000000000000b");
+		assertThat(this.channel.receive().getHeaders()).containsEntry("X-B3-ParentSpanId", "000000000000000b");
 	}

 	@Test
@@ -142,12 +140,10 @@ public class TracingChannelInterceptorTest {
 		accessor.setNativeHeader("X-B3-ParentSpanId", "000000000000000a");
 		accessor.setNativeHeader("X-B3-SpanId", "000000000000000b");

-		this.channel.send(MessageBuilder.withPayload("foo")
-				.copyHeaders(accessor.toMessageHeaders()).build());
+		this.channel.send(MessageBuilder.withPayload("foo").copyHeaders(accessor.toMessageHeaders()).build());

-		assertThat((Map) this.channel.receive().getHeaders().get(NATIVE_HEADERS))
-				.containsEntry("X-B3-ParentSpanId",
-						Collections.singletonList("000000000000000b"));
+		assertThat((Map) this.channel.receive().getHeaders().get(NATIVE_HEADERS)).containsEntry("X-B3-ParentSpanId",
+				Collections.singletonList("000000000000000b"));
 	}

 	/**
@@ -160,10 +156,9 @@ public class TracingChannelInterceptorTest {

 		this.channel.send(MessageBuilder.withPayload("foo").build());

-		assertThat(this.channel.receive().getHeaders()).containsKeys("X-B3-TraceId",
-				"X-B3-SpanId", "X-B3-Sampled", "nativeHeaders");
-		assertThat(this.spans).hasSize(1).flatExtracting(Span::kind)
-				.containsExactly(Span.Kind.CONSUMER);
+		assertThat(this.channel.receive().getHeaders()).containsKeys("X-B3-TraceId", "X-B3-SpanId", "X-B3-Sampled",
+				"nativeHeaders");
+		assertThat(this.spans).hasSize(1).flatExtracting(Span::kind).containsExactly(Span.Kind.CONSUMER);
 	}

 	@Test
@@ -172,9 +167,8 @@ public class TracingChannelInterceptorTest {

 		this.channel.send(MessageBuilder.withPayload("foo").build());

-		assertThat((Map) this.channel.receive().getHeaders().get(NATIVE_HEADERS))
-				.containsOnlyKeys("X-B3-TraceId", "X-B3-SpanId", "X-B3-Sampled",
-						"spanTraceId", "spanId", "spanSampled");
+		assertThat((Map) this.channel.receive().getHeaders().get(NATIVE_HEADERS)).containsOnlyKeys("X-B3-TraceId",
+				"X-B3-SpanId", "X-B3-Sampled", "spanTraceId", "spanId", "spanSampled");
 	}

 	@Test
@@ -186,10 +180,9 @@ public class TracingChannelInterceptorTest {

 		channel.send(MessageBuilder.withPayload("foo").build());

-		assertThat(messages.get(0).getHeaders()).doesNotContainKeys("X-B3-TraceId",
-				"X-B3-SpanId", "X-B3-Sampled", "nativeHeaders");
-		assertThat(this.spans).flatExtracting(Span::kind)
-				.containsExactly(Span.Kind.CONSUMER, null);
+		assertThat(messages.get(0).getHeaders()).doesNotContainKeys("X-B3-TraceId", "X-B3-SpanId", "X-B3-Sampled",
+				"nativeHeaders");
+		assertThat(this.spans).flatExtracting(Span::kind).containsExactly(Span.Kind.CONSUMER, null);
 	}

 	/**
@@ -206,8 +199,7 @@ public class TracingChannelInterceptorTest {

 		channel.send(MessageBuilder.withPayload("foo").build());

-		assertThat(messages.get(0).getHeaders()).doesNotContainKeys("X-B3-TraceId",
-				"X-B3-SpanId", "X-B3-Sampled");
+		assertThat(messages.get(0).getHeaders()).doesNotContainKeys("X-B3-TraceId", "X-B3-SpanId", "X-B3-Sampled");
 	}

 	@Test
@@ -219,8 +211,8 @@ public class TracingChannelInterceptorTest {

 		channel.send(MessageBuilder.withPayload("foo").build());

-		assertThat((Map) messages.get(0).getHeaders().get(NATIVE_HEADERS))
-				.doesNotContainKeys("X-B3-TraceId", "X-B3-SpanId", "X-B3-Sampled");
+		assertThat((Map) messages.get(0).getHeaders().get(NATIVE_HEADERS)).doesNotContainKeys("X-B3-TraceId",
+				"X-B3-SpanId", "X-B3-Sampled");
 	}

 	@Test
@@ -230,8 +222,8 @@ public class TracingChannelInterceptorTest {
 		this.channel.send(MessageBuilder.withPayload("foo").build());
 		this.channel.receive();

-		assertThat(this.spans).flatExtracting(Span::kind)
-				.containsExactlyInAnyOrder(Span.Kind.CONSUMER, Span.Kind.PRODUCER);
+		assertThat(this.spans).flatExtracting(Span::kind).containsExactlyInAnyOrder(Span.Kind.CONSUMER,
+				Span.Kind.PRODUCER);
 	}

 	@Test
@@ -243,8 +235,7 @@ public class TracingChannelInterceptorTest {

 		channel.send(MessageBuilder.withPayload("foo").build());

-		assertThat(this.spans).flatExtracting(Span::kind)
-				.containsExactly(Span.Kind.CONSUMER, null, Span.Kind.PRODUCER);
+		assertThat(this.spans).flatExtracting(Span::kind).containsExactly(Span.Kind.CONSUMER, null, Span.Kind.PRODUCER);
 	}

 	@Test
@@ -255,54 +246,41 @@ public class TracingChannelInterceptorTest {
 		Map<String, Object> errorChannelHeaders = new HashMap<>();
 		errorChannelHeaders.put(MessageHeaders.REPLY_CHANNEL, errorsReplyChannel);
 		errorChannelHeaders.put(MessageHeaders.ERROR_CHANNEL, errorsReplyChannel);
-		this.channel
-				.send(new ErrorMessage(
-						new MessagingException(MessageBuilder.withPayload("hi")
-								.setHeader(TraceMessageHeaders.TRACE_ID_NAME,
-										"000000000000000a")
-								.setHeader(TraceMessageHeaders.SPAN_ID_NAME,
-										"000000000000000a")
-								.setReplyChannel(deadReplyChannel)
-								.setErrorChannel(deadReplyChannel).build()),
-						errorChannelHeaders));
+		this.channel.send(new ErrorMessage(
+				new MessagingException(MessageBuilder.withPayload("hi")
+						.setHeader(TraceMessageHeaders.TRACE_ID_NAME, "000000000000000a")
+						.setHeader(TraceMessageHeaders.SPAN_ID_NAME, "000000000000000a")
+						.setReplyChannel(deadReplyChannel).setErrorChannel(deadReplyChannel).build()),
+				errorChannelHeaders));

 		this.message = this.channel.receive();

 		assertThat(this.message).isNotNull();
-		String spanId = this.message.getHeaders().get(TraceMessageHeaders.SPAN_ID_NAME,
-				String.class);
+		String spanId = this.message.getHeaders().get(TraceMessageHeaders.SPAN_ID_NAME, String.class);
 		assertThat(spanId).isNotNull();
-		String traceId = this.message.getHeaders().get(TraceMessageHeaders.TRACE_ID_NAME,
-				String.class);
+		String traceId = this.message.getHeaders().get(TraceMessageHeaders.TRACE_ID_NAME, String.class);
 		assertThat(traceId).isEqualTo("000000000000000a");
 		assertThat(spanId).isNotEqualTo("000000000000000a");
 		assertThat(this.spans).hasSize(2);
-		assertThat(this.message.getHeaders().getReplyChannel())
-				.isSameAs(errorsReplyChannel);
-		assertThat(this.message.getHeaders().getErrorChannel())
-				.isSameAs(errorsReplyChannel);
+		assertThat(this.message.getHeaders().getReplyChannel()).isSameAs(errorsReplyChannel);
+		assertThat(this.message.getHeaders().getErrorChannel()).isSameAs(errorsReplyChannel);
 	}

 	@Test
 	public void errorMessageOriginalMessageRetained() {
 		this.channel.addInterceptor(this.interceptor);
-		Message<?> originalMessage = MessageBuilder.withPayload("Hello")
-				.setHeader("header", "value").build();
-		Message<?> failedMessage = MessageBuilder.fromMessage(originalMessage)
-				.removeHeader("header").build();
-		this.channel.send(new ErrorMessage(new MessagingException(failedMessage),
-				originalMessage.getHeaders(), originalMessage));
+		Message<?> originalMessage = MessageBuilder.withPayload("Hello").setHeader("header", "value").build();
+		Message<?> failedMessage = MessageBuilder.fromMessage(originalMessage).removeHeader("header").build();
+		this.channel.send(
+				new ErrorMessage(new MessagingException(failedMessage), originalMessage.getHeaders(), originalMessage));

 		this.message = this.channel.receive();

 		assertThat(this.message).isNotNull();
-		assertThat(this.message).isInstanceOfSatisfying(ErrorMessage.class,
-				errorMessage -> {
-					assertThat(errorMessage.getOriginalMessage())
-							.isSameAs(originalMessage);
-					assertThat(errorMessage.getHeaders().get("header"))
-							.isEqualTo("value");
-				});
+		assertThat(this.message).isInstanceOfSatisfying(ErrorMessage.class, errorMessage -> {
+			assertThat(errorMessage.getOriginalMessage()).isSameAs(originalMessage);
+			assertThat(errorMessage.getHeaders().get("header")).isEqualTo("value");
+		});
 	}

 	@Test
@@ -311,22 +289,60 @@ public class TracingChannelInterceptorTest {
 		Map<String, Object> errorChannelHeaders = new HashMap<>();
 		errorChannelHeaders.put(TraceMessageHeaders.TRACE_ID_NAME, "000000000000000a");
 		errorChannelHeaders.put(TraceMessageHeaders.SPAN_ID_NAME, "000000000000000a");
-		this.channel.send(new ErrorMessage(new MessagingException("exception"),
-				errorChannelHeaders));
+		this.channel.send(new ErrorMessage(new MessagingException("exception"), errorChannelHeaders));

 		this.message = this.channel.receive();

 		assertThat(this.message).isNotNull();
-		String spanId = this.message.getHeaders().get(TraceMessageHeaders.SPAN_ID_NAME,
-				String.class);
+		String spanId = this.message.getHeaders().get(TraceMessageHeaders.SPAN_ID_NAME, String.class);
 		assertThat(spanId).isNotNull();
-		String traceId = this.message.getHeaders().get(TraceMessageHeaders.TRACE_ID_NAME,
-				String.class);
+		String traceId = this.message.getHeaders().get(TraceMessageHeaders.TRACE_ID_NAME, String.class);
 		assertThat(traceId).isEqualTo("000000000000000a");
 		assertThat(spanId).isNotEqualTo("000000000000000a");
 		assertThat(this.spans).hasSize(2);
 	}

+	@Test
+	public void should_store_kafka_as_remote_service_name_when_kafka_header_is_present() {
+		ExecutorSubscribableChannel channel = new ExecutorSubscribableChannel();
+		channel.addInterceptor(this.interceptor);
+		List<Message<?>> messages = new ArrayList<>();
+		channel.subscribe(messages::add);
+
+		Map<String, Object> headers = new HashMap<>();
+		headers.put(KafkaHeaders.MESSAGE_KEY, "hello");
+		channel.send(MessageBuilder.createMessage("foo", new MessageHeaders(headers)));
+
+		assertThat(this.spans).flatExtracting(Span::remoteServiceName).contains("kafka");
+	}
+
+	@Test
+	public void should_store_rabbitmq_as_remote_service_name_when_rabbit_header_is_present() {
+		ExecutorSubscribableChannel channel = new ExecutorSubscribableChannel();
+		channel.addInterceptor(this.interceptor);
+		List<Message<?>> messages = new ArrayList<>();
+		channel.subscribe(messages::add);
+
+		Map<String, Object> headers = new HashMap<>();
+		headers.put(AmqpHeaders.RECEIVED_ROUTING_KEY, "hello");
+		channel.send(MessageBuilder.createMessage("foo", new MessageHeaders(headers)));
+
+		assertThat(this.spans).flatExtracting(Span::remoteServiceName).contains("rabbitmq");
+	}
+
+	@Test
+	public void should_store_broker_as_remote_service_name_when_no_special_headers_were_found() {
+		ExecutorSubscribableChannel channel = new ExecutorSubscribableChannel();
+		channel.addInterceptor(this.interceptor);
+		List<Message<?>> messages = new ArrayList<>();
+		channel.subscribe(messages::add);
+
+		Map<String, Object> headers = new HashMap<>();
+		channel.send(MessageBuilder.createMessage("foo", new MessageHeaders(headers)));
+
+		assertThat(this.spans).flatExtracting(Span::remoteServiceName).containsOnly("broker", null);
+	}
+
 	ChannelInterceptor producerSideOnly(ChannelInterceptor delegate) {
 		return new ChannelInterceptorAdapter() {
 			@Override
@@ -335,8 +351,7 @@ public class TracingChannelInterceptorTest {
 			}

 			@Override
-			public void afterSendCompletion(Message<?> message, MessageChannel channel,
-					boolean sent, Exception ex) {
+			public void afterSendCompletion(Message<?> message, MessageChannel channel, boolean sent, Exception ex) {
 				delegate.afterSendCompletion(message, channel, sent, ex);
 			}
 		};
@@ -350,29 +365,24 @@ public class TracingChannelInterceptorTest {
 			}

 			@Override
-			public void afterReceiveCompletion(Message<?> message, MessageChannel channel,
-					Exception ex) {
+			public void afterReceiveCompletion(Message<?> message, MessageChannel channel, Exception ex) {
 				delegate.afterReceiveCompletion(message, channel, ex);
 			}
 		};
 	}

 	ExecutorChannelInterceptor executorSideOnly(ChannelInterceptor delegate) {
-		class ExecutorSideOnly extends ChannelInterceptorAdapter
-				implements ExecutorChannelInterceptor {
+		class ExecutorSideOnly extends ChannelInterceptorAdapter implements ExecutorChannelInterceptor {

 			@Override
-			public Message<?> beforeHandle(Message<?> message, MessageChannel channel,
-					MessageHandler handler) {
-				return ((ExecutorChannelInterceptor) delegate).beforeHandle(message,
-						channel, handler);
+			public Message<?> beforeHandle(Message<?> message, MessageChannel channel, MessageHandler handler) {
+				return ((ExecutorChannelInterceptor) delegate).beforeHandle(message, channel, handler);
 			}

 			@Override
-			public void afterMessageHandled(Message<?> message, MessageChannel channel,
-					MessageHandler handler, Exception ex) {
-				((ExecutorChannelInterceptor) delegate).afterMessageHandled(message,
-						channel, handler, ex);
+			public void afterMessageHandled(Message<?> message, MessageChannel channel, MessageHandler handler,
+					Exception ex) {
+				((ExecutorChannelInterceptor) delegate).afterMessageHandled(message, channel, handler, ex);
 			}

 		}