Merge branch '2.2.x'

spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/log/Slf4JSpanLoggerTest.java

@@ -175,10 +175,9 @@ public class Slf4JSpanLoggerTest {
 	public void should_only_include_whitelist() {
 		assertThat(this.slf4jScopeDecorator).extracting("delegate.fields")
 				.asInstanceOf(InstanceOfAssertFactories.array(CorrelationField[].class))
-				.extracting(CorrelationField::name)
-				// my-baggage-two is baggage not in the whitelist
-				.containsExactly("traceId", "parentId", "spanId", "spanExportable",
-						"my-baggage", "my-local", "my-propagation");
+				.extracting(CorrelationField::name).containsExactly("traceId", "parentId",
+						"spanId", "spanExportable", "my-baggage", "my-local",
+						"my-propagation"); // my-baggage-two is not in the whitelist
 	}
 
 	@Test

src/checkstyle/checkstyle-suppressions.xml

@@ -4,7 +4,7 @@
 	"https://www.puppycrawl.com/dtds/suppressions_1_1.dtd">
 <suppressions>
 	<suppress files=".*/test/.*" checks="JavadocVariable"/>
-        <suppress files=".*ReactorSleuth\.java" checks="InnerAssignment"/>
+	<suppress files=".*ReactorSleuth\.java" checks="InnerAssignment"/>
 	<suppress files=".*FinishedSpanHandlerTests.*" checks="LineLengthCheck"/>
 	<suppress files=".*GrpcTracingIntegrationTests.*" checks="LineLengthCheck"/>
 	<suppress files=".*IgnoreAutoConfiguredSkipPatternsIntegrationTests.*" checks="LineLengthCheck"/>

tests/spring-cloud-sleuth-instrumentation-reactor-tests/src/test/java/org/springframework/cloud/sleuth/instrument/reactor/ScopePassingSpanSubscriberTests.java

@@ -17,20 +17,33 @@
 package org.springframework.cloud.sleuth.instrument.reactor;
 
 import java.util.Objects;
+import java.util.function.Function;
 
+import brave.propagation.CurrentTraceContext;
 import brave.propagation.CurrentTraceContext.Scope;
 import brave.propagation.StrictCurrentTraceContext;
 import brave.propagation.TraceContext;
 import org.assertj.core.presentation.StandardRepresentation;
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.Test;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
 import reactor.core.CoreSubscriber;
 import reactor.core.publisher.BaseSubscriber;
+import reactor.core.publisher.Hooks;
+import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
 import reactor.util.context.Context;
 
 import org.springframework.context.annotation.AnnotationConfigApplicationContext;
 
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.BDDAssertions.then;
+import static org.springframework.cloud.sleuth.instrument.reactor.ReactorSleuth.scopePassingSpanOperator;
+import static org.springframework.cloud.sleuth.instrument.reactor.TraceReactorAutoConfiguration.SLEUTH_REACTOR_EXECUTOR_SERVICE_KEY;
+import static org.springframework.cloud.sleuth.instrument.reactor.TraceReactorAutoConfiguration.TraceReactorConfiguration.SLEUTH_TRACE_REACTOR_KEY;
 
 /**
  * @author Marcin Grzejszczak
@@ -54,9 +67,64 @@ public class ScopePassingSpanSubscriberTests {
 	TraceContext context2 = TraceContext.newBuilder().traceId(1).spanId(2).sampled(true)
 			.build();
 
+	Subscriber<Object> assertNotScopePassingSpanSubscriber = new CoreSubscriber<Object>() {
+		@Override
+		public void onSubscribe(Subscription s) {
+			s.request(Long.MAX_VALUE);
+			assertThat(s).isNotInstanceOf(ScopePassingSpanSubscriber.class);
+		}
+
+		@Override
+		public void onNext(Object o) {
+
+		}
+
+		@Override
+		public void onError(Throwable t) {
+
+		}
+
+		@Override
+		public void onComplete() {
+
+		}
+	};
+
+	Subscriber<Object> assertScopePassingSpanSubscriber = new CoreSubscriber<Object>() {
+		@Override
+		public void onSubscribe(Subscription s) {
+			s.request(Long.MAX_VALUE);
+			assertThat(s).isInstanceOf(ScopePassingSpanSubscriber.class);
+		}
+
+		@Override
+		public void onNext(Object o) {
+
+		}
+
+		@Override
+		public void onError(Throwable t) {
+
+		}
+
+		@Override
+		public void onComplete() {
+
+		}
+	};
+
 	AnnotationConfigApplicationContext springContext = new AnnotationConfigApplicationContext();
 
-	@AfterEach
+	@Before
+	public void resetHooks() {
+		// There's an assumption some other test is leaking hooks, so we clear them all to
+		// prevent should_not_scope_scalar_subscribe from being interfered with.
+		Hooks.resetOnEachOperator(SLEUTH_TRACE_REACTOR_KEY);
+		Hooks.resetOnLastOperator(SLEUTH_TRACE_REACTOR_KEY);
+		Schedulers.removeExecutorServiceDecorator(SLEUTH_REACTOR_EXECUTOR_SERVICE_KEY);
+	}
+
+	@After
 	public void close() {
 		springContext.close();
 		currentTraceContext.close();
@@ -70,6 +138,18 @@ public class ScopePassingSpanSubscriberTests {
 		then((String) subscriber.currentContext().get("foo")).isEqualTo("bar");
 	}
 
+	/**
+	 * This ensures when the desired context is in the reactor context we don't copy it.
+	 */
+	@Test
+	public void should_not_redundantly_copy_context() {
+		Context initial = Context.of(TraceContext.class, context);
+		ScopePassingSpanSubscriber<?> subscriber = new ScopePassingSpanSubscriber<>(null,
+				initial, this.currentTraceContext, context);
+
+		then(initial).isSameAs(subscriber.currentContext());
+	}
+
 	@Test
 	public void should_set_empty_context_when_context_is_null() {
 		ScopePassingSpanSubscriber<?> subscriber = new ScopePassingSpanSubscriber<>(null,
@@ -89,4 +169,47 @@ public class ScopePassingSpanSubscriberTests {
 		}
 	}
 
+	@Test
+	public void should_not_scope_scalar_subscribe() {
+		springContext.registerBean(CurrentTraceContext.class, () -> currentTraceContext);
+		springContext.refresh();
+
+		Function<? super Publisher<Integer>, ? extends Publisher<Integer>> transformer = scopePassingSpanOperator(
+				this.springContext);
+
+		try (Scope ws = this.currentTraceContext.newScope(context)) {
+
+			transformer.apply(Mono.just(1))
+					.subscribe(assertNotScopePassingSpanSubscriber);
+
+			transformer.apply(Mono.error(new Exception()))
+					.subscribe(assertNotScopePassingSpanSubscriber);
+
+			transformer.apply(Mono.empty())
+					.subscribe(assertNotScopePassingSpanSubscriber);
+
+		}
+	}
+
+	@Test
+	public void should_scope_scalar_hide_subscribe() {
+		springContext.registerBean(CurrentTraceContext.class, () -> currentTraceContext);
+		springContext.refresh();
+
+		Function<? super Publisher<Integer>, ? extends Publisher<Integer>> transformer = scopePassingSpanOperator(
+				this.springContext);
+
+		try (Scope ws = this.currentTraceContext.newScope(context)) {
+
+			transformer.apply(Mono.just(1).hide())
+					.subscribe(assertScopePassingSpanSubscriber);
+
+			transformer.apply(Mono.<Integer>error(new Exception()).hide())
+					.subscribe(assertScopePassingSpanSubscriber);
+
+			transformer.apply(Mono.<Integer>empty().hide())
+					.subscribe(assertScopePassingSpanSubscriber);
+		}
+	}
+
 }