Reduce scoping overhead of WebFlux and adds Benchmarks (#1111)

benchmarks/pom.xml

@@ -34,6 +34,7 @@
 		<maven.compiler.target>1.8</maven.compiler.target>
 		<maven.compiler.source>1.8</maven.compiler.source>
 		<spring-boot.version>2.1.0.M1</spring-boot.version>
+		<brave.version>5.4.3</brave.version>
 	</properties>
 
 	<dependencyManagement>
@@ -65,6 +66,10 @@
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-starter-web</artifactId>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-webflux</artifactId>
+		</dependency>
 		<dependency>
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-configuration-processor</artifactId>
@@ -108,6 +113,21 @@
 			<version>${jmh.version}</version>
 			<scope>provided</scope>
 		</dependency>
+		<dependency>
+			<groupId>com.squareup.okhttp3</groupId>
+			<artifactId>okhttp</artifactId>
+			<version>3.11.0</version>
+			<scope>provided</scope>
+		</dependency>
+		<dependency>
+			<groupId>io.zipkin.brave</groupId>
+			<artifactId>brave-instrumentation-httpclient</artifactId>
+			<version>${brave.version}</version>
+		</dependency>
+		<dependency>
+			<groupId>org.apache.httpcomponents</groupId>
+			<artifactId>httpclient</artifactId>
+		</dependency>
 	</dependencies>
 
 	<build>

benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/app/mvc/SleuthBenchmarkingSpringApp.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.cloud.sleuth.benchmarks.app;
+package org.springframework.cloud.sleuth.benchmarks.app.mvc;
 
 import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutionException;

benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/app/webflux/SleuthBenchmarkingSpringWebFluxApp.java

@@ -0,0 +1,96 @@
+/*
+ * Copyright 2013-2018 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
+ *
+ *      http://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.benchmarks.app.webflux;
+
+import brave.Tracer;
+import brave.sampler.Sampler;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.WebApplicationType;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.builder.SpringApplicationBuilder;
+import org.springframework.boot.web.embedded.netty.NettyReactiveWebServerFactory;
+import org.springframework.boot.web.reactive.context.ReactiveWebServerInitializedEvent;
+import org.springframework.cloud.sleuth.instrument.web.SkipPatternProvider;
+import org.springframework.context.ApplicationListener;
+import org.springframework.context.annotation.Bean;
+import org.springframework.util.SocketUtils;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+import reactor.core.publisher.Mono;
+import zipkin2.Span;
+import zipkin2.reporter.Reporter;
+
+import java.util.regex.Pattern;
+
+
+/**
+ * @author alvin
+ */
+@SpringBootApplication
+@RestController
+public class SleuthBenchmarkingSpringWebFluxApp implements ApplicationListener<ReactiveWebServerInitializedEvent> {
+
+	private static final Log log = LogFactory.getLog(SleuthBenchmarkingSpringWebFluxApp.class);
+
+	public int port;
+
+	@RequestMapping("/foo")
+	public Mono<String> foo() {
+		return Mono.just("foo");
+	}
+
+	@Bean
+	Sampler alwaysSampler() {
+		return Sampler.ALWAYS_SAMPLE;
+	}
+
+	@Bean
+	SkipPatternProvider patternProvider() {
+		return () -> Pattern.compile("");
+	}
+
+
+	@Bean
+	NettyReactiveWebServerFactory nettyReactiveWebServerFactory(@Value("${server.port:0}") int serverPort) {
+		log.info("Starting container at port [" + serverPort + "]");
+		return new NettyReactiveWebServerFactory(serverPort == 0 ? SocketUtils.findAvailableTcpPort() : serverPort);
+	}
+
+	public static void main(String... args) {
+		new SpringApplicationBuilder(SleuthBenchmarkingSpringWebFluxApp.class)
+				.web(WebApplicationType.REACTIVE)
+				.application()
+				.run(args);
+	}
+
+	@Bean
+	public Reporter<Span> reporter() {
+		return Reporter.NOOP;
+	}
+
+
+
+	@Override
+	public void onApplicationEvent(ReactiveWebServerInitializedEvent event) {
+		this.port = event.getWebServer().getPort();
+	}
+}
+

benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/jmh/benchmarks/AnnotationBenchmarks.java

@@ -31,7 +31,7 @@ import org.openjdk.jmh.annotations.TearDown;
 import org.openjdk.jmh.annotations.Threads;
 import org.openjdk.jmh.annotations.Warmup;
 import org.springframework.boot.SpringApplication;
-import org.springframework.cloud.sleuth.benchmarks.app.SleuthBenchmarkingSpringApp;
+import org.springframework.cloud.sleuth.benchmarks.app.mvc.SleuthBenchmarkingSpringApp;
 import org.springframework.context.ConfigurableApplicationContext;
 
 import static org.assertj.core.api.BDDAssertions.then;
@@ -75,4 +75,4 @@ public class AnnotationBenchmarks {
 			throws Exception {
 		then(context.sleuth.newSpan()).isEqualTo("continued");
 	}
-}
+}

benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/jmh/benchmarks/AsyncBenchmarks.java

@@ -30,7 +30,7 @@ import org.openjdk.jmh.annotations.TearDown;
 import org.openjdk.jmh.annotations.Threads;
 import org.openjdk.jmh.annotations.Warmup;
 import org.springframework.boot.SpringApplication;
-import org.springframework.cloud.sleuth.benchmarks.app.SleuthBenchmarkingSpringApp;
+import org.springframework.cloud.sleuth.benchmarks.app.mvc.SleuthBenchmarkingSpringApp;
 import org.springframework.context.ConfigurableApplicationContext;
 
 import static org.assertj.core.api.BDDAssertions.then;
@@ -86,4 +86,4 @@ public class AsyncBenchmarks {
 			throws Exception {
 		then(context.tracedAsyncMethodHavingBean.async().get()).isEqualTo("async");
 	}
-}
+}

benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/jmh/benchmarks/HttpFilterBenchmarks.java

@@ -39,7 +39,7 @@ import org.openjdk.jmh.annotations.TearDown;
 import org.openjdk.jmh.annotations.Threads;
 import org.openjdk.jmh.annotations.Warmup;
 import org.springframework.boot.SpringApplication;
-import org.springframework.cloud.sleuth.benchmarks.app.SleuthBenchmarkingSpringApp;
+import org.springframework.cloud.sleuth.benchmarks.app.mvc.SleuthBenchmarkingSpringApp;
 import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.http.MediaType;
 import org.springframework.mock.web.MockFilterChain;
@@ -167,4 +167,4 @@ public class HttpFilterBenchmarks {
 			return () -> "vanilla";
 		}
 	}
-}
+}

benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/jmh/benchmarks/ProcessLauncherState.java

@@ -30,7 +30,7 @@ import java.util.List;
 import org.openjdk.jmh.util.FileUtils;
 import org.openjdk.jmh.util.Utils;
 
-import org.springframework.cloud.sleuth.benchmarks.app.SleuthBenchmarkingSpringApp;
+import org.springframework.cloud.sleuth.benchmarks.app.mvc.SleuthBenchmarkingSpringApp;
 
 public class ProcessLauncherState {
 
@@ -140,4 +140,4 @@ public class ProcessLauncherState {
 	public File getHome() {
 		return home;
 	}
-}
+}

benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/jmh/benchmarks/RestTemplateBenchmark.java

@@ -34,7 +34,7 @@ import org.openjdk.jmh.annotations.TearDown;
 import org.openjdk.jmh.annotations.Threads;
 import org.openjdk.jmh.annotations.Warmup;
 import org.springframework.boot.SpringApplication;
-import org.springframework.cloud.sleuth.benchmarks.app.SleuthBenchmarkingSpringApp;
+import org.springframework.cloud.sleuth.benchmarks.app.mvc.SleuthBenchmarkingSpringApp;
 import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.test.web.client.MockMvcClientHttpRequestFactory;
 import org.springframework.test.web.servlet.MockMvc;
@@ -96,4 +96,4 @@ public class RestTemplateBenchmark {
 		then(context.tracedTemplate.getForObject("/foo", String.class)).isEqualTo("foo");
 	}
 
-}
+}

benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/jmh/benchmarks/SpringWebFluxBenchmarks.java

