Merge branch 'main' into 3.1.x

# Conflicts:
#	benchmarks/pom.xml
#	pom.xml
#	spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TraceMessageHandler.java

benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/app/stream/SleuthBenchmarkingStreamApplication.java

@@ -23,6 +23,7 @@ import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
 
+import org.awaitility.Awaitility;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import reactor.core.publisher.Flux;
@@ -60,15 +61,16 @@ public class SleuthBenchmarkingStreamApplication {
 		// "DECORATE_ON_LAST");
 		// System.setProperty("spring.sleuth.reactor.instrumentation-type", "MANUAL");
 		System.setProperty("spring.sleuth.reactor.instrumentation-type", "DECORATE_QUEUES");
+		System.setProperty("spring.sleuth.integration.enabled", "true");
 		System.setProperty("spring.sleuth.function.type", "DECORATE_QUEUES");
 		ConfigurableApplicationContext context = SpringApplication.run(SleuthBenchmarkingStreamApplication.class, args);
-		for (int i = 0; i < 1; i++) {
-			InputDestination input = context.getBean(InputDestination.class);
-			input.send(MessageBuilder.withPayload("hello".getBytes())
-					.setHeader("b3", "4883117762eb9420-4883117762eb9420-1").build());
-			log.info("Retrieving the message for tests");
-			OutputDestination output = context.getBean(OutputDestination.class);
-			Message<byte[]> message = output.receive(200L);
+		InputDestination input = context.getBean(InputDestination.class);
+		input.send(MessageBuilder.withPayload("hello".getBytes())
+				.setHeader("b3", "4883117762eb9420-4883117762eb9420-1").build());
+		log.info("Retrieving the message for tests");
+		OutputDestination output = context.getBean(OutputDestination.class);
+		Awaitility.await().untilAsserted( () -> {
+			Message<byte[]> message = output.receive(1L);
 			log.info("Got the message from output");
 			assertThat(message).isNotNull();
 			log.info("Message is not null");
@@ -77,7 +79,9 @@ public class SleuthBenchmarkingStreamApplication {
 			String b3 = message.getHeaders().get("b3", String.class);
 			log.info("Checking the b3 header [" + b3 + "]");
 			assertThat(b3).startsWith("4883117762eb9420");
-		}
+		});
+		context.close();
+		System.exit(0);
 	}
 
 	@Bean

benchmarks/src/test/java/org/springframework/cloud/sleuth/benchmarks/jmh/stream/MicroBenchmarkStreamTests.java

@@ -23,6 +23,7 @@ import java.util.concurrent.TimeUnit;
 
 import brave.Tracing;
 import jmh.mbr.junit5.Microbenchmark;
+import org.awaitility.Awaitility;
 import org.junit.platform.commons.annotation.Testable;
 import org.openjdk.jmh.annotations.Benchmark;
 import org.openjdk.jmh.annotations.BenchmarkMode;
@@ -114,12 +115,12 @@ public class MicroBenchmarkStreamTests {
 
 		void run() {
 			sendInputMessage();
-			assertThatOutputMessageGotReceived();
+			Awaitility.await().untilAsserted(this::assertThatOutputMessageGotReceived);
 		}
 
 		private void assertThatOutputMessageGotReceived() {
 			// System.out.println("Retrieving the message for tests");
-			Message<byte[]> message = output.receive(200L);
+			Message<byte[]> message = output.receive(1L);
 			// System.out.println("Got the message from output");
 			assertThat(message).isNotNull();
 			// System.out.println("Message is not null");

pom.xml

@@ -67,7 +67,7 @@
 		<spring-cloud-gateway.version>3.0.4-SNAPSHOT</spring-cloud-gateway.version>
 		<spring-cloud-config.version>3.0.4-SNAPSHOT</spring-cloud-config.version>
 		<spring-cloud-circuitbreaker.version>2.0.3-SNAPSHOT</spring-cloud-circuitbreaker.version>
-		<spring-cloud-stream.version>3.1.3</spring-cloud-stream.version>
+		<spring-cloud-stream.version>3.1.4-SNAPSHOT</spring-cloud-stream.version>
 		<spring-cloud-function.version>3.1.4-SNAPSHOT</spring-cloud-function.version>
 		<spring-cloud-netflix.version>3.0.4-SNAPSHOT</spring-cloud-netflix.version>
 		<spring-cloud-openfeign.version>3.0.4-SNAPSHOT</spring-cloud-openfeign.version>

spring-cloud-sleuth-autoconfigure/src/main/java/org/springframework/cloud/sleuth/autoconfig/brave/BraveBaggageConfiguration.java

@@ -105,7 +105,11 @@ class BraveBaggageConfiguration {
 	// See #1643
 	@Bean
 	@ConditionalOnMissingBean
-	PropagationFactorySupplier defaultPropagationFactorySupplier() {
+	PropagationFactorySupplier defaultPropagationFactorySupplier(SleuthPropagationProperties properties) {
+		if (properties.getType().contains(PropagationType.CUSTOM)) {
+			throw new IllegalStateException(
+					"Please register a bean with the following signature [extends Propagation.Factory implements Propagation<String>] to override the default Sleuth behaviour or [implements PropagationFactorySupplier] to reuse it.");
+		}
 		return () -> B3Propagation.newFactoryBuilder().injectFormat(B3Propagation.Format.SINGLE_NO_PARENT).build();
 	}
 
@@ -131,7 +135,6 @@ class BraveBaggageConfiguration {
 			@Qualifier(PROPAGATION_KEYS) List<String> propagationKeys, SleuthBaggageProperties sleuthBaggageProperties,
 			SleuthPropagationProperties sleuthPropagationProperties, PropagationFactorySupplier supplier,
 			@Nullable List<BaggagePropagationCustomizer> baggagePropagationCustomizers) {
-
 		Set<String> localFields = redirectOldPropertyToNew(LOCAL_KEYS, localKeys, "spring.sleuth.baggage.local-fields",
 				sleuthBaggageProperties.getLocalFields());
 		for (String fieldName : localFields) {

spring-cloud-sleuth-autoconfigure/src/main/java/org/springframework/cloud/sleuth/autoconfig/instrument/messaging/TraceFunctionAutoConfiguration.java

@@ -16,19 +16,28 @@
 
 package org.springframework.cloud.sleuth.autoconfig.instrument.messaging;
 
+import java.util.ArrayList;
+import java.util.List;
+
+import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.boot.autoconfigure.AutoConfigureAfter;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingClass;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.cloud.context.scope.refresh.RefreshScopeRefreshedEvent;
 import org.springframework.cloud.function.context.catalog.FunctionAroundWrapper;
+import org.springframework.cloud.sleuth.Span;
 import org.springframework.cloud.sleuth.Tracer;
 import org.springframework.cloud.sleuth.autoconfig.brave.BraveAutoConfiguration;
+import org.springframework.cloud.sleuth.instrument.messaging.FunctionMessageSpanCustomizer;
 import org.springframework.cloud.sleuth.instrument.messaging.TraceFunctionAroundWrapper;
 import org.springframework.cloud.sleuth.propagation.Propagator;
+import org.springframework.cloud.stream.messaging.DirectWithAttributesChannel;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.core.env.Environment;
+import org.springframework.messaging.Message;
 import org.springframework.messaging.support.MessageHeaderAccessor;
 
 /**
@@ -48,8 +57,61 @@ public class TraceFunctionAutoConfiguration {
 
 	@Bean
 	TraceFunctionAroundWrapper traceFunctionAroundWrapper(Environment environment, Tracer tracer, Propagator propagator,
-			Propagator.Setter<MessageHeaderAccessor> injector, Propagator.Getter<MessageHeaderAccessor> extractor) {
-		return new TraceFunctionAroundWrapper(environment, tracer, propagator, injector, extractor);
+			Propagator.Setter<MessageHeaderAccessor> injector, Propagator.Getter<MessageHeaderAccessor> extractor,
+			ObjectProvider<List<FunctionMessageSpanCustomizer>> customizers) {
+		return new TraceFunctionAroundWrapper(environment, tracer, propagator, injector, extractor,
+				customizers.getIfAvailable(ArrayList::new));
+	}
+
+	@Configuration(proxyBeanMethods = false)
+	@ConditionalOnClass(DirectWithAttributesChannel.class)
+	static class TraceFunctionStreamConfiguration {
+
+		@Configuration(proxyBeanMethods = false)
+		@ConditionalOnClass(name = "org.springframework.cloud.stream.binder.kafka.config.KafkaBinderConfiguration")
+		@ConditionalOnMissingClass("org.springframework.cloud.stream.binder.rabbit.properties.RabbitBinderConfigurationProperties")
+		static class KafkaOnlyStreamConfiguration {
+
+			@Bean
+			FunctionMessageSpanCustomizer traceKafkaFunctionMessageSpanCustomizer() {
+				return new FunctionMessageSpanCustomizer() {
+					@Override
+					public void customizeInputMessageSpan(Span span, Message<?> message) {
+						span.remoteServiceName("kafka");
+					}
+
+					@Override
+					public void customizeOutputMessageSpan(Span span, Message<?> message) {
+						span.remoteServiceName("kafka");
+					}
+				};
+			}
+
+		}
+
+		@Configuration(proxyBeanMethods = false)
+		@ConditionalOnClass(
+				name = "org.springframework.cloud.stream.binder.rabbit.properties.RabbitBinderConfigurationProperties")
+		@ConditionalOnMissingClass("org.springframework.cloud.stream.binder.kafka.config.KafkaBinderConfiguration")
+		static class RabbitOnlyStreamConfiguration {
+
+			@Bean
+			FunctionMessageSpanCustomizer traceRabbitFunctionMessageSpanCustomizer() {
+				return new FunctionMessageSpanCustomizer() {
+					@Override
+					public void customizeInputMessageSpan(Span span, Message<?> message) {
+						span.remoteServiceName("rabbitmq");
+					}
+
+					@Override
+					public void customizeOutputMessageSpan(Span span, Message<?> message) {
+						span.remoteServiceName("rabbitmq");
+					}
+				};
+			}
+
+		}
+
 	}
 
 }

spring-cloud-sleuth-autoconfigure/src/main/java/org/springframework/cloud/sleuth/autoconfig/instrument/web/SkipPatternConfiguration.java

@@ -23,7 +23,7 @@ import java.util.StringJoiner;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 
-import org.springframework.beans.factory.BeanCurrentlyInCreationException;
+import org.springframework.beans.factory.BeanCreationException;
 import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.boot.actuate.autoconfigure.endpoint.web.WebEndpointProperties;
 import org.springframework.boot.actuate.autoconfigure.web.server.ConditionalOnManagementPort;
@@ -83,7 +83,7 @@ class SkipPatternConfiguration {
 			}
 			return () -> result;
 		}
-		catch (BeanCurrentlyInCreationException e) {
+		catch (BeanCreationException e) {
 			// Most likely, there is an actuator endpoint that indirectly references an
 			// instrumented HTTP client.
 			return () -> consolidateSkipPatterns(patterns);

spring-cloud-sleuth-autoconfigure/src/test/java/org/springframework/cloud/sleuth/autoconfig/brave/baggage/CustomPropagationFactoryTests.java

@@ -0,0 +1,102 @@
+/*
+ * Copyright 2013-2021 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.sleuth.autoconfig.brave.baggage;
+
+import java.util.Collections;
+import java.util.List;
+
+import brave.internal.propagation.StringPropagationAdapter;
+import brave.propagation.Propagation;
+import brave.propagation.TraceContext;
+import brave.propagation.TraceContextOrSamplingFlags;
+import org.assertj.core.api.BDDAssertions;
+import org.junit.jupiter.api.Test;
+
+import org.springframework.boot.actuate.autoconfigure.security.servlet.ManagementWebSecurityAutoConfiguration;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.autoconfigure.mongo.MongoAutoConfiguration;
+import org.springframework.boot.autoconfigure.quartz.QuartzAutoConfiguration;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+import org.springframework.cloud.gateway.config.GatewayAutoConfiguration;
+import org.springframework.cloud.gateway.config.GatewayClassPathWarningAutoConfiguration;
+import org.springframework.cloud.gateway.config.GatewayMetricsAutoConfiguration;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+public class CustomPropagationFactoryTests {
+
+	@Test
+	void should_fail_to_start_the_context_when_propagation_type_custom_and_no_custom_propagation_provided() {
+		new ApplicationContextRunner().withUserConfiguration(Config.class)
+				.withPropertyValues("spring.sleuth.propagation.type=custom")
+				.run(context -> BDDAssertions.then(context).hasFailed());
+	}
+
+	@Test
+	void should_start_the_context_when_propagation_type_custom_and_no_custom_propagation_provided() {
+		new ApplicationContextRunner().withUserConfiguration(CustomConfig.class)
+				.withPropertyValues("spring.sleuth.propagation.type=custom").run(context -> BDDAssertions.then(context)
+						.hasNotFailed().getBean(CustomConfig.CustomPropagation.class));
+	}
+
+	@Configuration(proxyBeanMethods = false)
+	@EnableAutoConfiguration(exclude = { GatewayClassPathWarningAutoConfiguration.class, GatewayAutoConfiguration.class,
+			GatewayMetricsAutoConfiguration.class, ManagementWebSecurityAutoConfiguration.class,
+			MongoAutoConfiguration.class, QuartzAutoConfiguration.class })
+	static class Config {
+
+	}
+
+	@Configuration(proxyBeanMethods = false)
+	@EnableAutoConfiguration(exclude = { GatewayClassPathWarningAutoConfiguration.class, GatewayAutoConfiguration.class,
+			GatewayMetricsAutoConfiguration.class, ManagementWebSecurityAutoConfiguration.class,
+			MongoAutoConfiguration.class, QuartzAutoConfiguration.class })
+	static class CustomConfig {
+
+		@Bean
+		CustomPropagation customPropagation() {
+			return new CustomPropagation();
+		}
+
+		static class CustomPropagation extends Propagation.Factory implements Propagation<String> {
+
+			@Override
+			public List<String> keys() {
+				return Collections.emptyList();
+			}
+
+			@Override
+			public <R> TraceContext.Injector<R> injector(Setter<R, String> setter) {
+				return (traceContext, request) -> {
+				};
+			}
+
+			@Override
+			public <R> TraceContext.Extractor<R> extractor(Getter<R, String> getter) {
+				return request -> TraceContextOrSamplingFlags.EMPTY;
+			}
+
+			@Override
+			public <K> Propagation<K> create(KeyFactory<K> keyFactory) {
+				return StringPropagationAdapter.create(this, keyFactory);
+			}
+
+		}
+
+	}
+
+}

spring-cloud-sleuth-brave/src/main/java/org/springframework/cloud/sleuth/brave/bridge/BravePropagator.java

@@ -54,7 +54,6 @@ public class BravePropagator implements Propagator {
 	public <C> Span.Builder extract(C carrier, Getter<C> getter) {
 		TraceContextOrSamplingFlags extract = this.tracing.propagation().extractor(getter::get).extract(carrier);
 		if (extract.samplingFlags() == SamplingFlags.EMPTY) {
-			this.tracing.tracer().nextSpan();
 			return new BraveSpanBuilder(this.tracing.tracer());
 		}
 		return BraveSpanBuilder.toBuilder(this.tracing.tracer(), extract);

spring-cloud-sleuth-brave/src/main/java/org/springframework/cloud/sleuth/brave/bridge/CompositePropagationFactorySupplier.java

@@ -84,7 +84,7 @@ class CompositePropagationFactory extends Propagation.Factory implements Propaga
 		W3CPropagation w3CPropagation = new W3CPropagation(braveBaggageManager, localFields);
 		this.mapping.put(PropagationType.W3C, new AbstractMap.SimpleEntry<>(w3CPropagation, w3CPropagation.get()));
 		LazyPropagationFactory lazyPropagationFactory = new LazyPropagationFactory(
-				beanFactory.getBeanProvider(Factory.class));
+				beanFactory.getBeanProvider(PropagationFactorySupplier.class));
 		this.mapping.put(PropagationType.CUSTOM,
 				new AbstractMap.SimpleEntry<>(lazyPropagationFactory, lazyPropagationFactory.get()));
 	}
@@ -161,17 +161,17 @@ class CompositePropagationFactory extends Propagation.Factory implements Propaga
 	@SuppressWarnings("unchecked")
 	private static final class LazyPropagationFactory extends Propagation.Factory {
 
-		private final ObjectProvider<Propagation.Factory> delegate;
+		private final ObjectProvider<PropagationFactorySupplier> delegate;
 
 		private volatile Propagation.Factory propagationFactory;
 
-		private LazyPropagationFactory(ObjectProvider<Propagation.Factory> delegate) {
+		private LazyPropagationFactory(ObjectProvider<PropagationFactorySupplier> delegate) {
 			this.delegate = delegate;
 		}
 
 		private Propagation.Factory propagationFactory() {
 			if (this.propagationFactory == null) {
-				this.propagationFactory = this.delegate.getIfAvailable(() -> NoOpPropagation.INSTANCE);
+				this.propagationFactory = this.delegate.getIfAvailable(() -> () -> NoOpPropagation.INSTANCE).get();
 			}
 			return this.propagationFactory;
 		}

spring-cloud-sleuth-brave/src/main/java/org/springframework/cloud/sleuth/brave/propagation/PropagationType.java

@@ -40,7 +40,8 @@ public enum PropagationType {
 	W3C,
 
 	/**
-	 * Custom propagation type.
+	 * Custom propagation type. If picked, requires bean registration overriding the
+	 * default propagation mechanisms.
 	 */
 	CUSTOM
 

spring-cloud-sleuth-brave/src/test/java/org/springframework/cloud/sleuth/brave/bridge/CompositePropagationFactorySupplierTests.java

@@ -1,120 +0,0 @@
-/*
- * Copyright 2013-2021 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.
- * You may obtain a copy of the License at
- *
- *      https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.cloud.sleuth.brave.bridge;
-
-import java.util.Collections;
-import java.util.List;
-import java.util.Objects;
-
-import brave.internal.codec.HexCodec;
-import brave.internal.propagation.StringPropagationAdapter;
-import brave.propagation.Propagation;
-import brave.propagation.TraceContext;
-import brave.propagation.TraceContextOrSamplingFlags;
-import org.assertj.core.api.BDDAssertions;
-import org.junit.jupiter.api.Test;
-import org.mockito.Mockito;
-
-import org.springframework.beans.factory.BeanFactory;
-import org.springframework.cloud.loadbalancer.support.SimpleObjectProvider;
-import org.springframework.cloud.sleuth.brave.propagation.PropagationType;
-import org.springframework.util.StringUtils;
-
-class CompositePropagationFactorySupplierTests {
-
-	@Test
-	void should_pick_custom_registered_propagation_when_custom_mode_picked() {
-		BeanFactory beanFactory = Mockito.mock(BeanFactory.class);
-		Mockito.when(beanFactory.getBeanProvider(BraveBaggageManager.class))
-				.thenReturn(new SimpleObjectProvider(new BraveBaggageManager()));
-		Mockito.when(beanFactory.getBeanProvider(Propagation.Factory.class))
-				.thenReturn(new SimpleObjectProvider(new CustomTracePropagation()));
-		Mockito.when(beanFactory.getBeanProvider(Propagation.class))
-				.thenReturn(new SimpleObjectProvider(new CustomTracePropagation()));
-
-		CompositePropagationFactorySupplier supplier = new CompositePropagationFactorySupplier(beanFactory,
-				Collections.emptyList(), Collections.singletonList(PropagationType.CUSTOM));
-
-		BDDAssertions.then(supplier.get().get().keys()).containsExactly(CustomTraceExtractor.CUSTOM_TRACE_HEADER);
-	}
-
-}
-
-class CustomTracePropagation extends Propagation.Factory implements Propagation<String> {
-
-	public static final List<String> KEYS = Collections.singletonList(CustomTraceExtractor.CUSTOM_TRACE_HEADER);
-
-	@Override
-	public List<String> keys() {
-		return KEYS;
-	}
-
-	@Override
-	public <R> TraceContext.Injector<R> injector(Setter<R, String> setter) {
-		return (traceContext, request) -> {
-			String trace = traceContext.traceIdString() + ":" + traceContext.spanIdString();
-			setter.put(request, CustomTraceExtractor.CUSTOM_TRACE_HEADER, trace);
-		};
-	}
-
-	@Override
-	public <R> TraceContext.Extractor<R> extractor(Getter<R, String> getter) {
-		Objects.requireNonNull(getter);
-		return new CustomTraceExtractor<>(getter);
-	}
-
-	@Override
-	public <K> Propagation<K> create(KeyFactory<K> keyFactory) {
-		return StringPropagationAdapter.create(this, keyFactory);
-	}
-
-}
-
-class CustomTraceExtractor<R> implements TraceContext.Extractor<R> {
-
-	static final String CUSTOM_TRACE_HEADER = "x-custom-trace";
-
-	final Propagation.Getter<R, String> getter;
-
-	CustomTraceExtractor(Propagation.Getter<R, String> getter) {
-		this.getter = getter;
-	}
-
-	@Override
-	@SuppressWarnings("ReturnCount")
-	public TraceContextOrSamplingFlags extract(R request) {
-		String traceString = getter.get(request, CUSTOM_TRACE_HEADER);
-		if (!StringUtils.hasText(traceString)) {
-			return TraceContextOrSamplingFlags.EMPTY;
-		}
-		String[] trace = traceString.split(":");
-		if (trace.length != 2) {
-			return TraceContextOrSamplingFlags.EMPTY;
-		}
-
-		try {
-			TraceContext traceContext = TraceContext.newBuilder().traceId(HexCodec.lowerHexToUnsignedLong(trace[0]))
-					.spanId(HexCodec.lowerHexToUnsignedLong(trace[1])).build();
-
-			return TraceContextOrSamplingFlags.create(traceContext);
-		}
-		catch (NumberFormatException ex) {
-			return TraceContextOrSamplingFlags.EMPTY;
-		}
-	}
-
-}

spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/FunctionMessageSpanCustomizer.java

@@ -0,0 +1,58 @@
+/*
+ * Copyright 2013-2021 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.sleuth.instrument.messaging;
+
+import org.springframework.cloud.sleuth.Span;
+import org.springframework.messaging.Message;
+
+/**
+ * Allows customization of messaging spans for Spring Cloud Function instrumentation.
+ *
+ * @author Marcin Grzejszczak
+ * @since 3.0.4
+ */
+public interface FunctionMessageSpanCustomizer {
+
+	/**
+	 * Customizes the span created after wrapping the input message in a span
+	 * representation.
+	 * @param span current span to customize
+	 * @param message received or sent message
+	 */
+	default void customizeInputMessageSpan(Span span, Message<?> message) {
+
+	}
+
+	/**
+	 * Customizes the span wrapping the function execution.
+	 * @param span current span to customize
+	 * @param message message to be sent
+	 */
+	default void customizeFunctionSpan(Span span, Message<?> message) {
+
+	}
+
+	/**
+	 * Customizes the span created for the output message.
+	 * @param span current span to customize
+	 * @param message message to be sent
+	 */
+	default void customizeOutputMessageSpan(Span span, Message<?> message) {
+
+	}
+
+}

spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/MessagingSleuthOperators.java

@@ -188,6 +188,23 @@ public final class MessagingSleuthOperators {
 	 * @return instrumented message
 	 */
 	public static <T> Message<T> handleOutputMessage(BeanFactory beanFactory, Message<T> message, Throwable throwable) {
+		return handleOutputMessage(beanFactory, message, span -> {
+		}, throwable);
+	}
+
+	/**
+	 * Creates an output message with tracer headers and reports the corresponding
+	 * producer span. If the message contains a header called {@code destination} it will
+	 * be used to tag the span with destination name.
+	 * @param beanFactory - bean factory
+	 * @param message - message to which tracer headers should be injected
+	 * @param spanCustomizer - customizer of the output span
+	 * @param throwable - exception that took place while processing the message
+	 * @param <T> - message payload
+	 * @return instrumented message
+	 */
+	public static <T> Message<T> handleOutputMessage(BeanFactory beanFactory, Message<T> message,
+			Consumer<Span> spanCustomizer, Throwable throwable) {
 		TraceMessageHandler traceMessageHandler = TraceMessageHandler.forNonSpringIntegration(beanFactory);
 		Span span = traceMessageHandler.parentSpan(message);
 		span = span != null ? span : traceMessageHandler.consumerSpan(message);
@@ -198,6 +215,7 @@ public final class MessagingSleuthOperators {
 		}
 		MessageAndSpan messageAndSpan = traceMessageHandler.wrapOutputMessage(message, span,
 				String.valueOf(message.getHeaders().getOrDefault("destination", "")));
+		spanCustomizer.accept(messageAndSpan.span);
 		traceMessageHandler.afterMessageHandled(messageAndSpan.span, throwable);
 		return messageAndSpan.msg;
 	}

spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TraceFunctionAroundWrapper.java

@@ -16,6 +16,8 @@
 
 package org.springframework.cloud.sleuth.instrument.messaging;
 
+import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 
@@ -58,17 +60,26 @@ public class TraceFunctionAroundWrapper extends FunctionAroundWrapper
 
 	private final TraceMessageHandler traceMessageHandler;
 
+	private final List<FunctionMessageSpanCustomizer> customizers;
+
 	final Map<String, String> functionToDestinationCache = new ConcurrentHashMap<>();
 
 	public TraceFunctionAroundWrapper(Environment environment, Tracer tracer, Propagator propagator,
 			Propagator.Setter<MessageHeaderAccessor> injector, Propagator.Getter<MessageHeaderAccessor> extractor) {
+		this(environment, tracer, propagator, injector, extractor, Collections.emptyList());
+	}
+
+	public TraceFunctionAroundWrapper(Environment environment, Tracer tracer, Propagator propagator,
+			Propagator.Setter<MessageHeaderAccessor> injector, Propagator.Getter<MessageHeaderAccessor> extractor,
+			List<FunctionMessageSpanCustomizer> customizers) {
 		this.environment = environment;
 		this.tracer = tracer;
 		this.propagator = propagator;
 		this.injector = injector;
 		this.extractor = extractor;
+		this.customizers = customizers;
 		this.traceMessageHandler = TraceMessageHandler.forNonSpringIntegration(this.tracer, this.propagator,
-				this.injector, this.extractor);
+				this.injector, this.extractor, this.customizers);
 	}
 
 	@Override
@@ -76,23 +87,26 @@ public class TraceFunctionAroundWrapper extends FunctionAroundWrapper
 		MessageAndSpans invocationMessage = null;
 		Span span;
 		if (message == null && targetFunction.isSupplier()) { // Supplier
-			span = traceMessageHandler.tracer.nextSpan().name(targetFunction.getFunctionDefinition());
+			if (log.isDebugEnabled()) {
+				log.debug("Creating a span for a supplier");
+			}
+			span = this.tracer.nextSpan().name(targetFunction.getFunctionDefinition());
+			customizedInputMessageSpan(span, null);
 		}
 		else {
 			if (log.isDebugEnabled()) {
 				log.debug("Will retrieve the tracing headers from the message");
 			}
-			invocationMessage = traceMessageHandler.wrapInputMessage(message,
+			invocationMessage = this.traceMessageHandler.wrapInputMessage(message,
 					inputDestination(targetFunction.getFunctionDefinition()));
 			if (log.isDebugEnabled()) {
 				log.debug("Wrapped input msg " + invocationMessage);
 			}
 			span = invocationMessage.childSpan;
 		}
-
 		Object result;
 		Throwable throwable = null;
-		try (Tracer.SpanInScope ws = tracer.withSpan(span.start())) {
+		try (Tracer.SpanInScope ws = this.tracer.withSpan(span.start())) {
 			result = invocationMessage == null ? targetFunction.get() : targetFunction.apply(invocationMessage.msg);
 		}
 		catch (Exception e) {
@@ -100,7 +114,7 @@ public class TraceFunctionAroundWrapper extends FunctionAroundWrapper
 			throw e;
 		}
 		finally {
-			traceMessageHandler.afterMessageHandled(span, throwable);
+			this.traceMessageHandler.afterMessageHandled(span, throwable);
 		}
 		if (result == null) {
 			if (log.isDebugEnabled()) {
@@ -109,10 +123,12 @@ public class TraceFunctionAroundWrapper extends FunctionAroundWrapper
 			return null;
 		}
 		Message<?> msgResult = toMessage(result);
-
 		MessageAndSpan wrappedOutputMessage;
+		if (log.isDebugEnabled()) {
+			log.debug("Will instrument the output message");
+		}
 		if (invocationMessage != null) {
-			wrappedOutputMessage = traceMessageHandler.wrapOutputMessage(msgResult, invocationMessage.parentSpan,
+			wrappedOutputMessage = this.traceMessageHandler.wrapOutputMessage(msgResult, invocationMessage.parentSpan,
 					outputDestination(targetFunction.getFunctionDefinition()));
 		}
 		else {
@@ -129,6 +145,10 @@ public class TraceFunctionAroundWrapper extends FunctionAroundWrapper
 		return traceMessageHandler.wrapOutputMessage(resultMessage, spanFromMessage, outputDestination(name));
 	}
 
+	private void customizedInputMessageSpan(Span spanToCustomize, Message<?> msg) {
+		this.customizers.forEach(cust -> cust.customizeInputMessageSpan(spanToCustomize, msg));
+	}
+
 	private Message<?> toMessage(Object result) {
 		if (!(result instanceof Message)) {
 			return MessageBuilder.withPayload(result).build();

spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TraceMessageHandler.java

@@ -81,10 +81,12 @@ class TraceMessageHandler {
 
 	private final Function<Span, Span.Builder> outputMessageSpanFunction;
 
+	private final List<FunctionMessageSpanCustomizer> customizers;
+
 	TraceMessageHandler(Tracer tracer, Propagator propagator, Propagator.Setter<MessageHeaderAccessor> injector,
 			Propagator.Getter<MessageHeaderAccessor> extractor, Function<Span, Span> preSendFunction,
 			TriConsumer<MessageHeaderAccessor, Span, Span> preSendMessageManipulator,
-			Function<Span, Span.Builder> outputMessageSpanFunction) {
+			Function<Span, Span.Builder> outputMessageSpanFunction, List<FunctionMessageSpanCustomizer> customizers) {
 		this.tracer = tracer;
 		this.propagator = propagator;
 		this.injector = injector;
@@ -93,19 +95,21 @@ class TraceMessageHandler {
 		this.preSendFunction = preSendFunction;
 		this.preSendMessageManipulator = preSendMessageManipulator;
 		this.outputMessageSpanFunction = outputMessageSpanFunction;
+		this.customizers = customizers;
 	}
 
 	static TraceMessageHandler forNonSpringIntegration(Tracer tracer, Propagator propagator,
-			Propagator.Setter<MessageHeaderAccessor> injector, Propagator.Getter<MessageHeaderAccessor> extractor) {
+			Propagator.Setter<MessageHeaderAccessor> injector, Propagator.Getter<MessageHeaderAccessor> extractor,
+			List<FunctionMessageSpanCustomizer> customizers) {
 		Function<Span, Span> preSendFunction = span -> SleuthMessagingSpan.MESSAGING_SPAN.wrap(tracer.nextSpan(span))
-				.name("handle").start();
+				.name("function").start();
 		TriConsumer<MessageHeaderAccessor, Span, Span> preSendMessageManipulator = (headers, parentSpan, childSpan) -> {
 			headers.setHeader("traceHandlerParentSpan", parentSpan);
 			headers.setHeader(Span.class.getName(), childSpan);
 		};
 		Function<Span, Span.Builder> postReceiveFunction = span -> tracer.spanBuilder().setParent(span.context());
 		return new TraceMessageHandler(tracer, propagator, injector, extractor, preSendFunction,
-				preSendMessageManipulator, postReceiveFunction);
+				preSendMessageManipulator, postReceiveFunction, customizers);
 	}
 
 	@SuppressWarnings("unchecked")
@@ -113,7 +117,7 @@ class TraceMessageHandler {
 		Propagator.Setter<MessageHeaderAccessor> setter = firstBeanOrException(beanFactory, Propagator.Setter.class);
 		Propagator.Getter<MessageHeaderAccessor> getter = firstBeanOrException(beanFactory, Propagator.Getter.class);
 		return forNonSpringIntegration(beanFactory.getBean(Tracer.class), beanFactory.getBean(Propagator.class), setter,
-				getter);
+				getter, customizers(beanFactory));
 	}
 
 	private static <T> T firstBeanOrException(BeanFactory beanFactory, Class<T> clazz) {
@@ -126,6 +130,16 @@ class TraceMessageHandler {
 		return object;
 	}
 
+	private static List<FunctionMessageSpanCustomizer> customizers(BeanFactory beanFactory) {
+		List<FunctionMessageSpanCustomizer> customizers = new ArrayList<>();
+		ObjectProvider<FunctionMessageSpanCustomizer> provider = beanFactory
+				.getBeanProvider(FunctionMessageSpanCustomizer.class);
+		for (FunctionMessageSpanCustomizer functionMessageSpanCustomizer : provider) {
+			customizers.add(functionMessageSpanCustomizer);
+		}
+		return customizers;
+	}
+
 	/**
 	 * Wraps the given input message with tracing headers and returns a corresponding
 	 * span.
@@ -135,42 +149,35 @@ class TraceMessageHandler {
 	 */
 	MessageAndSpans wrapInputMessage(Message<?> message, String destinationName) {
 		MessageHeaderAccessor headers = mutableHeaderAccessor(message);
-		Span extracted = this.propagator.extract(headers, this.extractor).start();
-		// Start and finish a consumer span as we will immediately process it.
 		Span.Builder consumerSpanBuilder = SleuthMessagingSpan.MESSAGING_SPAN
-				.wrap(this.tracer.spanBuilder().setParent(extracted.context()));
-		Span consumerSpan = consumerSpan(destinationName, extracted, consumerSpanBuilder);
-		// create and scope a span for the message processor
-		Span span = this.preSendFunction.apply(consumerSpan);
-		// 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.
-		clearTracingHeaders(headers);
-		this.preSendMessageManipulator.accept(headers, consumerSpan, span);
+				.wrap(this.propagator.extract(headers, this.extractor));
+		Span consumerSpan = consumerSpan(destinationName, consumerSpanBuilder, message);
+
 		if (log.isDebugEnabled()) {
-			log.debug("Created a handle span after retrieving the message " + consumerSpanBuilder);
+			log.debug("Built a consumer span " + consumerSpan);
 		}
+		Span childSpan = this.preSendFunction.apply(consumerSpan);
+		clearTracingHeaders(headers);
+		this.preSendMessageManipulator.accept(headers, consumerSpan, childSpan);
+		this.customizers.forEach(customizer -> customizer.customizeFunctionSpan(childSpan, message));
 		if (message instanceof ErrorMessage) {
 			return new MessageAndSpans(new ErrorMessage((Throwable) message.getPayload(), headers.getMessageHeaders()),
-					consumerSpan, span);
+					consumerSpan, childSpan);
 		}
 		headers.setImmutable();
 		return new MessageAndSpans(new GenericMessage<>(message.getPayload(), headers.getMessageHeaders()),
-				consumerSpan, span);
+				consumerSpan, childSpan);
 	}
 
-	private Span consumerSpan(String destinationName, Span extracted, Span.Builder consumerSpanBuilder) {
-		Span consumerSpan;
-		if (!extracted.isNoop()) {
-			consumerSpanBuilder.kind(Span.Kind.CONSUMER).start();
-			addTags(consumerSpanBuilder, destinationName);
-			consumerSpanBuilder.remoteServiceName(REMOTE_SERVICE_NAME);
-			consumerSpan = consumerSpanBuilder.start();
-			consumerSpan.end();
-		}
-		else {
-			consumerSpan = consumerSpanBuilder.start();
-		}
+	private Span consumerSpan(String destinationName, Span.Builder consumerSpanBuilder, Message<?> message) {
+		consumerSpanBuilder.kind(Span.Kind.CONSUMER).name("handle");
+		addTags(consumerSpanBuilder, destinationName);
+		consumerSpanBuilder.remoteServiceName(REMOTE_SERVICE_NAME);
+		// this is the consumer part of the producer->consumer mechanism
+		Span consumerSpan = consumerSpanBuilder.start();
+		this.customizers.forEach(customizer -> customizer.customizeInputMessageSpan(consumerSpan, message));
+		// we're ending this immediately just to have a properly nested graph
+		consumerSpan.end();
 		return consumerSpan;
 	}
 
@@ -230,7 +237,7 @@ class TraceMessageHandler {
 		MessageHeaderAccessor headers = mutableHeaderAccessor(retrievedMessage);
 		Span.Builder span = this.outputMessageSpanFunction.apply(parentSpan);
 		clearTracingHeaders(headers);
-		Span producerSpan = createProducerSpan(headers, span, destinationName);
+		Span producerSpan = createProducerSpan(headers, span, destinationName, message);
 		this.propagator.inject(producerSpan.context(), headers, this.injector);
 		if (log.isDebugEnabled()) {
 			log.debug("Created a new span output message " + span);
@@ -238,12 +245,14 @@ class TraceMessageHandler {
 		return new MessageAndSpan(outputMessage(message, retrievedMessage, headers), producerSpan);
 	}
 
-	private Span createProducerSpan(MessageHeaderAccessor headers, Span.Builder spanBuilder, String destinationName) {
+	private Span createProducerSpan(MessageHeaderAccessor headers, Span.Builder spanBuilder, String destinationName,
+			Message<?> message) {
 		spanBuilder.kind(Span.Kind.PRODUCER).name("send").remoteServiceName(toRemoteServiceName(headers));
 		Span span = spanBuilder.start();
 		if (!span.isNoop()) {
 			addTags(spanBuilder, destinationName);
 		}
+		this.customizers.forEach(customizer -> customizer.customizeOutputMessageSpan(span, message));
 		return span;
 	}
 

spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptor.java

@@ -109,7 +109,6 @@ public final class TracingChannelInterceptor implements ExecutorChannelIntercept
 	public TracingChannelInterceptor(Tracer tracer, Propagator propagator,
 			Propagator.Setter<MessageHeaderAccessor> setter, Propagator.Getter<MessageHeaderAccessor> getter,
 			Function<String, String> remoteServiceNameMapper, MessageSpanCustomizer messageSpanCustomizer) {
-
 		this.tracer = tracer;
 		this.propagator = propagator;
 		this.injector = setter;

spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/rsocket/TracingResponderRSocketProxy.java

@@ -159,8 +159,7 @@ public class TracingResponderRSocketProxy extends RSocketProxy {
 					traceId = EncodingUtils.fromLong(traceIdHigh) + traceId;
 				}
 				TraceContext.Builder parentBuilder = this.tracer.traceContextBuilder()
-						.sampled(tracingMetadata.isDebug() || tracingMetadata.isSampled())
-						.traceId(traceId)
+						.sampled(tracingMetadata.isDebug() || tracingMetadata.isSampled()).traceId(traceId)
 						.spanId(EncodingUtils.fromLong(tracingMetadata.spanId()))
 						.parentId(EncodingUtils.fromLong(tracingMetadata.parentId()));
 				return builder.setParent(parentBuilder.build());

tests/common/src/main/java/org/springframework/cloud/sleuth/brave/BraveTestSpanHandler.java

@@ -18,6 +18,7 @@ package org.springframework.cloud.sleuth.brave;
 
 import java.util.Iterator;
 import java.util.List;
+import java.util.Queue;
 import java.util.stream.Collectors;
 
 import brave.test.IntegrationTestSpanHandler;
@@ -27,6 +28,8 @@ import org.springframework.cloud.sleuth.brave.bridge.BraveAccessor;
 import org.springframework.cloud.sleuth.exporter.FinishedSpan;
 import org.springframework.cloud.sleuth.test.TestSpanHandler;
 
+import static org.assertj.core.api.BDDAssertions.then;
+
 public class BraveTestSpanHandler implements TestSpanHandler {
 
 	final brave.test.TestSpanHandler spans;
@@ -81,6 +84,33 @@ public class BraveTestSpanHandler implements TestSpanHandler {
 		return BraveAccessor.finishedSpan(this.spans.get(index));
 	}
 
+	@Override
+	public void assertAllSpansWereFinishedOrAbandoned(Queue<Span> createdSpans) {
+		List<FinishedSpan> finishedSpans = reportedSpans();
+		then(finishedSpans).as("There should be that many finished spans as many created ones")
+				.hasSize(createdSpans.size());
+		// finished -> a,b,c ; created -> b,c,d => matchedFinished = b,c
+		List<FinishedSpan> matchedFinishedSpans = finishedSpans.stream()
+				.filter(f -> createdSpans.stream().anyMatch(cs -> f.getSpanId().equals(cs.context().spanId())))
+				.collect(Collectors.toList());
+		// finished -> a,b,c ; created -> b,c,d => matchedCreated = b,c
+		List<Span> matchedCreatedSpans = createdSpans.stream()
+				.filter(cs -> finishedSpans.stream().anyMatch(f -> cs.context().spanId().equals(f.getSpanId())))
+				.collect(Collectors.toList());
+		// finished -> a,b,c ; created -> b,c,d => missingFinished = a
+		List<FinishedSpan> missingFinishedSpans = finishedSpans.stream()
+				.filter(f -> matchedFinishedSpans.stream().noneMatch(m -> m.getSpanId().equals(f.getSpanId())))
+				.collect(Collectors.toList());
+		// finished -> a,b,c ; created -> b,c,d => missingCreated = d
+		List<Span> missingCreatedSpans = createdSpans.stream().filter(
+				f -> matchedCreatedSpans.stream().noneMatch(m -> m.context().spanId().equals(f.context().spanId())))
+				.collect(Collectors.toList());
+		if (!missingFinishedSpans.isEmpty() || !missingCreatedSpans.isEmpty()) {
+			throw new AssertionError("There were unmatched created spans " + missingCreatedSpans
+					+ " and/or finished span " + missingFinishedSpans);
+		}
+	}
+
 	@Override
 	public Iterator<FinishedSpan> iterator() {
 		return reportedSpans().iterator();

tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TraceFunctionAroundWrapperTests.java

@@ -26,6 +26,8 @@ import org.springframework.boot.builder.SpringApplicationBuilder;
 import org.springframework.cloud.function.context.FunctionCatalog;
 import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry.FunctionInvocationWrapper;
 import org.springframework.cloud.sleuth.test.TestSpanHandler;
+import org.springframework.cloud.sleuth.test.TestTracer;
+import org.springframework.cloud.sleuth.test.TestTracingBeanPostProcessor;
 import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.messaging.Message;
@@ -48,11 +50,13 @@ public abstract class TraceFunctionAroundWrapperTests {
 			assertThat(spanHandler.reportedSpans()).isEmpty();
 			FunctionCatalog catalog = context.getBean(FunctionCatalog.class);
 			FunctionInvocationWrapper function = catalog.lookup("greeter");
-			function.setSkipOutputConversion(true);
+
 			Message<?> result = (Message<?>) function.get();
-			assertThat(result.getPayload()).isEqualTo("hello");
+
+			assertThat(result.getPayload()).isEqualTo("hello".getBytes());
 			assertThat(spanHandler.reportedSpans().size()).isEqualTo(2);
 			assertThat(((String) result.getHeaders().get("b3"))).contains(spanHandler.get(0).getTraceId());
+			spanHandler.assertAllSpansWereFinishedOrAbandoned(context.getBean(TestTracer.class).createdSpans());
 		}
 	}
 
@@ -65,11 +69,13 @@ public abstract class TraceFunctionAroundWrapperTests {
 			assertThat(spanHandler.reportedSpans()).isEmpty();
 			FunctionCatalog catalog = context.getBean(FunctionCatalog.class);
 			FunctionInvocationWrapper function = catalog.lookup("uppercase");
-			function.setSkipOutputConversion(true);
+
 			Message<?> result = (Message<?>) function.apply(MessageBuilder.withPayload("hello").build());
+
 			assertThat(result.getPayload()).isEqualTo("HELLO");
 			assertThat(spanHandler.reportedSpans().size()).isEqualTo(3);
 			assertThat(((String) result.getHeaders().get("b3"))).contains(spanHandler.get(0).getTraceId());
+			spanHandler.assertAllSpansWereFinishedOrAbandoned(context.getBean(TestTracer.class).createdSpans());
 		}
 	}
 
@@ -88,6 +94,11 @@ public abstract class TraceFunctionAroundWrapperTests {
 			return v -> v.toUpperCase();
 		}
 
+		@Bean
+		static TestTracingBeanPostProcessor testTracerBeanPostProcessor() {
+			return new TestTracingBeanPostProcessor();
+		}
+
 	}
 
 };

tests/common/src/main/java/org/springframework/cloud/sleuth/test/TestPropagator.java

@@ -0,0 +1,54 @@
+/*
+ * Copyright 2013-2021 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.sleuth.test;
+
+import java.util.List;
+
+import org.springframework.cloud.sleuth.Span;
+import org.springframework.cloud.sleuth.TraceContext;
+import org.springframework.cloud.sleuth.propagation.Propagator;
+
+/**
+ * {@link Propagator} that stores information about started spans.
+ */
+public class TestPropagator implements Propagator {
+
+	private final Propagator delegate;
+
+	private final TestTracer testTracer;
+
+	public TestPropagator(Propagator delegate, TestTracer testTracer) {
+		this.delegate = delegate;
+		this.testTracer = testTracer;
+	}
+
+	@Override
+	public List<String> fields() {
+		return this.delegate.fields();
+	}
+
+	@Override
+	public <C> void inject(TraceContext context, C carrier, Setter<C> setter) {
+		this.delegate.inject(context, carrier, setter);
+	}
+
+	@Override
+	public <C> Span.Builder extract(C carrier, Getter<C> getter) {
+		return new TestSpanBuilder(this.delegate.extract(carrier, getter), this.testTracer);
+	}
+
+}

tests/common/src/main/java/org/springframework/cloud/sleuth/test/TestSpanBuilder.java

@@ -0,0 +1,88 @@
+/*
+ * Copyright 2013-2021 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.sleuth.test;
+
+import org.springframework.cloud.sleuth.Span;
+import org.springframework.cloud.sleuth.TraceContext;
+
+class TestSpanBuilder implements Span.Builder {
+
+	private final Span.Builder delegate;
+
+	private final TestTracer testTracer;
+
+	TestSpanBuilder(Span.Builder delegate, TestTracer testTracer) {
+		this.delegate = delegate;
+		this.testTracer = testTracer;
+	}
+
+	@Override
+	public Span.Builder setParent(TraceContext context) {
+		delegate.setParent(context);
+		return this;
+	}
+
+	@Override
+	public Span.Builder setNoParent() {
+		delegate.setNoParent();
+		return this;
+	}
+
+	@Override
+	public Span.Builder name(String name) {
+		delegate.name(name);
+		return this;
+	}
+
+	@Override
+	public Span.Builder event(String value) {
+		delegate.event(value);
+		return this;
+	}
+
+	@Override
+	public Span.Builder tag(String key, String value) {
+		delegate.tag(key, value);
+		return this;
+	}
+
+	@Override
+	public Span.Builder error(Throwable throwable) {
+		delegate.error(throwable);
+		return this;
+	}
+
+	@Override
+	public Span.Builder kind(Span.Kind spanKind) {
+		delegate.kind(spanKind);
+		return this;
+	}
+
+	@Override
+	public Span.Builder remoteServiceName(String remoteServiceName) {
+		delegate.remoteServiceName(remoteServiceName);
+		return this;
+	}
+
+	@Override
+	public Span start() {
+		Span span = delegate.start();
+		this.testTracer.createdSpans.add(span);
+		return span;
+	}
+
+}

tests/common/src/main/java/org/springframework/cloud/sleuth/test/TestSpanHandler.java

@@ -17,6 +17,7 @@
 package org.springframework.cloud.sleuth.test;
 
 import java.util.List;
+import java.util.Queue;
 
 import org.springframework.cloud.sleuth.Span;
 import org.springframework.cloud.sleuth.exporter.FinishedSpan;
@@ -35,4 +36,6 @@ public interface TestSpanHandler extends Iterable<FinishedSpan> {
 
 	FinishedSpan get(int index);
 
+	void assertAllSpansWereFinishedOrAbandoned(Queue<Span> createdSpans);
+
 }

tests/common/src/main/java/org/springframework/cloud/sleuth/test/TestTracer.java

@@ -0,0 +1,116 @@
+/*
+ * Copyright 2013-2021 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.sleuth.test;
+
+import java.util.LinkedList;
+import java.util.Map;
+import java.util.Queue;
+
+import org.springframework.cloud.sleuth.BaggageInScope;
+import org.springframework.cloud.sleuth.ScopedSpan;
+import org.springframework.cloud.sleuth.Span;
+import org.springframework.cloud.sleuth.SpanCustomizer;
+import org.springframework.cloud.sleuth.TraceContext;
+import org.springframework.cloud.sleuth.Tracer;
+import org.springframework.lang.Nullable;
+
+public class TestTracer implements Tracer, AutoCloseable {
+
+	private final Tracer delegate;
+
+	final Queue<Span> createdSpans = new LinkedList<>();
+
+	public TestTracer(Tracer delegate) {
+		this.delegate = delegate;
+	}
+
+	@Override
+	public Map<String, String> getAllBaggage() {
+		return delegate.getAllBaggage();
+	}
+
+	@Override
+	public BaggageInScope getBaggage(String name) {
+		return delegate.getBaggage(name);
+	}
+
+	@Override
+	public BaggageInScope getBaggage(TraceContext traceContext, String name) {
+		return delegate.getBaggage(traceContext, name);
+	}
+
+	@Override
+	public BaggageInScope createBaggage(String name) {
+		return delegate.createBaggage(name);
+	}
+
+	@Override
+	public BaggageInScope createBaggage(String name, String value) {
+		return delegate.createBaggage(name, value);
+	}
+
+	@Override
+	public Span nextSpan() {
+		Span span = delegate.nextSpan();
+		this.createdSpans.add(span);
+		return span;
+	}
+
+	@Override
+	public Span nextSpan(Span parent) {
+		Span span = delegate.nextSpan(parent);
+		this.createdSpans.add(span);
+		return span;
+	}
+
+	@Override
+	public SpanInScope withSpan(Span span) {
+		return delegate.withSpan(span);
+	}
+
+	@Override
+	public ScopedSpan startScopedSpan(String name) {
+		return delegate.startScopedSpan(name);
+	}
+
+	@Override
+	public Span.Builder spanBuilder() {
+		return new TestSpanBuilder(delegate.spanBuilder(), this);
+	}
+
+	@Override
+	@Nullable
+	public SpanCustomizer currentSpanCustomizer() {
+		return delegate.currentSpanCustomizer();
+	}
+
+	@Override
+	@Nullable
+	public Span currentSpan() {
+		return delegate.currentSpan();
+	}
+
+	@Override
+	public void close() throws Exception {
+		this.createdSpans.clear();
+	}
+
+	public Queue<Span> createdSpans() {
+		return createdSpans;
+	}
+
+}

tests/common/src/main/java/org/springframework/cloud/sleuth/test/TestTracingBeanPostProcessor.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2013-2021 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.sleuth.test;
+
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.config.BeanPostProcessor;
+import org.springframework.cloud.sleuth.Tracer;
+import org.springframework.cloud.sleuth.propagation.Propagator;
+
+/**
+ * Wraps all tracing related components into test representations. That way additional
+ * assertions can take place.
+ */
+public class TestTracingBeanPostProcessor implements BeanPostProcessor {
+
+	TestTracer testTracer;
+
+	@Override
+	public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
+		if (bean instanceof Tracer && !(bean instanceof TestTracer)) {
+			this.testTracer = new TestTracer((Tracer) bean);
+			return this.testTracer;
+		}
+		else if (bean instanceof Propagator && !(bean instanceof TestPropagator)) {
+			return new TestPropagator((Propagator) bean, this.testTracer);
+		}
+		return bean;
+	}
+
+}