fix: Fix HttpClient resource leak in SelectorManager threads (#610)

Replace shouldCloseHttpClient boolean with Consumer<HttpClient> pattern.

- Remove .clientBuilder, .customizeClient, .connectTimeout methods
- Add .onCloseClient(Consumer<HttpClient>) with reflection cleanup default
- Replace boolean flag with Consumer pattern in constructors
- Use sun.misc.Unsafe to bypass JDK module restrictions
- Support both JDK 21+ close() and JDK 17 SelectorManager reflection
- Update tests with proper HTTP request validation

Author: 曾响

mcp-core/src/main/java/io/modelcontextprotocol/client/transport/HttpClientSseClientTransport.java

@@ -5,6 +5,8 @@
 package io.modelcontextprotocol.client.transport;
 
 import java.io.IOException;
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
 import java.net.URI;
 import java.net.http.HttpClient;
 import java.net.http.HttpRequest;
@@ -36,6 +38,7 @@
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.Sinks;
+import sun.misc.Unsafe;
 
 /**
  * Server-Sent Events (SSE) implementation of the
@@ -93,13 +96,6 @@ public class HttpClientSseClientTransport implements McpClientTransport {
 	 */
 	private final HttpClient httpClient;
 
-	/**
-	 * Flag indicating whether this transport should close the HttpClient when closing
-	 * gracefully. Set to true when the HttpClient is created internally by the builder,
-	 * false when provided externally.
-	 */
-	private final boolean shouldCloseHttpClient;
-
 	/** HTTP request builder for building requests to send messages to the server */
 	private final HttpRequest.Builder requestBuilder;
 
@@ -123,6 +119,12 @@ public class HttpClientSseClientTransport implements McpClientTransport {
 	 */
 	private final McpAsyncHttpClientRequestCustomizer httpRequestCustomizer;
 
+	/**
+	 * Consumer to handle HttpClient closure. If null, no cleanup is performed (external
+	 * HttpClient).
+	 */
+	private final Consumer<HttpClient> onCloseClient;
+
 	/**
 	 * Creates a new transport instance with custom HTTP client builder, object mapper,
 	 * and headers.
@@ -137,7 +139,7 @@ public class HttpClientSseClientTransport implements McpClientTransport {
 	 */
 	HttpClientSseClientTransport(HttpClient httpClient, HttpRequest.Builder requestBuilder, String baseUri,
 			String sseEndpoint, McpJsonMapper jsonMapper, McpAsyncHttpClientRequestCustomizer httpRequestCustomizer,
-			boolean shouldCloseHttpClient) {
+			Consumer<HttpClient> onCloseClient) {
 		Assert.notNull(jsonMapper, "jsonMapper must not be null");
 		Assert.hasText(baseUri, "baseUri must not be empty");
 		Assert.hasText(sseEndpoint, "sseEndpoint must not be empty");
@@ -150,7 +152,7 @@ public class HttpClientSseClientTransport implements McpClientTransport {
 		this.httpClient = httpClient;
 		this.requestBuilder = requestBuilder;
 		this.httpRequestCustomizer = httpRequestCustomizer;
-		this.shouldCloseHttpClient = shouldCloseHttpClient;
+		this.onCloseClient = onCloseClient;
 	}
 
 	@Override
@@ -188,6 +190,8 @@ public static class Builder {
 
 		private Duration connectTimeout = Duration.ofSeconds(10);
 
+		private Consumer<HttpClient> onCloseClient;
+
 		/**
 		 * Creates a new builder instance.
 		 */
@@ -230,18 +234,6 @@ public Builder sseEndpoint(String sseEndpoint) {
 			return this;
 		}
 
-		/**
-		 * Sets the HTTP client builder.
-		 * @param clientBuilder the HTTP client builder
-		 * @return this builder
-		 */
-		public Builder clientBuilder(HttpClient.Builder clientBuilder) {
-			Assert.notNull(clientBuilder, "clientBuilder must not be null");
-			this.clientBuilder = clientBuilder;
-			this.externalHttpClient = null; // Clear external client if builder is set
-			return this;
-		}
-
 		/**
 		 * Sets an external HttpClient instance to use instead of creating a new one. When
 		 * an external HttpClient is provided, the transport will not attempt to close it
@@ -255,17 +247,6 @@ public Builder httpClient(HttpClient httpClient) {
 			return this;
 		}
 
-		/**
-		 * Customizes the HTTP client builder.
-		 * @param clientCustomizer the consumer to customize the HTTP client builder
-		 * @return this builder
-		 */
-		public Builder customizeClient(final Consumer<HttpClient.Builder> clientCustomizer) {
-			Assert.notNull(clientCustomizer, "clientCustomizer must not be null");
-			clientCustomizer.accept(clientBuilder);
-			return this;
-		}
-
 		/**
 		 * Sets the HTTP request builder.
 		 * @param requestBuilder the HTTP request builder
@@ -335,13 +316,13 @@ public Builder asyncHttpRequestCustomizer(McpAsyncHttpClientRequestCustomizer as
 		}
 
 		/**
-		 * Sets the connection timeout for the HTTP client.
-		 * @param connectTimeout the connection timeout duration
+		 * Sets a custom consumer to handle HttpClient closure when the transport is
+		 * closed.
+		 * @param onCloseClient the consumer to handle HttpClient closure
 		 * @return this builder
 		 */
-		public Builder connectTimeout(Duration connectTimeout) {
-			Assert.notNull(connectTimeout, "connectTimeout must not be null");
-			this.connectTimeout = connectTimeout;
+		public Builder onCloseClient(Consumer<HttpClient> onCloseClient) {
+			this.onCloseClient = onCloseClient;
 			return this;
 		}
 
@@ -351,22 +332,22 @@ public Builder connectTimeout(Duration connectTimeout) {
 		 */
 		public HttpClientSseClientTransport build() {
 			HttpClient httpClient;
-			boolean shouldCloseHttpClient;
+			Consumer<HttpClient> closeHandler;
 
 			if (externalHttpClient != null) {
-				// Use external HttpClient, don't close it
+				// Use external HttpClient, use custom close handler or no-op
 				httpClient = externalHttpClient;
-				shouldCloseHttpClient = false;
+				closeHandler = onCloseClient; // null means no cleanup
 			}
 			else {
-				// Create new HttpClient, should close it
+				// Create new HttpClient, use custom close handler or default cleanup
 				httpClient = this.clientBuilder.connectTimeout(this.connectTimeout).build();
-				shouldCloseHttpClient = true;
+				closeHandler = onCloseClient != null ? onCloseClient
+						: HttpClientSseClientTransport::closeHttpClientResourcesStatic;
 			}
 
 			return new HttpClientSseClientTransport(httpClient, requestBuilder, baseUri, sseEndpoint,
-					jsonMapper == null ? McpJsonMapper.getDefault() : jsonMapper, httpRequestCustomizer,
-					shouldCloseHttpClient);
+					jsonMapper == null ? McpJsonMapper.getDefault() : jsonMapper, httpRequestCustomizer, closeHandler);
 		}
 
 	}
@@ -534,34 +515,52 @@ public Mono<Void> closeGracefully() {
 			if (subscription != null && !subscription.isDisposed()) {
 				subscription.dispose();
 			}
-		}).then(shouldCloseHttpClient ? Mono.fromRunnable(this::closeHttpClientResources) : Mono.empty());
+		}).then(onCloseClient != null ? Mono.fromRunnable(() -> onCloseClient.accept(httpClient)) : Mono.empty());
 	}
 
 	/**
-	 * Closes HttpClient resources including connection pools and selector threads. This
-	 * method uses reflection to access internal HttpClient implementation details.
+	 * Static method to close HttpClient resources using reflection.
 	 */