@@ -0,0 +1,185 @@
+/*
+ * Copyright 2013-2018 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
+ *
+ *      http://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.benchmarks.jmh.benchmarks;
+
+import brave.Tracing;
+import brave.http.HttpTracing;
+import brave.httpclient.TracingHttpClientBuilder;
+import brave.propagation.CurrentTraceContext;
+import brave.propagation.TraceContext;
+import brave.sampler.Sampler;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.HttpClients;
+import org.apache.http.util.EntityUtils;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Threads;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.runner.Runner;
+import org.openjdk.jmh.runner.RunnerException;
+import org.openjdk.jmh.runner.options.Options;
+import org.openjdk.jmh.runner.options.OptionsBuilder;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.WebApplicationType;
+import org.springframework.boot.builder.SpringApplicationBuilder;
+import org.springframework.cloud.sleuth.benchmarks.app.webflux.SleuthBenchmarkingSpringWebFluxApp;
+import org.springframework.context.ConfigurableApplicationContext;
+import zipkin2.reporter.Reporter;
+
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+
+@Measurement(iterations = 5, time = 1)
+@Warmup(iterations = 10, time = 1)
+@Fork(3)
+@BenchmarkMode(Mode.SampleTime)
+@OutputTimeUnit(TimeUnit.MICROSECONDS)
+@Threads(2)
+@State(Scope.Benchmark)
+public class SpringWebFluxBenchmarks {
+
+	protected ConfigurableApplicationContext applicationContext;
+	protected SleuthBenchmarkingSpringWebFluxApp springWebFluxApp;
+	CloseableHttpClient client;
+	CloseableHttpClient tracedClient;
+	CloseableHttpClient unsampledClient;
+
+	protected static TraceContext defaultTraceContext =
+			TraceContext.newBuilder().traceIdHigh(333L).traceId(444L).spanId(3).sampled(true).build();
+
+	public String getBaseUrl() {
+		return baseUrl;
+	}
+
+	private String baseUrl;
+
+	protected CloseableHttpClient newClient(HttpTracing httpTracing) {
+		return TracingHttpClientBuilder.create(httpTracing)
+				.disableAutomaticRetries()
+				.build();
+	}
+
+	protected CloseableHttpClient newClient() {
+		return HttpClients.custom()
+				.disableAutomaticRetries()
+				.build();
+	}
+
+	protected void get(CloseableHttpClient client) throws Exception {
+		EntityUtils.consume(client.execute(new HttpGet(getBaseUrl())).getEntity());
+	}
+
+	protected void close(CloseableHttpClient client) throws IOException {
+		client.close();
+	}
+
+
+	@Setup
+	public void setup() {
+		ConfigurableApplicationContext context = initContext();
+		this.applicationContext = context;
+		this.springWebFluxApp = this.applicationContext.getBean(
+				SleuthBenchmarkingSpringWebFluxApp.class);
+		baseUrl = "http://127.0.0.1:" + springWebFluxApp.port + "/foo";
+		client = newClient();
+		tracedClient = newClient(HttpTracing.create(
+				Tracing.newBuilder().spanReporter(Reporter.NOOP).build()
+		));
+		unsampledClient = newClient(HttpTracing.create(
+				Tracing.newBuilder().sampler(Sampler.NEVER_SAMPLE).spanReporter(Reporter.NOOP).build()
+		));
+		postSetUp();
+	}
+
+	protected ConfigurableApplicationContext initContext() {
+		SpringApplication application = new SpringApplicationBuilder(SleuthBenchmarkingSpringWebFluxApp.class)
+				.web(WebApplicationType.REACTIVE)
+				.application();
+		customSpringApplication(application);
+		return application
+				.run(runArgs());
+	}
+
+	protected void customSpringApplication(SpringApplication springApplication) {
+
+	}
+
+	protected void postSetUp() {
+	}
+
+
+	protected String[] runArgs() {
+		return new String[]{"--spring.jmx.enabled=false",
+				"--spring.application.name=defaultTraceContext",
+				"--spring.sleuth.enabled=true"};
+	}
+
+
+	@TearDown
+	public void clean() throws Exception {
+		close(client);
+		close(unsampledClient);
+		close(tracedClient);
+		Tracing.current().close();
+		try {
+
+			this.applicationContext.close();
+		} catch (Exception ig) {
+
+		}
+	}
+
+	@Benchmark
+	public void client_get() throws Exception {
+		get(client);
+	}
+
+	@Benchmark
+	public void unsampledClient_get() throws Exception {
+		get(unsampledClient);
+	}
+
+	@Benchmark
+	public void tracedClient_get() throws Exception {
+		get(tracedClient);
+	}
+
+	@Benchmark
+	public void tracedClient_get_resumeTrace() throws Exception {
+		try (CurrentTraceContext.Scope scope = Tracing.current().currentTraceContext().newScope(defaultTraceContext)) {
+			get(tracedClient);
+		}
+	}
+
+	public static void main(String[] args) throws RunnerException {
+		Options opt = new OptionsBuilder()
+				.include(".*" + SpringWebFluxBenchmarks.class.getSimpleName() + ".*")
+				.build();
+
+		new Runner(opt).run();
+	}
+
+
+}

benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/jmh/benchmarks/WithOutReactorSleuthSpringWebFluxBenchmarks.java

@@ -0,0 +1,50 @@
+/*
+ * Copyright 2013-2018 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
+ *
+ *      http://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.benchmarks.jmh.benchmarks;
+
+import org.openjdk.jmh.runner.Runner;
+import org.openjdk.jmh.runner.RunnerException;
+import org.openjdk.jmh.runner.options.Options;
+import org.openjdk.jmh.runner.options.OptionsBuilder;
+
+/**
+ * @author alvin
+ */
+public class WithOutReactorSleuthSpringWebFluxBenchmarks extends SpringWebFluxBenchmarks {
+
+	@Override
+	protected String[] runArgs() {
+		return new String[]{"--spring.jmx.enabled=false",
+				"--spring.application.name=defaultTraceContext",
+				"--spring.sleuth.enabled=true",
+				"--spring.sleuth.reactor.enabled=false"
+
+		};
+	}
+
+	@Override
+	protected void postSetUp() {
+		super.postSetUp();
+	}
+
+	public static void main(String[] args) throws RunnerException {
+		Options opt = new OptionsBuilder()
+				.include(".*" + WithOutReactorSleuthSpringWebFluxBenchmarks.class.getSimpleName() + ".*")
+				.build();
+
+		new Runner(opt).run();
+	}
+}

benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/jmh/benchmarks/WithOutSleuthSpringWebFluxBenchmarks.java

@@ -0,0 +1,47 @@
+/*
+ * Copyright 2013-2018 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
+ *
+ *      http://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.benchmarks.jmh.benchmarks;
+
+import org.openjdk.jmh.runner.Runner;
+import org.openjdk.jmh.runner.RunnerException;
+import org.openjdk.jmh.runner.options.Options;
+import org.openjdk.jmh.runner.options.OptionsBuilder;
+
+/**
+ * @author alvin
+ */
+public class WithOutSleuthSpringWebFluxBenchmarks extends SpringWebFluxBenchmarks {
+
+	@Override
+	protected String[] runArgs() {
+		return new String[]{"--spring.jmx.enabled=false",
+				"--spring.application.name=defaultTraceContext",
+				"--spring.sleuth.enabled=false"};
+	}
+
+	@Override
+	protected void postSetUp() {
+		super.postSetUp();
+	}
+
+	public static void main(String[] args) throws RunnerException {
+		Options opt = new OptionsBuilder()
+				.include(".*" + WithOutSleuthSpringWebFluxBenchmarks.class.getSimpleName() + ".*")
+				.build();
+
+		new Runner(opt).run();
+	}
+}

benchmarks/src/main/resources/application.yml

@@ -1,3 +1,3 @@
 logging.level:
   org.springframework: ERROR
-  org.springframework.cloud.sleuth.benchmarks: INFO
+  org.springframework.cloud.sleuth.benchmarks: INFO

spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriber.java

@@ -16,11 +16,11 @@
 
 package org.springframework.cloud.sleuth.instrument.reactor;
 
-import java.util.concurrent.atomic.AtomicBoolean;
-
 import brave.Span;
 import brave.Tracer;
 import brave.Tracing;
