Fixes the invalid blocking of threads with WebClient; fixes gh-2127

spring-cloud-sleuth-autoconfigure/src/main/java/org/springframework/cloud/sleuth/autoconfig/zipkin2/ZipkinAutoConfiguration.java

@@ -16,9 +16,14 @@
 
 package org.springframework.cloud.sleuth.autoconfig.zipkin2;
 
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
+import javax.annotation.PreDestroy;
+
 import io.micrometer.core.instrument.MeterRegistry;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -79,6 +84,8 @@ import org.springframework.web.client.RestTemplate;
 @Import({ ZipkinSenderConfigurationImportSelector.class, ZipkinBraveConfiguration.class })
 public class ZipkinAutoConfiguration {
 
+	private final ExecutorService zipkinExecutor = Executors.newSingleThreadExecutor();
+
 	/**
 	 * Zipkin reporter bean name. Name of the bean matters for supporting multiple tracing
 	 * systems.
@@ -93,32 +100,21 @@ public class ZipkinAutoConfiguration {
 
 	private static final Log log = LogFactory.getLog(ZipkinAutoConfiguration.class);
 
+	@PreDestroy
+	void cleanup() {
+		this.zipkinExecutor.shutdown();
+	}
+
 	/** Limits {@link Sender#check()} to {@code deadlineMillis}. */
-	static CheckResult checkResult(Sender sender, long deadlineMillis) {
+	static CheckResult checkResult(ExecutorService zipkinExecutor, Sender sender, long deadlineMillis) {
-		CheckResult[] outcome = new CheckResult[1];
+		Future<CheckResult> future = zipkinExecutor.submit(sender::check);
-		Thread thread = new Thread(sender + " check()") {
-			@Override
-			public void run() {
-				try {
-					outcome[0] = sender.check();
-				}
-				catch (Throwable e) {
-					outcome[0] = CheckResult.failed(e);
-				}
-			}
-		};
-		thread.start();
 		try {
-			thread.join(deadlineMillis);
+			return future.get(deadlineMillis, TimeUnit.MILLISECONDS);
-			if (outcome[0] != null) {
-				return outcome[0];
-			}
-			thread.interrupt();
-			return CheckResult
-					.failed(new TimeoutException(thread.getName() + " timed out after " + deadlineMillis + "ms"));
 		}
-		catch (InterruptedException e) {
+		catch (TimeoutException e) {
-			Thread.currentThread().interrupt();
+			return CheckResult.failed(new TimeoutException("Timed out after " + deadlineMillis + "ms"));
+		}
+		catch (Exception e) {
 			return CheckResult.failed(e);
 		}
 	}
@@ -127,7 +123,7 @@ public class ZipkinAutoConfiguration {
 	@ConditionalOnMissingBean(name = REPORTER_BEAN_NAME)
 	Reporter<Span> reporter(ReporterMetrics reporterMetrics, ZipkinProperties zipkin,
 			@Qualifier(SENDER_BEAN_NAME) Sender sender) {
-		CheckResult checkResult = checkResult(sender, 1_000L);
+		CheckResult checkResult = checkResult(zipkinExecutor, sender, 1_000L);
 		logCheckResult(sender, checkResult);
 
 		// historical constraint. Note: AsyncReporter supports memory bounds

spring-cloud-sleuth-autoconfigure/src/test/java/org/springframework/cloud/sleuth/autoconfig/zipkin2/ZipkinHttpSenderConfigurationReactiveTests.java

@@ -48,7 +48,8 @@ class ZipkinHttpSenderConfigurationReactiveTests {
 		SpringApplication springApplication = new SpringApplication(Config.class);
 		springApplication.setWebApplicationType(WebApplicationType.REACTIVE);
 
-		try (ConfigurableApplicationContext context = springApplication.run("--spring.sleuth.noop.enabled=true")) {
+		try (ConfigurableApplicationContext context = springApplication.run("--spring.sleuth.noop.enabled=true",
+				"--server.port=0")) {
 			then(context.getBean(Sender.class)).isInstanceOf(WebClientSender.class);
 		}
 	}

tests/common/src/main/java/org/springframework/cloud/sleuth/autoconfig/zipkin2/ZipkinAutoConfigurationTests.java

@@ -18,6 +18,8 @@ package org.springframework.cloud.sleuth.autoconfig.zipkin2;
 
 import java.io.IOException;
 import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
@@ -77,6 +79,8 @@ public abstract class ZipkinAutoConfigurationTests {
 
 	protected abstract Class tracerConfiguration();
 
+	ExecutorService zipkinExecutor = Executors.newSingleThreadExecutor();
+
 	@BeforeEach
 	void setup() throws IOException {
 		server.start();
@@ -85,6 +89,7 @@ public abstract class ZipkinAutoConfigurationTests {
 	@AfterEach
 	void clean() throws IOException {
 		server.close();
+		zipkinExecutor.shutdown();
 	}
 
 	@Test
@@ -226,7 +231,7 @@ public abstract class ZipkinAutoConfigurationTests {
 		Sender sender = mock(Sender.class);
 		when(sender.check()).thenReturn(CheckResult.OK);
 
-		assertThat(ZipkinAutoConfiguration.checkResult(sender, 200).ok()).isTrue();
+		assertThat(ZipkinAutoConfiguration.checkResult(zipkinExecutor, sender, 200).ok()).isTrue();
 	}
 
 	@Test
@@ -235,7 +240,7 @@ public abstract class ZipkinAutoConfigurationTests {
 		RuntimeException exception = new RuntimeException("dead");
 		when(sender.check()).thenReturn(CheckResult.failed(exception));
 
-		assertThat(ZipkinAutoConfiguration.checkResult(sender, 200).error()).isSameAs(exception);
+		assertThat(ZipkinAutoConfiguration.checkResult(zipkinExecutor, sender, 200).error()).isSameAs(exception);
 	}
 
 	/** Bug in {@link Sender} as it shouldn't throw */
@@ -245,12 +250,12 @@ public abstract class ZipkinAutoConfigurationTests {
 		RuntimeException exception = new RuntimeException("dead");
 		when(sender.check()).thenThrow(exception);
 
-		assertThat(ZipkinAutoConfiguration.checkResult(sender, 200).error()).isSameAs(exception);
+		assertThat(ZipkinAutoConfiguration.checkResult(zipkinExecutor, sender, 200).error()).hasCause(exception);
 	}
 
 	@Test
 	public void checkResult_slow() {
-		assertThat(ZipkinAutoConfiguration.checkResult(new Sender() {
+		assertThat(ZipkinAutoConfiguration.checkResult(zipkinExecutor, new Sender() {
 			@Override
 			public CheckResult check() {
 				try {
@@ -286,7 +291,7 @@ public abstract class ZipkinAutoConfigurationTests {
 			public String toString() {
 				return "FakeSender{}";
 			}
-		}, 200).error()).isInstanceOf(TimeoutException.class).hasMessage("FakeSender{} check() timed out after 200ms");
+		}, 200).error()).isInstanceOf(TimeoutException.class).hasMessage("Timed out after 200ms");
 	}
 
 	@Configuration(proxyBeanMethods = false)