fix httpclient5.x connection leak

spring-cloud-gateway-integration-tests/httpclient/pom.xml

@@ -0,0 +1,77 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+		 xmlns="http://maven.apache.org/POM/4.0.0"
+		 xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+
+	<artifactId>httpclient</artifactId>
+	<packaging>jar</packaging>
+
+	<name>Spring Cloud Gateway HttpClient Integration Test</name>
+	<description>Spring Cloud Gateway HttpClient Integration Test</description>
+
+	<properties>
+	</properties>
+
+	<parent>
+		<groupId>org.springframework.cloud</groupId>
+		<artifactId>spring-cloud-gateway-integration-tests</artifactId>
+		<version>4.3.0-SNAPSHOT</version>
+		<relativePath>..</relativePath> <!-- lookup parent from repository -->
+	</parent>
+
+
+	<dependencies>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-web</artifactId>
+		</dependency>
+
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-webflux</artifactId>
+		</dependency>
+
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-starter-gateway-server-webmvc</artifactId>
+		</dependency>
+
+		<dependency>
+			<groupId>org.springframework.retry</groupId>
+			<artifactId>spring-retry</artifactId>
+		</dependency>
+
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-starter-loadbalancer</artifactId>
+		</dependency>
+
+		<dependency>
+			<groupId>org.apache.httpcomponents.client5</groupId>
+			<artifactId>httpclient5</artifactId>
+		</dependency>
+
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-test</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.assertj</groupId>
+			<artifactId>assertj-core</artifactId>
+			<scope>test</scope>
+		</dependency>
+	</dependencies>
+
+	<build>
+		<plugins>
+			<plugin>
+				<artifactId>maven-deploy-plugin</artifactId>
+				<configuration>
+					<skip>true</skip>
+				</configuration>
+			</plugin>
+		</plugins>
+	</build>
+</project>

spring-cloud-gateway-integration-tests/httpclient/src/main/java/org/springframework/cloud/gateway/tests/httpclient/HttpClientApplication.java

@@ -0,0 +1,121 @@
+/*
+ * Copyright 2013-2025 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.gateway.tests.httpclient;
+
+import java.time.Duration;
+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.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.SpringBootConfiguration;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.http.client.ClientHttpRequestFactoryBuilder;
+import org.springframework.boot.http.client.HttpComponentsClientHttpRequestFactoryBuilder;
+import org.springframework.cloud.client.DefaultServiceInstance;
+import org.springframework.cloud.loadbalancer.annotation.LoadBalancerClient;
+import org.springframework.cloud.loadbalancer.core.ServiceInstanceListSupplier;
+import org.springframework.cloud.loadbalancer.support.ServiceInstanceListSuppliers;
+import org.springframework.context.annotation.Bean;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.servlet.function.RouterFunction;
+import org.springframework.web.servlet.function.ServerResponse;
+
+import static org.springframework.cloud.gateway.server.mvc.filter.FilterFunctions.prefixPath;
+import static org.springframework.cloud.gateway.server.mvc.filter.LoadBalancerFilterFunctions.lb;
+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;
+
+/**
+ * @author jiangyuan
+ */
+@SpringBootConfiguration
+@EnableAutoConfiguration
+@LoadBalancerClient(name = "myservice", configuration = MyServiceConf.class)
+public class HttpClientApplication {
+
+	public static void main(String[] args) {
+		SpringApplication.run(HttpClientApplication.class, args);
+	}
+
+	@Bean
+	public HttpComponentsClientHttpRequestFactoryBuilder httpComponentsClientHttpRequestFactoryBuilder() {
+		return ClientHttpRequestFactoryBuilder.httpComponents()
+			.withConnectionManagerCustomizer(builder -> builder.setMaxConnTotal(2).setMaxConnPerRoute(2))
+			.withDefaultRequestConfigCustomizer(
+					c -> c.setConnectionRequestTimeout(Timeout.of(Duration.ofMillis(3000))));
+	}
+
+	@Bean
+	public RouterFunction<ServerResponse> gatewayRouterFunctionsRetry() {
+		return route("test-retry").GET("/retry", http())
+			.filter(lb("myservice"))
+			.filter(prefixPath("/do"))
+			.filter(retry(3))
+			.build();
+	}
+
+	@RestController
+	protected static class RetryController {
+
+		Log log = LogFactory.getLog(getClass());
+
+		ConcurrentHashMap<String, AtomicInteger> map = new ConcurrentHashMap<>();
+
+		@GetMapping("/do/retry")
+		public ResponseEntity<String> retry(@RequestParam("key") String key,
+				@RequestParam(name = "count", defaultValue = "3") int count,
+				@RequestParam(name = "failStatus", required = false) Integer failStatus) {
+			AtomicInteger num = map.computeIfAbsent(key, s -> new AtomicInteger());
+			int i = num.incrementAndGet();
+			log.warn("Retry count: " + i);
+			String body = String.valueOf(i);
+			if (i < count) {
+				HttpStatus httpStatus = HttpStatus.INTERNAL_SERVER_ERROR;
+				if (failStatus != null) {
+					httpStatus = HttpStatus.resolve(failStatus);
+				}
+				return ResponseEntity.status(httpStatus).header("X-Retry-Count", body).body("temporarily broken");
+			}
+			return ResponseEntity.status(HttpStatus.OK).header("X-Retry-Count", body).body(body);
+		}
+
+	}
+
+}
+
+class MyServiceConf {
+
+	@Value("${local.server.port}")
+	private int port = 0;
+
+	@Bean
+	public ServiceInstanceListSupplier staticServiceInstanceListSupplier() {
+		return ServiceInstanceListSuppliers.from("myservice",
+				new DefaultServiceInstance("myservice-1", "myservice", "localhost", port, false));
+	}
+
+}

spring-cloud-gateway-integration-tests/httpclient/src/main/resources/application.yml

@@ -0,0 +1,3 @@
+logging:
+  level:
+    org.springframework.cloud.gateway: TRACE

spring-cloud-gateway-integration-tests/httpclient/src/test/java/org/springframework/cloud/gateway/tests/httpclient/HttpClientApplicationTests.java

@@ -0,0 +1,42 @@
+/*
+ * Copyright 2013-2025 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.gateway.tests.httpclient;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.web.server.LocalServerPort;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.web.reactive.server.WebTestClient;
+
+/**
+ * @author jiangyuan
+ */
+@SpringBootTest(classes = HttpClientApplication.class, webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
+@DirtiesContext
+public class HttpClientApplicationTests {
+
+	@LocalServerPort
+	private int port;
+
+	@Test
+	public void retryWorks() {
+		WebTestClient client = WebTestClient.bindToServer().baseUrl("http://localhost:" + port).build();
+		client.get().uri("/retry?key=get").exchange().expectStatus().isOk().expectBody(String.class).isEqualTo("3");
+	}
+
+}

spring-cloud-gateway-integration-tests/pom.xml

@@ -24,6 +24,7 @@
 		<module>grpc</module>
 		<module>http2</module>
 		<module>mvc-failure-analyzer</module>
+		<module>httpclient</module>
 	</modules>
 
 	<build>

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) {