Fixes null bugs on WebClient canceled request (#1548)

spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/web/client/TraceWebClientBeanPostProcessor.java

@@ -18,6 +18,7 @@ package org.springframework.cloud.sleuth.instrument.web.client;
 
 import java.util.Collections;
 import java.util.List;
+import java.util.concurrent.CancellationException;
 import java.util.function.Consumer;
 import java.util.function.Function;
 
@@ -131,9 +132,14 @@ final class TraceExchangeFilterFunction implements ExchangeFilterFunction {
 		}
 	};
 
-	private static final String CLIENT_SPAN_KEY = "sleuth.webclient.clientSpan";
+	static final String CLIENT_SPAN_KEY = "sleuth.webclient.clientSpan";
 
-	private static final String CANCELLED_SUBSCRIPTION_ERROR = "CANCELLED";
+	static final Exception CANCELLED_ERROR = new CancellationException("CANCELLED") {
+		@Override
+		public Throwable fillInStackTrace() {
+			return this; // stack trace doesn't add value here
+		}
+	};
 
 	final ConfigurableApplicationContext springContext;
 
@@ -170,6 +176,8 @@ final class TraceExchangeFilterFunction implements ExchangeFilterFunction {
 		}
 		MonoWebClientTrace trace = new MonoWebClientTrace(next, wrapper.buildRequest(),
 				this, span);
+		// TODO: investigate why this commit leaks a scope:
+		// 8f5bcdabd7af23df443e771432eb85597f3b3076
 		tracer().withSpanInScope(parentSpan);
 		return trace;
 	}
@@ -356,13 +364,14 @@ final class TraceExchangeFilterFunction implements ExchangeFilterFunction {
 				return this.context;
 			}
 
-			void handleReceive(Span clientSpan, ClientResponse clientResponse,
-					Throwable throwable) {
+			void handleReceive(Span clientSpan, @Nullable ClientResponse res,
+					@Nullable Throwable error) {
 				if (log.isTraceEnabled()) {
 					log.trace("Handling receive");
 				}
-				this.handler.handleReceive(new HttpClientResponse(clientResponse),
-						throwable, clientSpan);
+				HttpClientResponse response = res != null ? new HttpClientResponse(res)
+						: null;
+				this.handler.handleReceive(response, error, clientSpan);
 				if (log.isTraceEnabled()) {
 					log.trace("Closed scope");
 				}
@@ -374,32 +383,31 @@ final class TraceExchangeFilterFunction implements ExchangeFilterFunction {
 							+ this.span + "]");
 				}
 
-				this.span.tag("error", CANCELLED_SUBSCRIPTION_ERROR);
-				handleReceive(this.span, null, null);
+				handleReceive(this.span, null, CANCELLED_ERROR);
 			}
 
 			void terminateSpan(@Nullable ClientResponse clientResponse,
-					@Nullable Throwable throwable) {
+					@Nullable Throwable error) {
 				if (clientResponse == null) {
 					if (log.isDebugEnabled()) {
 						log.debug("No response was returned. Will close the span ["
 								+ this.span + "]");
 					}
-					handleReceive(this.span, clientResponse, throwable);
+					handleReceive(this.span, null, error);
 					return;
 				}
 				int statusCode = clientResponse.rawStatusCode();
-				boolean error = statusCode >= 400;
-				if (error) {
+				boolean isHttpError = statusCode >= 400;
+				if (isHttpError) {
 					if (log.isDebugEnabled()) {
 						log.debug(
 								"Non positive status code was returned from the call. Will close the span ["
 										+ this.span + "]");
 					}
-					throwable = new RestClientException(
+					error = new RestClientException(
 							"Status code of the response is [" + statusCode + "]");
 				}
-				handleReceive(this.span, clientResponse, throwable);
+				handleReceive(this.span, clientResponse, error);
 			}
 
 		}

spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/web/client/integration/WebClientTests.java

@@ -40,6 +40,7 @@ import com.netflix.loadbalancer.ILoadBalancer;
 import com.netflix.loadbalancer.Server;
 import junitparams.JUnitParamsRunner;
 import junitparams.Parameters;
+import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.http.HttpResponse;
 import org.apache.http.client.methods.HttpGet;
@@ -55,6 +56,8 @@ import org.junit.ClassRule;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.runner.RunWith;
+import org.reactivestreams.Subscription;
+import reactor.core.publisher.BaseSubscriber;
 import zipkin2.Annotation;
 import zipkin2.reporter.Reporter;
 
@@ -112,8 +115,7 @@ public class WebClientTests {
 	static final String SAMPLED_NAME = "X-B3-Sampled";
 	static final String PARENT_ID_NAME = "X-B3-ParentSpanId";
 
-	private static final org.apache.commons.logging.Log log = LogFactory
-			.getLog(WebClientTests.class);
+	private static final Log log = LogFactory.getLog(WebClientTests.class);
 
 	@Rule
 	public final SpringMethodRule springMethodRule = new SpringMethodRule();
@@ -351,7 +353,7 @@ public class WebClientTests {
 
 	@Test
 	@SuppressWarnings("unchecked")
-	public void shouldWorkWhenCustomStatusCodeIsReturned() throws InterruptedException {
+	public void shouldWorkWhenCustomStatusCodeIsReturned() {
 		Span span = this.tracer.nextSpan().name("foo").start();
 
 		try (Tracer.SpanInScope ws = this.tracer.withSpanInScope(span)) {
@@ -370,6 +372,21 @@ public class WebClientTests {
 				.contains("CLIENT");
 	}
 
+	@Test
+	public void shouldTagOnCancel() {
+		this.webClient.get().uri("http://localhost:" + this.port + "/doNotSkip")
+				.retrieve().bodyToMono(String.class)
+				.subscribe(new BaseSubscriber<String>() {
+					@Override
+					protected void hookOnSubscribe(Subscription subscription) {
+						cancel();
+					}
+				});
+
+		then(this.reporter.getSpans()).isNotEmpty();
+		then(this.reporter.getSpans().get(0).tags()).containsEntry("error", "CANCELLED");
+	}
+
 	@Test
 	public void shouldRespectSkipPattern() {
 		this.webClient.get().uri("http://localhost:" + this.port + "/skip").retrieve()