fix httpclient5.x connection leak

spring-cloud-gateway-server-mvc/pom.xml

@@ -135,5 +135,10 @@
 			<artifactId>rabbitmq</artifactId>
 			<scope>test</scope>
 		</dependency>
+		<dependency>
+			<groupId>org.apache.httpcomponents.client5</groupId>
+			<artifactId>httpclient5</artifactId>
+			<scope>test</scope>
+		</dependency>
     </dependencies>
 </project>

spring-cloud-gateway-server-mvc/src/main/java/org/springframework/cloud/gateway/server/mvc/common/MvcUtils.java

@@ -63,6 +63,11 @@ public abstract class MvcUtils {
 	 */
 	public static final String CLIENT_RESPONSE_INPUT_STREAM_ATTR = qualify("cachedClientResponseBody");
 
+	/**
+	 * Client response key.
+	 */
+	public static final String CLIENT_RESPONSE_ATTR = qualify("cachedClientResponse");
+
 	/**
 	 * CircuitBreaker execution exception attribute name.
 	 */

spring-cloud-gateway-server-mvc/src/main/java/org/springframework/cloud/gateway/server/mvc/filter/RetryFilterFunctions.java

@@ -33,6 +33,7 @@
 import org.springframework.http.HttpMethod;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.HttpStatusCode;
+import org.springframework.http.client.ClientHttpResponse;
 import org.springframework.retry.RetryContext;
 import org.springframework.retry.RetryPolicy;
 import org.springframework.retry.policy.CompositeRetryPolicy;
@@ -75,6 +76,7 @@ public static HandlerFilterFunction<ServerResponse, ServerResponse> retry(RetryC
 			if (config.isCacheBody()) {
 				MvcUtils.getOrCacheBody(request);
 			}
+			reset(request);
 			ServerResponse serverResponse = next.handle(request);
 
 			if (isRetryableStatusCode(serverResponse.statusCode(), config)
@@ -86,6 +88,14 @@ && isRetryableMethod(request.method(), config)) {
 		});
 	}
 
+	private static void reset(ServerRequest request) throws IOException {
+		ClientHttpResponse clientHttpResponse = MvcUtils.getAttribute(request, MvcUtils.CLIENT_RESPONSE_ATTR);
+		if (clientHttpResponse != null) {
+			clientHttpResponse.close();
+			MvcUtils.putAttribute(request, MvcUtils.CLIENT_RESPONSE_ATTR, null);
+		}
+	}
+
 	private static boolean isRetryableStatusCode(HttpStatusCode httpStatus, RetryConfig config) {
 		return config.getSeries().stream().anyMatch(series -> HttpStatus.Series.resolve(httpStatus.value()) == series);
 	}

spring-cloud-gateway-server-mvc/src/main/java/org/springframework/cloud/gateway/server/mvc/handler/ClientHttpRequestFactoryProxyExchange.java