+import brave.propagation.CurrentTraceContext;
+import brave.propagation.TraceContext;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.reactivestreams.Subscriber;
@@ -38,23 +38,25 @@ final class ScopePassingSpanSubscriber<T> implements SpanSubscription<T> {
 
 	private static final Log log = LogFactory.getLog(ScopePassingSpanSubscriber.class);
 
-	private final Span span;
-
 	private final Subscriber<? super T> subscriber;
 
 	private final Context context;
 
+	private final CurrentTraceContext currentTraceContext;
+	private final TraceContext traceContext;
 	private final Tracer tracer;
 
 	private Subscription s;
 
 	ScopePassingSpanSubscriber(Subscriber<? super T> subscriber, Context ctx,
-			Tracing tracing) {
+	                           Tracing tracing) {
 		this.subscriber = subscriber;
 		this.tracer = tracing.tracer();
-		Span root = ctx != null ? ctx.getOrDefault(Span.class, this.tracer.currentSpan())
+		this.currentTraceContext = tracing.currentTraceContext();
+		Span root = ctx != null ? ctx.hasKey(Span.class) ? ctx.get(Span.class)
+				: this.tracer.currentSpan()
 				: null;
-		this.span = root;
+		this.traceContext = root == null ? null : root.context();
 		this.context = ctx != null && root != null ? ctx.put(Span.class, root)
 				: ctx != null ? ctx : Context.empty();
 		if (log.isTraceEnabled()) {
@@ -65,42 +67,43 @@ final class ScopePassingSpanSubscriber<T> implements SpanSubscription<T> {
 	@Override
 	public void onSubscribe(Subscription subscription) {
 		this.s = subscription;
-		try (Tracer.SpanInScope inScope = this.tracer.withSpanInScope(this.span)) {
+		try (CurrentTraceContext.Scope scope = this.currentTraceContext.maybeScope(this.traceContext)) {
 			this.subscriber.onSubscribe(this);
 		}
 	}
 
 	@Override
 	public void request(long n) {
-		try (Tracer.SpanInScope inScope = this.tracer.withSpanInScope(this.span)) {
+		try (CurrentTraceContext.Scope scope = this.currentTraceContext.maybeScope(this.traceContext)) {
 			this.s.request(n);
 		}
 	}
 
 	@Override
 	public void cancel() {
-		try (Tracer.SpanInScope inScope = this.tracer.withSpanInScope(this.span)) {
+		try (CurrentTraceContext.Scope scope = this.currentTraceContext.maybeScope(this.traceContext)) {
 			this.s.cancel();
 		}
+
 	}
 
 	@Override
 	public void onNext(T o) {
-		try (Tracer.SpanInScope inScope = this.tracer.withSpanInScope(this.span)) {
+		try (CurrentTraceContext.Scope scope = this.currentTraceContext.maybeScope(this.traceContext)) {
 			this.subscriber.onNext(o);
 		}
 	}
 
 	@Override
 	public void onError(Throwable throwable) {
-		try (Tracer.SpanInScope inScope = this.tracer.withSpanInScope(this.span)) {
+		try (CurrentTraceContext.Scope scope = this.currentTraceContext.maybeScope(this.traceContext)) {
 			this.subscriber.onError(throwable);
 		}
 	}
 
 	@Override
 	public void onComplete() {
-		try (Tracer.SpanInScope inScope = this.tracer.withSpanInScope(this.span)) {
+		try (CurrentTraceContext.Scope scope = this.currentTraceContext.maybeScope(this.traceContext)) {
 			this.subscriber.onComplete();
 		}
 	}
@@ -110,8 +113,4 @@ final class ScopePassingSpanSubscriber<T> implements SpanSubscription<T> {
 		return this.context;
 	}
 
-	private void clearSpan() {
-		this.tracer.withSpanInScope(null);
-	}
-
 }

spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/reactor/SpanSubscriptionProvider.java

@@ -63,7 +63,7 @@ class SpanSubscriptionProvider<T> implements Supplier<SpanSubscription<T>> {
 	}
 
 	SpanSubscription<T> newCoreSubscriber(Tracing tracing) {
-		return new SpanSubscriber<>(this.subscriber, this.context, tracing, this.name);
+		return new ScopePassingSpanSubscriber<>(this.subscriber, this.context, tracing);
 	}
 
 	private Tracing tracing() {