Fix HttpClient resource leak with ExecutorService-based cleanup

Implement automatic ExecutorService shutdown for internal HttpClient and add support for external HttpClient injection to prevent thread accumulation. Related to #610

Author: TsengX

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

@@ -5,15 +5,16 @@
 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;
 import java.net.http.HttpResponse;
 import java.time.Duration;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 import java.util.function.Function;
@@ -38,7 +39,6 @@
 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
@@ -146,6 +146,7 @@ public class HttpClientSseClientTransport implements McpClientTransport {
 		Assert.notNull(httpClient, "httpClient must not be null");
 		Assert.notNull(requestBuilder, "requestBuilder must not be null");
 		Assert.notNull(httpRequestCustomizer, "httpRequestCustomizer must not be null");
+		Assert.notNull(onCloseClient, "onCloseClient must not be null");
 		this.baseUri = URI.create(baseUri);
 		this.sseEndpoint = sseEndpoint;
 		this.jsonMapper = jsonMapper;
@@ -178,8 +179,6 @@ public static class Builder {
 
 		private String sseEndpoint = DEFAULT_SSE_ENDPOINT;
 
-		private HttpClient.Builder clientBuilder = HttpClient.newBuilder().version(HttpClient.Version.HTTP_1_1);
-
 		private HttpClient externalHttpClient;
 
 		private McpJsonMapper jsonMapper;
@@ -190,7 +189,8 @@ public static class Builder {
 
 		private Duration connectTimeout = Duration.ofSeconds(10);
 
-		private Consumer<HttpClient> onCloseClient;
+		private Consumer<HttpClient> onCloseClient = (HttpClient client) -> {
+		};
 
 		/**
 		 * Creates a new builder instance.
@@ -235,13 +235,17 @@ public Builder sseEndpoint(String sseEndpoint) {
 		}
 
 		/**
-		 * 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
-		 * during graceful shutdown, leaving resource management to the caller.
+		 * Provides 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 during graceful shutdown, leaving resource management to the caller.
+		 * <p>
+		 * Use this method when you want to share a single HttpClient instance across
+		 * multiple transports or when you need fine-grained control over HttpClient
+		 * lifecycle.
 		 * @param httpClient the HttpClient instance to use
 		 * @return this builder
 		 */
-		public Builder httpClient(HttpClient httpClient) {
+		public Builder withExternalHttpClient(HttpClient httpClient) {
 			Assert.notNull(httpClient, "httpClient must not be null");
 			this.externalHttpClient = httpClient;
 			return this;
@@ -317,11 +321,15 @@ public Builder asyncHttpRequestCustomizer(McpAsyncHttpClientRequestCustomizer as
 
 		/**
 		 * Sets a custom consumer to handle HttpClient closure when the transport is
-		 * closed.
+		 * closed. This allows for custom cleanup logic beyond the default behavior.
+		 * <p>
+		 * Note: This is typically used for advanced use cases. The default behavior
+		 * (shutting down the internal ExecutorService) is sufficient for most scenarios.
 		 * @param onCloseClient the consumer to handle HttpClient closure
 		 * @return this builder
 		 */
-		public Builder onCloseClient(Consumer<HttpClient> onCloseClient) {
+		public Builder onHttpClientClose(Consumer<HttpClient> onCloseClient) {
+			Assert.notNull(onCloseClient, "onCloseClient must not be null");
 			this.onCloseClient = onCloseClient;
 			return this;
 		}
@@ -337,13 +345,29 @@ public HttpClientSseClientTransport build() {
 			if (externalHttpClient != null) {
 				// Use external HttpClient, use custom close handler or no-op
 				httpClient = externalHttpClient;
-				closeHandler = onCloseClient; // null means no cleanup
+				closeHandler = onCloseClient;
 			}
 			else {
-				// Create new HttpClient, use custom close handler or default cleanup
-				httpClient = this.clientBuilder.connectTimeout(this.connectTimeout).build();
-				closeHandler = onCloseClient != null ? onCloseClient
-						: HttpClientSseClientTransport::closeHttpClientResourcesStatic;
+				// Create internal HttpClient with custom ExecutorService
+				// Create a custom ExecutorService with meaningful thread names
+				ExecutorService internalExecutor = Executors.newCachedThreadPool(runnable -> {
+					Thread thread = new Thread(runnable);
+					thread.setName("MCP-HttpClient-" + thread.getId());
+					thread.setDaemon(true);
+					return thread;
+				});
+
+				httpClient = HttpClient.newBuilder()
+					.version(HttpClient.Version.HTTP_1_1)
+					.connectTimeout(this.connectTimeout)
+					.executor(internalExecutor)
+					.build();
+
+				// Combine default cleanup (shutdown executor) with custom handler if
+				// provided
+				closeHandler = (client) -> shutdownHttpClientExecutor(internalExecutor);
+				closeHandler = closeHandler.andThen(onCloseClient);
+
 			}
 
 			return new HttpClientSseClientTransport(httpClient, requestBuilder, baseUri, sseEndpoint,
@@ -519,53 +543,43 @@ public Mono<Void> closeGracefully() {
 	}
 
 	/**
-	 * Static method to close HttpClient resources using reflection.
+	 * Closes HttpClient resources by shutting down its associated ExecutorService. This
+	 * allows the GC to reclaim HttpClient-related threads (including SelectorManager) on
+	 * the next garbage collection cycle.
+	 * <p>
+	 * This approach avoids using reflection, Unsafe, or Java 21+ specific APIs, making it
+	 * compatible with Java 17+.
+	 * @param executor the ExecutorService to shutdown
 	 */
-	private static void closeHttpClientResourcesStatic(HttpClient httpClient) {
+	private static void shutdownHttpClientExecutor(ExecutorService executor) {
+		if (executor == null) {
+			return;
+		}
+
 		try {
-			// 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);
+			logger.debug("Shutting down HttpClient ExecutorService");
+			executor.shutdown();
 
-			try {
-				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");
+			// Wait for graceful shutdown
+			if (!executor.awaitTermination(5, TimeUnit.SECONDS)) {
+				logger.debug("ExecutorService did not terminate in time, forcing shutdown");
+				executor.shutdownNow();
+
+				// Wait a bit more after forced shutdown
+				if (!executor.awaitTermination(2, TimeUnit.SECONDS)) {
+					logger.warn("ExecutorService did not terminate even after shutdownNow()");
 				}
 			}
-			catch (NoSuchFieldException | NoSuchMethodException e) {
-				// Field/method structure might differ across JDK versions
-				logger.debug("SelectorManager field/method not found, skipping internal cleanup: {}", e.getMessage());
-			}
 
+			logger.debug("HttpClient ExecutorService shutdown completed");
+		}
+		catch (InterruptedException e) {
+			logger.warn("Interrupted while shutting down HttpClient ExecutorService");
+			executor.shutdownNow();
+			Thread.currentThread().interrupt();
 		}
 		catch (Exception e) {
-			logger.warn("Failed to close HttpClient resources cleanly: {}", e.getMessage());
+			logger.warn("Failed to shutdown HttpClient ExecutorService cleanly: {}", e.getMessage());
 		}
 	}