Honor timeout in ZipkinWebClientSender

Unfortunately there's no good way to configure connect and read timeout
separately, which works for all supported reactive clients. This
implementation applies a timeout through Reactor's timeout method. The
timeout from the properties is summed together and this is the applied
timeout. While not perfect, this is better than no timeout at all.

Closes spring-projectsgh-31496

Author: mhalbritter

spring-boot-project/spring-boot-actuator-autoconfigure/src/main/java/org/springframework/boot/actuate/autoconfigure/tracing/zipkin/ZipkinConfigurations.java

@@ -119,7 +119,8 @@ ZipkinWebClientSender webClientSender(ZipkinProperties properties,
 				.getIfAvailable(() -> new PropertiesZipkinConnectionDetails(properties));
 			WebClient.Builder builder = WebClient.builder();
 			customizers.orderedStream().forEach((customizer) -> customizer.customize(builder));
-			return new ZipkinWebClientSender(connectionDetails.getSpanEndpoint(), builder.build());
+			return new ZipkinWebClientSender(connectionDetails.getSpanEndpoint(), builder.build(),
+					properties.getConnectTimeout().plus(properties.getReadTimeout()));
 		}
 
 	}

spring-boot-project/spring-boot-actuator-autoconfigure/src/main/java/org/springframework/boot/actuate/autoconfigure/tracing/zipkin/ZipkinWebClientSender.java

@@ -16,6 +16,8 @@
 
 package org.springframework.boot.actuate.autoconfigure.tracing.zipkin;
 
+import java.time.Duration;
+
 import reactor.core.publisher.Mono;
 import zipkin2.Call;
 import zipkin2.Callback;
@@ -28,21 +30,25 @@
  * An {@link HttpSender} which uses {@link WebClient} for HTTP communication.
  *
  * @author Stefan Bratanov