-	private void closeHttpClientResources() {
+	private static void closeHttpClientResourcesStatic(HttpClient httpClient) {
 		try {
-			// Access HttpClientImpl internal fields via reflection
-			Class<?> httpClientClass = httpClient.getClass();
+			// unsafe
+			Class<?> UnsafeClass = Class.forName("sun.misc.Unsafe");
+			Field unsafeField = UnsafeClass.getDeclaredField("theUnsafe");
+			unsafeField.setAccessible(true);
+			Unsafe unsafe = (Unsafe) unsafeField.get(null);
+			Module ObjectModule = Object.class.getModule();
+			Class<HttpClientSseClientTransport> currentClass = HttpClientSseClientTransport.class;
+			long addr = unsafe.objectFieldOffset(Class.class.getDeclaredField("module"));
+			unsafe.getAndSetObject(currentClass, addr, ObjectModule);
 
-			// Close SelectorManager if present
 			try {
-				java.lang.reflect.Field selectorManagerField = httpClientClass.getDeclaredField("selectorManager");
-				selectorManagerField.setAccessible(true);
-				Object selectorManager = selectorManagerField.get(httpClient);
-
-				if (selectorManager != null) {
-					java.lang.reflect.Method shutdownMethod = selectorManager.getClass().getMethod("shutdown");
-					shutdownMethod.invoke(selectorManager);
-					logger.debug("HttpClient SelectorManager shutdown completed");
+				Method closeMethod = httpClient.getClass().getMethod("close");
+				closeMethod.invoke(httpClient);
+				logger.debug("Successfully used JDK 21+ close() method to close HttpClient");
+				return;
+			}
+			catch (NoSuchMethodException e) {
+				logger.debug("JDK 21+ close() method not available, falling back to internal reflection");
+			}
+			// This prevents the accumulation of HttpClient-xxx-SelectorManager threads
+			try {
+				java.lang.reflect.Field implField = httpClient.getClass().getDeclaredField("impl");
+				implField.setAccessible(true);
+				Object implObj = implField.get(httpClient);
+				java.lang.reflect.Field selmgrField = implObj.getClass().getDeclaredField("selmgr");
+				selmgrField.setAccessible(true);
+				Object selmgrObj = selmgrField.get(implObj);
+
+				if (selmgrObj != null) {
+					Method shutDownMethod = selmgrObj.getClass().getDeclaredMethod("shutdown");
+					shutDownMethod.setAccessible(true);
+					shutDownMethod.invoke(selmgrObj);
+					logger.debug("HttpClient SelectorManager shutdown completed via reflection");
 				}
 			}
 			catch (NoSuchFieldException | NoSuchMethodException e) {
-				// Field/method might not exist in different JDK versions, continue with
-				// other cleanup
-				logger.debug("SelectorManager field/method not found, skipping: {}", e.getMessage());
+				// Field/method structure might differ across JDK versions
+				logger.debug("SelectorManager field/method not found, skipping internal cleanup: {}", e.getMessage());
 			}
 
 		}