@@ -56,6 +56,7 @@ public ServerResponse exchange(Request request) {
 			InputStream body = clientHttpResponse.getBody();
 			// put the body input stream in a request attribute so filters can read it.
 			MvcUtils.putAttribute(request.getServerRequest(), MvcUtils.CLIENT_RESPONSE_INPUT_STREAM_ATTR, body);
+			MvcUtils.putAttribute(request.getServerRequest(), MvcUtils.CLIENT_RESPONSE_ATTR, clientHttpResponse);
 			ServerResponse serverResponse = GatewayServerResponse.status(clientHttpResponse.getStatusCode())
 				.build((req, httpServletResponse) -> {
 					try (clientHttpResponse) {

spring-cloud-gateway-server-mvc/src/main/java/org/springframework/cloud/gateway/server/mvc/handler/RestClientProxyExchange.java

@@ -72,6 +72,7 @@ private ServerResponse doExchange(Request request, ClientHttpResponse clientResp
 		InputStream body = clientResponse.getBody();
 		// put the body input stream in a request attribute so filters can read it.
 		MvcUtils.putAttribute(request.getServerRequest(), MvcUtils.CLIENT_RESPONSE_INPUT_STREAM_ATTR, body);
+		MvcUtils.putAttribute(request.getServerRequest(), MvcUtils.CLIENT_RESPONSE_ATTR, clientResponse);
 		ServerResponse serverResponse = GatewayServerResponse.status(clientResponse.getStatusCode())
 			.build((req, httpServletResponse) -> {
 				try (clientResponse) {

spring-cloud-gateway-server-mvc/src/test/java/org/springframework/cloud/gateway/server/mvc/GatewayServerMvcAutoConfigurationTests.java

@@ -33,7 +33,7 @@
 import org.springframework.boot.builder.SpringApplicationBuilder;
 import org.springframework.boot.http.client.ClientHttpRequestFactoryBuilder;
 import org.springframework.boot.http.client.ClientHttpRequestFactorySettings;
-import org.springframework.boot.http.client.SimpleClientHttpRequestFactoryBuilder;
+import org.springframework.boot.http.client.ReactorClientHttpRequestFactoryBuilder;
 import org.springframework.boot.test.context.runner.ApplicationContextRunner;
 import org.springframework.cloud.gateway.server.mvc.filter.FormFilter;
 import org.springframework.cloud.gateway.server.mvc.filter.ForwardedRequestHeadersFilter;
@@ -47,6 +47,7 @@
 import org.springframework.context.ConfigurableApplicationContext;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.springframework.boot.autoconfigure.http.client.HttpClientProperties.Factory.REACTOR;
 
 public class GatewayServerMvcAutoConfigurationTests {
 
@@ -151,7 +152,7 @@ void gatewayHttpClientPropertiesWork() {
 		assertThat(properties.getConnectTimeout()).hasSeconds(1);
 		assertThat(properties.getReadTimeout()).hasSeconds(2);
 		assertThat(properties.getSsl().getBundle()).isEqualTo("mybundle");
-		assertThat(properties.getFactory()).isNull();
+		assertThat(properties.getFactory()).isEqualTo(REACTOR);
 		assertThat(settings.readTimeout()).isEqualTo(Duration.ofSeconds(2));
 		assertThat(settings.connectTimeout()).isEqualTo(Duration.ofSeconds(1));
 		assertThat(settings.sslBundle()).isNotNull();
@@ -187,10 +188,10 @@ void bootHttpClientPropertiesWork() {
 	@Test
 	void settingHttpClientFactoryWorks() {
 		ConfigurableApplicationContext context = new SpringApplicationBuilder(TestConfig.class)
-			.properties("spring.main.web-application-type=none", "spring.http.client.factory=simple")
+			.properties("spring.main.web-application-type=none")
 			.run();
 		ClientHttpRequestFactoryBuilder<?> builder = context.getBean(ClientHttpRequestFactoryBuilder.class);
-		assertThat(builder).isInstanceOf(SimpleClientHttpRequestFactoryBuilder.class);
+		assertThat(builder).isInstanceOf(ReactorClientHttpRequestFactoryBuilder.class);
 	}
 
 	@SpringBootConfiguration

spring-cloud-gateway-server-mvc/src/test/java/org/springframework/cloud/gateway/server/mvc/filter/RetryFilterFunctionTests.java

@@ -16,42 +16,55 @@
 
 package org.springframework.cloud.gateway.server.mvc.filter;
 
+import java.time.Duration;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.apache.hc.core5.util.Timeout;
 import org.junit.jupiter.api.Test;
 
+import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.SpringBootConfiguration;
 import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.http.client.ClientHttpRequestFactoryBuilder;
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.boot.test.context.SpringBootTest.WebEnvironment;
 import org.springframework.boot.test.web.server.LocalServerPort;
+import org.springframework.cloud.gateway.server.mvc.config.GatewayMvcProperties;
+import org.springframework.cloud.gateway.server.mvc.handler.ProxyExchange;
+import org.springframework.cloud.gateway.server.mvc.handler.ProxyExchangeHandlerFunction;
+import org.springframework.cloud.gateway.server.mvc.handler.RestClientProxyExchange;
 import org.springframework.cloud.gateway.server.mvc.test.HttpbinTestcontainers;
 import org.springframework.cloud.gateway.server.mvc.test.LocalServerPortUriResolver;
 import org.springframework.cloud.gateway.server.mvc.test.TestLoadBalancerConfig;
 import org.springframework.cloud.gateway.server.mvc.test.client.TestRestClient;
 import org.springframework.cloud.loadbalancer.annotation.LoadBalancerClient;
+import org.springframework.context.ApplicationContext;
 import org.springframework.context.annotation.Bean;
+import org.springframework.context.event.ContextRefreshedEvent;
 import org.springframework.core.log.LogMessage;
 import org.springframework.http.HttpMethod;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.ResponseEntity;
+import org.springframework.http.client.ClientHttpRequestFactory;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.util.StringUtils;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.bind.annotation.RequestBody;
 import org.springframework.web.bind.annotation.RequestParam;
 import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.client.RestClient;
 import org.springframework.web.servlet.function.RouterFunction;
 import org.springframework.web.servlet.function.ServerResponse;
 
 import static org.springframework.cloud.gateway.server.mvc.filter.FilterFunctions.adaptCachedBody;
 import static org.springframework.cloud.gateway.server.mvc.filter.FilterFunctions.prefixPath;
+import static org.springframework.cloud.gateway.server.mvc.filter.FilterFunctions.setPath;
 import static org.springframework.cloud.gateway.server.mvc.filter.RetryFilterFunctions.retry;
 import static org.springframework.cloud.gateway.server.mvc.handler.GatewayRouterFunctions.route;
 import static org.springframework.cloud.gateway.server.mvc.handler.HandlerFunctions.http;
@@ -93,6 +106,17 @@ public void retryBodyWorks() {
 			.isEqualTo("3");
 	}
 
+	@Test
+	public void retryWorksWithHttpComponentsClient() {
+		restClient.get()
+			.uri("/retrywithhttpcomponentsclient?key=retryWorksWithHttpComponentsClient")
+			.exchange()
+			.expectStatus()
+			.isOk()
+			.expectBody(String.class)
+			.isEqualTo("3");
+	}
+
 	@SpringBootConfiguration
 	@EnableAutoConfiguration
 	@LoadBalancerClient(name = "httpbin", configuration = TestLoadBalancerConfig.Httpbin.class)
@@ -110,6 +134,41 @@ public RouterFunction<ServerResponse> gatewayRouterFunctionsRetry() {
 			// @formatter:on
 		}
 
+		@Bean
+		public RouterFunction<ServerResponse> gatewayRouterFunctionsRetryWithHttpComponentsClient(
+				GatewayMvcProperties properties,
+				ObjectProvider<HttpHeadersFilter.RequestHttpHeadersFilter> requestHttpHeadersFilters,
+				ObjectProvider<HttpHeadersFilter.ResponseHttpHeadersFilter> responseHttpHeadersFilters,
+				ApplicationContext applicationContext) {
+
+			// build httpComponents client factory
+			ClientHttpRequestFactory clientHttpRequestFactory = ClientHttpRequestFactoryBuilder.httpComponents()
+				.withConnectionManagerCustomizer(builder -> builder.setMaxConnTotal(2).setMaxConnPerRoute(2))
+				.withDefaultRequestConfigCustomizer(
+						c -> c.setConnectionRequestTimeout(Timeout.of(Duration.ofMillis(3000))))
+				.build();
+
+			// build proxyExchange use httpComponents
+			RestClient.Builder restClientBuilder = RestClient.builder();
+			restClientBuilder.requestFactory(clientHttpRequestFactory);
+			ProxyExchange proxyExchange = new RestClientProxyExchange(restClientBuilder.build(), properties);
+
+			// build handler function use httpComponents
+			ProxyExchangeHandlerFunction function = new ProxyExchangeHandlerFunction(proxyExchange,
+					requestHttpHeadersFilters, responseHttpHeadersFilters);
+			function.onApplicationEvent(new ContextRefreshedEvent(applicationContext));
+
+			// @formatter:off
+			return route("testretrywithhttpcomponentsclient")
+					.GET("/retrywithhttpcomponentsclient", function)
+					.before(new LocalServerPortUriResolver())
+					.filter(retry(3))
+					.filter(setPath("/retry"))
+					.filter(prefixPath("/do"))
+					.build();
+			// @formatter:on
+		}
+
 		@Bean
 		public RouterFunction<ServerResponse> gatewayRouterFunctionsRetryBody() {
 			// @formatter:off

spring-cloud-gateway-server-mvc/src/test/resources/application.yml

@@ -6,4 +6,7 @@ logging:
     org.springframework.retry: TRACE
 spring:
   mvc:
-    log-request-details: true
+    log-request-details: true
+  http:
+    client:
+      factory: REACTOR