+ * @author Moritz Halbritter
  */
 class ZipkinWebClientSender extends HttpSender {
 
 	private final String endpoint;
 
 	private final WebClient webClient;
 
-	ZipkinWebClientSender(String endpoint, WebClient webClient) {
+	private final Duration timeout;
+
+	ZipkinWebClientSender(String endpoint, WebClient webClient, Duration timeout) {
 		this.endpoint = endpoint;
 		this.webClient = webClient;
+		this.timeout = timeout;
 	}
 
 	@Override
 	public HttpPostCall sendSpans(byte[] batchedEncodedSpans) {
-		return new WebClientHttpPostCall(this.endpoint, batchedEncodedSpans, this.webClient);
+		return new WebClientHttpPostCall(this.endpoint, batchedEncodedSpans, this.webClient, this.timeout);
 	}
 
 	private static class WebClientHttpPostCall extends HttpPostCall {
@@ -51,15 +57,18 @@ private static class WebClientHttpPostCall extends HttpPostCall {
 
 		private final WebClient webClient;
 
-		WebClientHttpPostCall(String endpoint, byte[] body, WebClient webClient) {
+		private final Duration timeout;
+
+		WebClientHttpPostCall(String endpoint, byte[] body, WebClient webClient, Duration timeout) {
 			super(body);
 			this.endpoint = endpoint;
 			this.webClient = webClient;
+			this.timeout = timeout;
 		}
 
 		@Override
 		public Call<Void> clone() {
-			return new WebClientHttpPostCall(this.endpoint, getUncompressedBody(), this.webClient);
+			return new WebClientHttpPostCall(this.endpoint, getUncompressedBody(), this.webClient, this.timeout);
 		}
 
 		@Override
@@ -79,7 +88,8 @@ private Mono<ResponseEntity<Void>> sendRequest() {
 				.headers(this::addDefaultHeaders)
 				.bodyValue(getBody())
 				.retrieve()
-				.toBodilessEntity();
+				.toBodilessEntity()
+				.timeout(this.timeout);
 		}
 
 		private void addDefaultHeaders(HttpHeaders headers) {

spring-boot-project/spring-boot-actuator-autoconfigure/src/test/java/org/springframework/boot/actuate/autoconfigure/tracing/zipkin/ZipkinHttpSenderTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2012-2022 the original author or authors.
+ * Copyright 2012-2023 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.
@@ -25,6 +25,7 @@
 import java.util.concurrent.atomic.AtomicReference;
 
 import org.awaitility.Awaitility;
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import zipkin2.Callback;
@@ -42,19 +43,25 @@
  */
 abstract class ZipkinHttpSenderTests {
 
-	protected Sender sut;
+	protected Sender sender;
 
-	abstract Sender createSut();
+	abstract Sender createSender();
 
 	@BeforeEach
-	void setUp() {
-		this.sut = createSut();
+	void beforeEach() throws Exception {
+		this.sender = createSender();
+	}
+
+	@AfterEach
+	void afterEach() throws IOException {
+		this.sender.close();
 	}
 
 	@Test
 	void sendSpansShouldThrowIfCloseWasCalled() throws IOException {
-		this.sut.close();
-		assertThatThrownBy(() -> this.sut.sendSpans(Collections.emptyList())).isInstanceOf(ClosedSenderException.class);
+		this.sender.close();
+		assertThatThrownBy(() -> this.sender.sendSpans(Collections.emptyList()))
+			.isInstanceOf(ClosedSenderException.class);
 	}
 
 	protected void makeRequest(List<byte[]> encodedSpans, boolean async) throws IOException {
@@ -68,8 +75,12 @@ protected void makeRequest(List<byte[]> encodedSpans, boolean async) throws IOEx
 	}
 
 	protected CallbackResult makeAsyncRequest(List<byte[]> encodedSpans) {
+		return makeAsyncRequest(this.sender, encodedSpans);
+	}
+
+	protected CallbackResult makeAsyncRequest(Sender sender, List<byte[]> encodedSpans) {
 		AtomicReference<CallbackResult> callbackResult = new AtomicReference<>();
-		this.sut.sendSpans(encodedSpans).enqueue(new Callback<>() {
+		sender.sendSpans(encodedSpans).enqueue(new Callback<>() {
 			@Override
 			public void onSuccess(Void value) {
 				callbackResult.set(new CallbackResult(true, null));
@@ -84,7 +95,11 @@ public void onError(Throwable t) {
 	}
 
 	protected void makeSyncRequest(List<byte[]> encodedSpans) throws IOException {
-		this.sut.sendSpans(encodedSpans).execute();
+		makeSyncRequest(this.sender, encodedSpans);
+	}
+
+	protected void makeSyncRequest(Sender sender, List<byte[]> encodedSpans) throws IOException {
+		sender.sendSpans(encodedSpans).execute();
 	}
 
 	protected byte[] toByteArray(String input) {

spring-boot-project/spring-boot-actuator-autoconfigure/src/test/java/org/springframework/boot/actuate/autoconfigure/tracing/zipkin/ZipkinRestTemplateSenderTests.java

@@ -54,14 +54,16 @@ class ZipkinRestTemplateSenderTests extends ZipkinHttpSenderTests {
 	private MockRestServiceServer mockServer;
 
 	@Override
-	Sender createSut() {
+	Sender createSender() {
 		RestTemplate restTemplate = new RestTemplate();
 		this.mockServer = MockRestServiceServer.createServer(restTemplate);
 		return new ZipkinRestTemplateSender(ZIPKIN_URL, restTemplate);
 	}
 
 	@AfterEach
-	void tearDown() {
+	@Override
+	void afterEach() throws IOException {
+		super.afterEach();
 		this.mockServer.verify();
 	}
 
@@ -71,15 +73,15 @@ void checkShouldSendEmptySpanList() {
 			.andExpect(method(HttpMethod.POST))
 			.andExpect(content().string("[]"))
 			.andRespond(withStatus(HttpStatus.ACCEPTED));
-		assertThat(this.sut.check()).isEqualTo(CheckResult.OK);
+		assertThat(this.sender.check()).isEqualTo(CheckResult.OK);
 	}
 
 	@Test
 	void checkShouldNotRaiseException() {
 		this.mockServer.expect(requestTo(ZIPKIN_URL))
 			.andExpect(method(HttpMethod.POST))
 			.andRespond(withStatus(HttpStatus.INTERNAL_SERVER_ERROR));
-		CheckResult result = this.sut.check();
+		CheckResult result = this.sender.check();
 		assertThat(result.ok()).isFalse();
 		assertThat(result.error()).hasMessageContaining("500 Internal Server Error");
 	}

spring-boot-project/spring-boot-actuator-autoconfigure/src/test/java/org/springframework/boot/actuate/autoconfigure/tracing/zipkin/ZipkinWebClientSenderTests.java

@@ -17,16 +17,21 @@
 package org.springframework.boot.actuate.autoconfigure.tracing.zipkin;
 
 import java.io.IOException;
+import java.time.Duration;
 import java.util.Base64;
 import java.util.Collections;
 import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.function.Consumer;
 
 import okhttp3.mockwebserver.MockResponse;
 import okhttp3.mockwebserver.MockWebServer;
+import okhttp3.mockwebserver.QueueDispatcher;
 import okhttp3.mockwebserver.RecordedRequest;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.ValueSource;
@@ -45,33 +50,48 @@
  */
 class ZipkinWebClientSenderTests extends ZipkinHttpSenderTests {
 
+	private static ClearableDispatcher dispatcher;
+
 	private static MockWebServer mockBackEnd;
 
 	private static String ZIPKIN_URL;
 
 	@BeforeAll
 	static void beforeAll() throws IOException {
+		dispatcher = new ClearableDispatcher();
 		mockBackEnd = new MockWebServer();
+		mockBackEnd.setDispatcher(dispatcher);
 		mockBackEnd.start();
-		ZIPKIN_URL = "http://localhost:%s/api/v2/spans".formatted(mockBackEnd.getPort());
+		ZIPKIN_URL = mockBackEnd.url("/api/v2/spans").toString();
 	}
 
 	@AfterAll
-	static void tearDown() throws IOException {
+	static void afterAll() throws IOException {
 		mockBackEnd.shutdown();
 	}
 
 	@Override
-	Sender createSut() {
+	@BeforeEach
+	void beforeEach() throws Exception {
+		super.beforeEach();
+		clearResponses();
+		clearRequests();
+	}
+
+	@Override
+	Sender createSender() {
+		return createSender(Duration.ofSeconds(10));
+	}
+
+	Sender createSender(Duration timeout) {
 		WebClient webClient = WebClient.builder().build();
-		return new ZipkinWebClientSender(ZIPKIN_URL, webClient);
+		return new ZipkinWebClientSender(ZIPKIN_URL, webClient, timeout);
 	}
 
 	@Test
 	void checkShouldSendEmptySpanList() throws InterruptedException {
 		mockBackEnd.enqueue(new MockResponse());
-		assertThat(this.sut.check()).isEqualTo(CheckResult.OK);
-
+		assertThat(this.sender.check()).isEqualTo(CheckResult.OK);
 		requestAssertions((request) -> {
 			assertThat(request.getMethod()).isEqualTo("POST");
 			assertThat(request.getBody().readUtf8()).isEqualTo("[]");
@@ -81,10 +101,9 @@ void checkShouldSendEmptySpanList() throws InterruptedException {
 	@Test
 	void checkShouldNotRaiseException() throws InterruptedException {
 		mockBackEnd.enqueue(new MockResponse().setResponseCode(500));
-		CheckResult result = this.sut.check();
+		CheckResult result = this.sender.check();
 		assertThat(result.ok()).isFalse();
 		assertThat(result.error()).hasMessageContaining("500 Internal Server Error");
-
 		requestAssertions((request) -> assertThat(request.getMethod()).isEqualTo("POST"));
 	}
 
@@ -94,7 +113,6 @@ void sendSpansShouldSendSpansToZipkin(boolean async) throws IOException, Interru
 		mockBackEnd.enqueue(new MockResponse());
 		List<byte[]> encodedSpans = List.of(toByteArray("span1"), toByteArray("span2"));
 		makeRequest(encodedSpans, async);
-
 		requestAssertions((request) -> {
 			assertThat(request.getMethod()).isEqualTo("POST");
 			assertThat(request.getHeader("Content-Type")).isEqualTo("application/json");
@@ -115,7 +133,6 @@ void sendSpansShouldHandleHttpFailures(boolean async) throws InterruptedExceptio
 			assertThatThrownBy(() -> makeSyncRequest(Collections.emptyList()))
 				.hasMessageContaining("500 Internal Server Error");
 		}
-
 		requestAssertions((request) -> assertThat(request.getMethod()).isEqualTo("POST"));
 	}
 
@@ -126,23 +143,56 @@ void sendSpansShouldCompressData(boolean async) throws IOException, InterruptedE
 		// This is gzip compressed 10000 times 'a'
 		byte[] compressed = Base64.getDecoder()
 			.decode("H4sIAAAAAAAA/+3BMQ0AAAwDIKFLj/k3UR8NcA8AAAAAAAAAAAADUsAZfeASJwAA");
-
 		mockBackEnd.enqueue(new MockResponse());
-
 		makeRequest(List.of(toByteArray(uncompressed)), async);
-
 		requestAssertions((request) -> {
 			assertThat(request.getMethod()).isEqualTo("POST");
 			assertThat(request.getHeader("Content-Type")).isEqualTo("application/json");
 			assertThat(request.getHeader("Content-Encoding")).isEqualTo("gzip");
 			assertThat(request.getBody().readByteArray()).isEqualTo(compressed);
 		});
+	}
 
+	@ParameterizedTest
+	@ValueSource(booleans = { true, false })
+	void shouldTimeout(boolean async) {
+		Sender sender = createSender(Duration.ofMillis(1));
+		MockResponse response = new MockResponse().setResponseCode(200).setHeadersDelay(100, TimeUnit.MILLISECONDS);
+		mockBackEnd.enqueue(response);
+		if (async) {
+			CallbackResult callbackResult = makeAsyncRequest(sender, Collections.emptyList());
+			assertThat(callbackResult.success()).isFalse();
+			assertThat(callbackResult.error()).isNotNull().isInstanceOf(TimeoutException.class);
+		}
+		else {
+			assertThatThrownBy(() -> makeSyncRequest(sender, Collections.emptyList()))
+				.hasCauseInstanceOf(TimeoutException.class);
+		}
 	}
 
 	private void requestAssertions(Consumer<RecordedRequest> assertions) throws InterruptedException {
 		RecordedRequest request = mockBackEnd.takeRequest();
 		assertThat(request).satisfies(assertions);
 	}
 
+	private static void clearRequests() throws InterruptedException {
+		RecordedRequest request;
+		do {
+			request = mockBackEnd.takeRequest(0, TimeUnit.SECONDS);
+		}
+		while (request != null);
+	}
+
+	private static void clearResponses() {
+		dispatcher.clear();
+	}
+
+	private static class ClearableDispatcher extends QueueDispatcher {
+
+		void clear() {
+			getResponseQueue().clear();
+		}
+
+	}
+
 }