Use executor for blocking I/O in Reactor request factory

Prior to this commit, the `ReactorClientHttpRequestFactory` and the
`ReactorClientHttpRequest` would use the `Executor` from the current
event loop for performing write operations.
Depending on I/O demand, this work could be blocked and would result in
blocked Netty event loop executors and the HTTP client hanging.

This commit ensures that the client uses a separate Executor for such
operations. If the application does not provide one on the request
factory, a `Schedulers#boundedElastic` instance will be used.

Fixes spring-projectsgh-34707

Author: bclozel

spring-web/src/main/java/org/springframework/http/client/ReactorClientHttpRequest.java

@@ -21,13 +21,13 @@
 import java.net.URI;
 import java.time.Duration;
 import java.util.concurrent.Executor;
-import java.util.concurrent.atomic.AtomicReference;
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
 import org.reactivestreams.FlowAdapters;
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
 import reactor.netty.NettyOutbound;
 import reactor.netty.http.client.HttpClient;
 import reactor.netty.http.client.HttpClientRequest;
@@ -43,6 +43,7 @@
  *
  * @author Arjen Poutsma
  * @author Juergen Hoeller
+ * @author Brian Clozel
  * @since 6.1
  */
 final class ReactorClientHttpRequest extends AbstractStreamingClientHttpRequest {
@@ -53,6 +54,8 @@ final class ReactorClientHttpRequest extends AbstractStreamingClientHttpRequest
 
 	private final URI uri;
 
+	private final Executor executor;
+
 	@Nullable
 	private final Duration exchangeTimeout;
 
@@ -65,19 +68,31 @@ final class ReactorClientHttpRequest extends AbstractStreamingClientHttpRequest
 	 * @since 6.2
 	 */
 	public ReactorClientHttpRequest(HttpClient httpClient, HttpMethod method, URI uri) {
-		this.httpClient = httpClient;
-		this.method = method;
-		this.uri = uri;
-		this.exchangeTimeout = null;
+		this(httpClient, method, uri, null);
+	}
+
+	/**
+	 * Create an instance.
+	 * <p>If no executor is provided, the request will use an {@link Schedulers#boundedElastic() elastic scheduler}
+	 * for performing blocking I/O operations.
+	 * @param httpClient the client to perform the request with
+	 * @param executor the executor to use
+	 * @param method the HTTP method
+	 * @param uri the URI for the request
+	 * @since 6.2.13
+	 */
+	public ReactorClientHttpRequest(HttpClient httpClient, HttpMethod method, URI uri, @Nullable Executor executor) {
+		this(httpClient, method, uri, executor, null);
 	}
 
 	/**
 	 * Package private constructor for use until exchangeTimeout is removed.
 	 */
-	ReactorClientHttpRequest(HttpClient httpClient, HttpMethod method, URI uri, @Nullable Duration exchangeTimeout) {
+	ReactorClientHttpRequest(HttpClient httpClient, HttpMethod method, URI uri, @Nullable Executor executor, @Nullable Duration exchangeTimeout) {
 		this.httpClient = httpClient;
 		this.method = method;
 		this.uri = uri;
+		this.executor = (executor != null) ? executor : Schedulers.boundedElastic()::schedule;
 		this.exchangeTimeout = exchangeTimeout;
 	}
 
@@ -92,11 +107,7 @@ public ReactorClientHttpRequest(HttpClient httpClient, HttpMethod method, URI ur
 	public ReactorClientHttpRequest(
 			HttpClient httpClient, URI uri, HttpMethod method,
 			@Nullable Duration exchangeTimeout, @Nullable Duration readTimeout) {
-
-		this.httpClient = httpClient;
-		this.method = method;
-		this.uri = uri;
-		this.exchangeTimeout = exchangeTimeout;
+		this(httpClient, method, uri, null, exchangeTimeout);
 	}
 
 
@@ -150,13 +161,10 @@ private Publisher<Void> send(
 			return Mono.empty();
 		}
 
-		AtomicReference<Executor> executorRef = new AtomicReference<>();
-
 		return outbound
-				.withConnection(connection -> executorRef.set(connection.channel().eventLoop()))
 				.send(FlowAdapters.toPublisher(new OutputStreamPublisher<>(
 						os -> body.writeTo(StreamUtils.nonClosing(os)), new ByteBufMapper(outbound),
-						executorRef.getAndSet(null), null)));
+						this.executor, null)));
 	}
 
 	static IOException convertException(RuntimeException ex) {

spring-web/src/main/java/org/springframework/http/client/ReactorClientHttpRequestFactory.java

@@ -19,11 +19,13 @@
 import java.io.IOException;
 import java.net.URI;
 import java.time.Duration;
+import java.util.concurrent.Executor;
 import java.util.function.Function;
 
 import io.netty.channel.ChannelOption;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import reactor.core.scheduler.Schedulers;
 import reactor.netty.http.client.HttpClient;
 import reactor.netty.resources.ConnectionProvider;
 import reactor.netty.resources.LoopResources;
@@ -42,6 +44,7 @@
  * @author Arjen Poutsma
  * @author Juergen Hoeller
  * @author Sebastien Deleuze
+ * @author Brian Clozel
  * @since 6.2
  */
 public class ReactorClientHttpRequestFactory implements ClientHttpRequestFactory, SmartLifecycle {
@@ -58,6 +61,9 @@ public class ReactorClientHttpRequestFactory implements ClientHttpRequestFactory
 	@Nullable
 	private final Function<HttpClient, HttpClient> mapper;
 
+	@Nullable
+	private Executor executor;
+
 	@Nullable
 	private Integer connectTimeout;
 
@@ -129,6 +135,16 @@ private HttpClient createHttpClient(ReactorResourceFactory factory, Function<Htt
 		return client;
 	}
 
+	/**
+	 * Set the {@code Executor} to use for performing blocking I/O operations.
+	 * <p>If no executor is provided, the request will use an {@link Schedulers#boundedElastic() elastic scheduler}.
+	 * @param executor the executor to use.
+	 * @since 6.2.13
+	 */
+	public void setExecutor(Executor executor) {
+		Assert.notNull(executor, "Executor must not be null");
+		this.executor = executor;
+	}
 
 	/**
 	 * Set the connect timeout value on the underlying client.
@@ -219,7 +235,7 @@ public ClientHttpRequest createRequest(URI uri, HttpMethod httpMethod) throws IO
 					"Expected HttpClient or ResourceFactory and mapper");
 			client = createHttpClient(this.resourceFactory, this.mapper);
 		}
-		return new ReactorClientHttpRequest(client, httpMethod, uri, this.exchangeTimeout);
+		return new ReactorClientHttpRequest(client, httpMethod, uri, this.executor, this.exchangeTimeout);
 	}