Adds support for SC CircuitBreaker Reactive (#1914)

* Adds support for SC CircuitBreaker Reactive

* Changes following the review

* Lazilly initializes the TraceFunction

* Updated version

fixes gh-1910

docs/src/main/asciidoc/integrations.adoc

@@ -565,5 +565,5 @@ IMPORTANT: The suggested approach to reactive programming and Sleuth is to use t
 
 This feature is available for all tracer implementations.
 
-If you have Spring Cloud CircuitBreaker on the classpath, we will wrap the passed command `Supplier` and the fallback `Function` in its trace representations.
+If you have Spring Cloud CircuitBreaker on the classpath, we will wrap the passed command `Supplier` and the fallback `Function` in its trace representations. We will also instrument the reactive implementation of the CircuitBreaker.
 In order to disable this instrumentation set `spring.sleuth.circuitbreaker.enabled` to `false`.

spring-cloud-sleuth-autoconfigure/src/main/java/org/springframework/cloud/sleuth/autoconfig/instrument/circuitbreaker/TraceCircuitBreakerAutoConfiguration.java

@@ -21,10 +21,11 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
-import org.springframework.cloud.client.circuitbreaker.CircuitBreaker;
+import org.springframework.cloud.sleuth.CurrentTraceContext;
 import org.springframework.cloud.sleuth.Tracer;
 import org.springframework.cloud.sleuth.autoconfig.brave.BraveAutoConfiguration;
 import org.springframework.cloud.sleuth.instrument.circuitbreaker.TraceCircuitBreakerFactoryAspect;
+import org.springframework.cloud.sleuth.instrument.circuitbreaker.TraceReactiveCircuitBreakerFactoryAspect;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
@@ -36,7 +37,6 @@ import org.springframework.context.annotation.Configuration;
  * @since 2.2.1
  */
 @Configuration(proxyBeanMethods = false)
-@ConditionalOnClass(CircuitBreaker.class)
 @ConditionalOnBean(Tracer.class)
 @ConditionalOnProperty(value = "spring.sleuth.circuitbreaker.enabled", matchIfMissing = true)
 @EnableConfigurationProperties(SleuthCircuitBreakerProperties.class)
@@ -44,8 +44,17 @@ import org.springframework.context.annotation.Configuration;
 public class TraceCircuitBreakerAutoConfiguration {
 
 	@Bean
+	@ConditionalOnClass(name = "org.springframework.cloud.client.circuitbreaker.CircuitBreaker")
 	TraceCircuitBreakerFactoryAspect traceCircuitBreakerFactoryAspect(Tracer tracer) {
 		return new TraceCircuitBreakerFactoryAspect(tracer);
 	}
 
+	@Bean
+	@ConditionalOnClass(name = { "reactor.core.publisher.Mono",
+			"org.springframework.cloud.client.circuitbreaker.ReactiveCircuitBreaker" })
+	TraceReactiveCircuitBreakerFactoryAspect traceReactiveCircuitBreakerFactoryAspect(Tracer tracer,
+			CurrentTraceContext currentTraceContext) {
+		return new TraceReactiveCircuitBreakerFactoryAspect(tracer, currentTraceContext);
+	}
+
 }

spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/circuitbreaker/TraceCircuitBreakerFactoryAspect.java

@@ -19,7 +19,6 @@ package org.springframework.cloud.sleuth.instrument.circuitbreaker;
 import org.aspectj.lang.ProceedingJoinPoint;
 import org.aspectj.lang.annotation.Around;
 import org.aspectj.lang.annotation.Aspect;
-import org.aspectj.lang.annotation.Pointcut;
 
 import org.springframework.cloud.client.circuitbreaker.CircuitBreaker;
 import org.springframework.cloud.sleuth.Tracer;
@@ -39,11 +38,7 @@ public class TraceCircuitBreakerFactoryAspect {
 		this.tracer = tracer;
 	}
 
-	@Pointcut("execution(public * org.springframework.cloud.client.circuitbreaker.CircuitBreakerFactory.create(..))")
-	private void anyCircuitBreakerFactoryCreate() {
-	} // NOSONAR
-
-	@Around("anyCircuitBreakerFactoryCreate()")
+	@Around("execution(public * org.springframework.cloud.client.circuitbreaker.CircuitBreakerFactory.create(..))")
 	public Object wrapFactory(ProceedingJoinPoint pjp) throws Throwable {
 		CircuitBreaker circuitBreaker = (CircuitBreaker) pjp.proceed();
 		return new TraceCircuitBreaker(circuitBreaker, this.tracer);

spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/circuitbreaker/TraceFunction.java

@@ -44,6 +44,7 @@ class TraceFunction<T> implements Function<Throwable, T> {
 
 	@Override
 	public T apply(Throwable throwable) {
+		// TODO: This name needs to be better
 		String name = this.delegate.getClass().getSimpleName();
 		Span span = this.span.get().name(name);
 		Throwable tr = null;

spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/circuitbreaker/TraceReactiveCircuitBreaker.java

@@ -0,0 +1,138 @@
+/*
+ * Copyright 2018-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.circuitbreaker;
+
+import java.util.function.Function;
+import java.util.function.Supplier;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.util.context.Context;
+
+import org.springframework.cloud.client.circuitbreaker.ReactiveCircuitBreaker;
+import org.springframework.cloud.sleuth.CurrentTraceContext;
+import org.springframework.cloud.sleuth.Span;
+import org.springframework.cloud.sleuth.TraceContext;
+import org.springframework.cloud.sleuth.Tracer;
+
+class TraceReactiveCircuitBreaker implements ReactiveCircuitBreaker {
+
+	private static final Log log = LogFactory.getLog(TraceReactiveCircuitBreaker.class);
+
+	private final ReactiveCircuitBreaker delegate;
+
+	private final Tracer tracer;
+
+	private final CurrentTraceContext currentTraceContext;
+
+	TraceReactiveCircuitBreaker(ReactiveCircuitBreaker delegate, Tracer tracer,
+			CurrentTraceContext currentTraceContext) {
+		this.delegate = delegate;
+		this.tracer = tracer;
+		this.currentTraceContext = currentTraceContext;
+	}
+
+	@Override
+	public <T> Mono<T> run(Mono<T> toRun) {
+		return runAndTraceMono(() -> this.delegate.run(toRun));
+	}
+
+	@Override
+	public <T> Mono<T> run(Mono<T> toRun, Function<Throwable, Mono<T>> fallback) {
+		return runAndTraceMono(
+				() -> this.delegate.run(toRun, fallback != null ? new TraceFunction<>(this.tracer, fallback) : null));
+	}
+
+	@Override
+	public <T> Flux<T> run(Flux<T> toRun) {
+		return runAndTraceFlux(() -> this.delegate.run(toRun));
+	}
+
+	@Override
+	public <T> Flux<T> run(Flux<T> toRun, Function<Throwable, Flux<T>> fallback) {
+		return runAndTraceFlux(
+				() -> this.delegate.run(toRun, fallback != null ? new TraceFunction<>(this.tracer, fallback) : null));
+	}
+
+	private <T> Mono<T> runAndTraceMono(Supplier<Mono<T>> mono) {
+		return Mono.deferContextual(contextView -> {
+			Span span = contextView.get(Span.class);
+			Tracer.SpanInScope scope = contextView.get(Tracer.SpanInScope.class);
+			return mono.get().doOnError(span::error).doFinally(signalType -> {
+				span.end();
+				scope.close();
+			});
+		}).contextWrite(this::enhanceContext);
+	}
+
+	private <T> Flux<T> runAndTraceFlux(Supplier<Flux<T>> flux) {
+		return Flux.deferContextual(contextView -> {
+			Span span = contextView.get(Span.class);
+			Tracer.SpanInScope scope = contextView.get(Tracer.SpanInScope.class);
+			return flux.get().doOnError(span::error).doFinally(signalType -> {
+				span.end();
+				scope.close();
+			});
+		}).contextWrite(this::enhanceContext);
+	}
+
+	private Span spanFromContext(reactor.util.context.Context context) {
+		TraceContext traceContext = context.getOrDefault(TraceContext.class, null);
+		Span span = null;
+		if (traceContext == null) {
+			span = context.getOrDefault(Span.class, null);
+		}
+		if (traceContext == null && span == null) {
+			span = this.tracer.nextSpan();
+			if (log.isDebugEnabled()) {
+				log.debug("There was no previous span in reactor context, created a new one [" + span + "]");
+			}
+		}
+		else if (traceContext != null) {
+			// there was a previous span - we create a child one
+			try (CurrentTraceContext.Scope scope = this.currentTraceContext.maybeScope(traceContext)) {
+				if (log.isDebugEnabled()) {
+					log.debug("Found a trace context in reactor context [" + traceContext + "]");
+				}
+				span = this.tracer.nextSpan();
+				if (log.isDebugEnabled()) {
+					log.debug("Created a child span [" + span + "]");
+				}
+			}
+		}
+		else {
+			if (log.isDebugEnabled()) {
+				log.debug("Found a span in reactor context [" + span + "]");
+			}
+			span = this.tracer.nextSpan(span);
+			if (log.isDebugEnabled()) {
+				log.debug("Created a child span [" + span + "]");
+			}
+		}
+		// TODO: Better name?
+		return span.name("function");
+	}
+
+	private Context enhanceContext(Context context) {
+		Span span = spanFromContext(context);
+		return context.put(Span.class, span).put(TraceContext.class, span.context()).put(Tracer.SpanInScope.class,
+				this.tracer.withSpan(span));
+	}
+
+}

spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/circuitbreaker/TraceReactiveCircuitBreakerFactoryAspect.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2018-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.circuitbreaker;
+
+import org.aspectj.lang.ProceedingJoinPoint;
+import org.aspectj.lang.annotation.Around;
+import org.aspectj.lang.annotation.Aspect;
+
+import org.springframework.cloud.client.circuitbreaker.ReactiveCircuitBreaker;
+import org.springframework.cloud.sleuth.CurrentTraceContext;
+import org.springframework.cloud.sleuth.Tracer;
+
+@Aspect
+public class TraceReactiveCircuitBreakerFactoryAspect {
+
+	private final Tracer tracer;
+
+	private final CurrentTraceContext currentTraceContext;
+
+	public TraceReactiveCircuitBreakerFactoryAspect(Tracer tracer, CurrentTraceContext currentTraceContext) {
+		this.tracer = tracer;
+		this.currentTraceContext = currentTraceContext;
+	}
+
+	@Around("execution(public * org.springframework.cloud.client.circuitbreaker.ReactiveCircuitBreakerFactory.create(..))")
+	public Object wrapFactory(ProceedingJoinPoint pjp) throws Throwable {
+		ReactiveCircuitBreaker circuitBreaker = (ReactiveCircuitBreaker) pjp.proceed();
+		return new TraceReactiveCircuitBreaker(circuitBreaker, this.tracer, this.currentTraceContext);
+	}
+
+}

spring-cloud-sleuth-instrumentation/src/main/java/org/springframework/cloud/sleuth/instrument/circuitbreaker/TraceSupplier.java

@@ -44,6 +44,7 @@ class TraceSupplier<T> implements Supplier<T> {
 
 	@Override
 	public T get() {
+		// TODO: This name needs to be better
 		String name = this.delegate.getClass().getSimpleName();
 		Span span = this.span.get().name(name);
 		Throwable tr = null;

tests/brave/pom.xml

@@ -39,6 +39,7 @@
 		<module>spring-cloud-sleuth-instrumentation-async-tests</module>
 		<module>spring-cloud-sleuth-instrumentation-baggage-tests</module>
 		<module>spring-cloud-sleuth-instrumentation-circuitbreaker-tests</module>
+		<module>spring-cloud-sleuth-instrumentation-circuitbreaker-reactive-tests</module>
 		<module>spring-cloud-sleuth-instrumentation-feign-tests</module>
 		<module>spring-cloud-sleuth-instrumentation-gateway-tests</module>
 		<module>spring-cloud-sleuth-instrumentation-grpc-tests</module>

tests/brave/spring-cloud-sleuth-instrumentation-circuitbreaker-reactive-tests/pom.xml

@@ -0,0 +1,89 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ 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.
+  ~
+  ~
+  -->
+
+<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+		 xmlns="http://maven.apache.org/POM/4.0.0"
+		 xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+
+	<artifactId>spring-cloud-sleuth-instrumentation-circuitbreaker-reactive-tests</artifactId>
+	<packaging>jar</packaging>
+	<name>Spring Cloud Sleuth Brave Circuitbreaker Reactive Instrumentation Tests</name>
+	<description>Spring Cloud Sleuth Brave Circuitbreaker Reactive Instrumentation Tests</description>
+
+	<parent>
+		<groupId>org.springframework.cloud</groupId>
+		<artifactId>spring-cloud-sleuth-tests-brave</artifactId>
+		<version>3.1.0-SNAPSHOT</version>
+		<relativePath>..</relativePath>
+	</parent>
+
+	<properties>
+		<sonar.skip>true</sonar.skip>
+	</properties>
+
+	<build>
+		<plugins>
+			<plugin>
+				<!--skip deploy -->
+				<artifactId>maven-deploy-plugin</artifactId>
+				<configuration>
+					<skip>true</skip>
+				</configuration>
+			</plugin>
+		</plugins>
+	</build>
+
+	<dependencies>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-sleuth-tests-common</artifactId>
+			<version>${project.version}</version>
+		</dependency>
+		<dependency>
+			<groupId>io.projectreactor</groupId>
+			<artifactId>reactor-core</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-aop</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-starter-circuitbreaker-reactor-resilience4j</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-starter-sleuth</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-test</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>io.zipkin.brave</groupId>
+			<artifactId>brave-tests</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.awaitility</groupId>
+			<artifactId>awaitility</artifactId>
+		</dependency>
+	</dependencies>
+
+</project>

tests/brave/spring-cloud-sleuth-instrumentation-circuitbreaker-reactive-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/circuitbreaker/ReactiveCircuitBreakerIntegrationTests.java

@@ -0,0 +1,60 @@
+/*
+ * 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.instrument.circuitbreaker;
+
+import brave.sampler.Sampler;
+import org.assertj.core.api.BDDAssertions;
+
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.cloud.sleuth.brave.BraveTestSpanHandler;
+import org.springframework.cloud.sleuth.exporter.FinishedSpan;
+import org.springframework.cloud.sleuth.test.TestSpanHandler;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.test.context.ContextConfiguration;
+
+@SpringBootTest
+@ContextConfiguration(classes = ReactiveCircuitBreakerIntegrationTests.Config.class)
+public class ReactiveCircuitBreakerIntegrationTests
+		extends org.springframework.cloud.sleuth.instrument.circuitbreaker.ReactiveCircuitBreakerIntegrationTests {
+
+	@Override
+	public void assertException(FinishedSpan finishedSpan) {
+		BDDAssertions.then(finishedSpan.getTags().get("error")).contains("boom");
+	}
+
+	@Configuration(proxyBeanMethods = false)
+	static class Config {
+
+		@Bean
+		TestSpanHandler testSpanHandlerSupplier(brave.test.TestSpanHandler testSpanHandler) {
+			return new BraveTestSpanHandler(testSpanHandler);
+		}
+
+		@Bean
+		Sampler alwaysSampler() {
+			return Sampler.ALWAYS_SAMPLE;
+		}
+
+		@Bean
+		brave.test.TestSpanHandler braveTestSpanHandler() {
+			return new brave.test.TestSpanHandler();
+		}
+
+	}
+
+}

tests/brave/spring-cloud-sleuth-instrumentation-circuitbreaker-reactive-tests/src/test/java/org/springframework/cloud/sleuth/brave/instrument/circuitbreaker/ReactiveCircuitBreakerTests.java

@@ -0,0 +1,43 @@
+/*
+ * 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.instrument.circuitbreaker;
+
+import org.assertj.core.api.BDDAssertions;
+
+import org.springframework.cloud.sleuth.brave.BraveTestTracing;
+import org.springframework.cloud.sleuth.exporter.FinishedSpan;
+import org.springframework.cloud.sleuth.test.TestTracingAware;
+
+public class ReactiveCircuitBreakerTests
+		extends org.springframework.cloud.sleuth.instrument.circuitbreaker.ReactiveCircuitBreakerTests {
+
+	BraveTestTracing testTracing;
+
+	@Override
+	public TestTracingAware tracerTest() {
+		if (this.testTracing == null) {
+			this.testTracing = new BraveTestTracing();
+		}
+		return this.testTracing;
+	}
+
+	@Override
+	public void additionalAssertions(FinishedSpan finishedSpan) {
+		BDDAssertions.then(finishedSpan.getTags().get("error")).contains("boom2");
+	}
+
+}

tests/brave/spring-cloud-sleuth-instrumentation-circuitbreaker-reactive-tests/src/test/resources/application.yml

@@ -0,0 +1,5 @@
+logging.level.org.springframework.cloud: DEBUG
+logging.level.com.netflix.discovery.InstanceInfoReplicator: ERROR
+logging.level.org.springframework.cloud.sleuth.brave.instrument.web.client.feign: TRACE
+
+spring.autoconfigure.exclude: org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration, org.springframework.boot.autoconfigure.orm.jpa.HibernateJpaAutoConfiguration, org.springframework.boot.autoconfigure.security.servlet.SecurityAutoConfiguration, org.springframework.boot.autoconfigure.data.web.SpringDataWebAutoConfiguration, org.springframework.cloud.gateway.config.GatewayAutoConfiguration, org.springframework.cloud.gateway.config.GatewayClassPathWarningAutoConfiguration, org.springframework.cloud.gateway.config.GatewayMetricsAutoConfiguration

tests/common/pom.xml

@@ -89,6 +89,11 @@
 			<artifactId>spring-cloud-starter-circuitbreaker-resilience4j</artifactId>
 			<optional>true</optional>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-starter-circuitbreaker-reactor-resilience4j</artifactId>
+			<optional>true</optional>
+		</dependency>
 		<dependency>
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-starter-quartz</artifactId>

tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/circuitbreaker/ReactiveCircuitBreakerIntegrationTests.java

@@ -0,0 +1,146 @@
+/*
+ * 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.circuitbreaker;
+
+import org.assertj.core.api.BDDAssertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import reactor.core.publisher.Mono;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.cloud.circuitbreaker.resilience4j.ReactiveResilience4JCircuitBreakerFactory;
+import org.springframework.cloud.client.circuitbreaker.ReactiveCircuitBreakerFactory;
+import org.springframework.cloud.sleuth.ScopedSpan;
+import org.springframework.cloud.sleuth.Span;
+import org.springframework.cloud.sleuth.Tracer;
+import org.springframework.cloud.sleuth.exporter.FinishedSpan;
+import org.springframework.cloud.sleuth.test.TestSpanHandler;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.test.context.ContextConfiguration;
+
+@ContextConfiguration(classes = ReactiveCircuitBreakerIntegrationTests.TestConfig.class)
+public abstract class ReactiveCircuitBreakerIntegrationTests {
+
+	@Autowired
+	TestSpanHandler spans;
+
+	@Autowired
+	Tracer tracer;
+
+	@Autowired
+	ReactiveCircuitBreakerFactory factory;
+
+	@Autowired
+	CircuitService circuitService;
+
+	@BeforeEach
+	public void setup() {
+		this.spans.clear();
+	}
+
+	@Test
+	public void should_pass_tracing_information_when_using_circuit_breaker() {
+		// given
+		Tracer tracer = this.tracer;
+		ScopedSpan scopedSpan = null;
+		try {
+			scopedSpan = tracer.startScopedSpan("start");
+			// when
+			Span span = this.factory.create("name").run(Mono.defer(() -> Mono.just(tracer.currentSpan()))).block();
+
+			BDDAssertions.then(span).isNotNull();
+			BDDAssertions.then(scopedSpan.context().traceId()).isEqualTo(span.context().traceId());
+		}
+		finally {
+			scopedSpan.end();
+		}
+	}
+
+	@Test
+	public void should_pass_tracing_information_when_using_circuit_breaker_with_fallback() {
+		// when
+		BDDAssertions.then(this.circuitService.call().block()).isEqualTo("fallback");
+
+		BDDAssertions.then(this.spans).hasSize(2);
+		String traceId = this.circuitService.firstSpan.context().traceId();
+		BDDAssertions.then(this.circuitService.secondSpan.context().traceId()).isEqualTo(traceId);
+
+		FinishedSpan finishedSpan = this.spans.get(0);
+		BDDAssertions.then(finishedSpan.getName()).contains("CircuitBreakerIntegrationTests");
+
+		finishedSpan = this.spans.get(1);
+		BDDAssertions.then(finishedSpan.getName()).contains("function");
+	}
+
+	public void assertException(FinishedSpan finishedSpan) {
+		throw new UnsupportedOperationException("Implement this assertion");
+	}
+
+	@Configuration(proxyBeanMethods = false)
+	@EnableAutoConfiguration
+	public static class TestConfig {
+
+		@Bean
+		ReactiveResilience4JCircuitBreakerFactory reactiveResilience4JCircuitBreakerFactory() {
+			return new ReactiveResilience4JCircuitBreakerFactory();
+		}
+
+		@Bean
+		CircuitService circuitService(ReactiveCircuitBreakerFactory reactiveCircuitBreakerFactory, Tracer tracer) {
+			return new CircuitService(reactiveCircuitBreakerFactory, tracer);
+		}
+
+	}
+
+	static class CircuitService {
+
+		private static final Logger log = LoggerFactory.getLogger(CircuitService.class);
+
+		private final ReactiveCircuitBreakerFactory factory;
+
+		private final Tracer tracer;
+
+		Span firstSpan;
+
+		Span secondSpan;
+
+		CircuitService(ReactiveCircuitBreakerFactory factory, Tracer tracer) {
+			this.factory = factory;
+			this.tracer = tracer;
+		}
+
+		Mono<String> call() {
+			return this.factory.create("circuit").run(Mono.defer(() -> {
+				this.firstSpan = this.tracer.currentSpan();
+				log.info("<ACCEPTANCE_TEST> <TRACE:{}> Hello from consumer",
+						this.tracer.currentSpan().context().traceId());
+				return Mono.error(new IllegalStateException("boom"));
+			}), throwable -> {
+				this.secondSpan = this.tracer.currentSpan();
+				log.info("<ACCEPTANCE_TEST> <TRACE:{}> Hello from producer",
+						this.tracer.currentSpan().context().traceId());
+				return Mono.just("fallback");
+			});
+		}
+
+	}
+
+}

tests/common/src/main/java/org/springframework/cloud/sleuth/instrument/circuitbreaker/ReactiveCircuitBreakerTests.java

@@ -0,0 +1,94 @@
+/*
+ * 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.circuitbreaker;
+
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.assertj.core.api.BDDAssertions;
+import org.junit.jupiter.api.Test;
+import reactor.core.publisher.Mono;
+
+import org.springframework.cloud.circuitbreaker.resilience4j.ReactiveResilience4JCircuitBreakerFactory;
+import org.springframework.cloud.sleuth.ScopedSpan;
+import org.springframework.cloud.sleuth.Span;
+import org.springframework.cloud.sleuth.Tracer;
+import org.springframework.cloud.sleuth.exporter.FinishedSpan;
+import org.springframework.cloud.sleuth.test.TestTracingAwareSupplier;
+
+public abstract class ReactiveCircuitBreakerTests implements TestTracingAwareSupplier {
+
+	@Test
+	public void should_pass_tracing_information_when_using_circuit_breaker() {
+		// given
+		Tracer tracer = tracerTest().tracing().tracer();
+		ScopedSpan scopedSpan = null;
+		try {
+			scopedSpan = tracer.startScopedSpan("start");
+			// when
+			Span span = new TraceReactiveCircuitBreaker(new ReactiveResilience4JCircuitBreakerFactory().create("name"),
+					tracer, tracerTest().tracing().currentTraceContext())
+							.run(Mono.defer(() -> Mono.just(tracer.currentSpan()))).block();
+
+			BDDAssertions.then(span).isNotNull();
+			BDDAssertions.then(scopedSpan.context().traceId()).isEqualTo(span.context().traceId());
+		}
+		finally {
+			scopedSpan.end();
+		}
+	}
+
+	@Test
+	public void should_pass_tracing_information_when_using_circuit_breaker_with_fallback() {
+		// given
+		Tracer tracer = tracerTest().tracing().tracer();
+		AtomicReference<Span> first = new AtomicReference<>();
+		AtomicReference<Span> second = new AtomicReference<>();
+		ScopedSpan scopedSpan = null;
+		try {
+			scopedSpan = tracer.startScopedSpan("start");
+			// when
+			BDDAssertions.thenThrownBy(() -> new TraceReactiveCircuitBreaker(
+					new ReactiveResilience4JCircuitBreakerFactory().create("name"), tracer,
+					tracerTest().tracing().currentTraceContext()).run(Mono.defer(() -> {
+						first.set(tracer.currentSpan());
+						throw new IllegalStateException("boom");
+					}), throwable -> {
+						second.set(tracer.currentSpan());
+						throw new IllegalStateException("boom2");
+					}).block()).isInstanceOf(IllegalStateException.class).hasMessageContaining("boom2");
+
+			BDDAssertions.then(tracerTest().handler().reportedSpans()).hasSize(2);
+			BDDAssertions.then(first.get()).isNotNull();
+			BDDAssertions.then(second.get()).isNotNull();
+			BDDAssertions.then(scopedSpan.context().traceId()).isEqualTo(first.get().context().traceId());
+			BDDAssertions.then(scopedSpan.context().traceId()).isEqualTo(second.get().context().traceId());
+			BDDAssertions.then(first.get().context().spanId()).isNotEqualTo(second.get().context().spanId());
+
+			FinishedSpan finishedSpan = tracerTest().handler().reportedSpans().get(1);
+			BDDAssertions.then(finishedSpan.getName()).contains("function");
+			additionalAssertions(finishedSpan);
+		}
+		finally {
+			scopedSpan.end();
+		}
+	}
+
+	public void additionalAssertions(FinishedSpan finishedSpan) {
+		throw new UnsupportedOperationException("Assert errors");
+	}
+
+}