Fix AsyncHttpEncryptionAwareConduitFactory thread leak on WinRMService close

src/main/java/org/metricshub/winrm/service/WinRMService.java

@@ -56,6 +56,7 @@
 import org.apache.cxf.Bus.BusState;
 import org.apache.cxf.BusFactory;
 import org.apache.cxf.endpoint.Client;
+import org.apache.cxf.transport.http.HTTPConduitFactory;
 import org.apache.cxf.transport.http.asyncclient.AsyncHTTPConduit;
 import org.apache.cxf.transport.http.asyncclient.AsyncHTTPConduitFactory;
 import org.apache.cxf.transport.http.asyncclient.AsyncHTTPConduitFactory.UseAsyncPolicy;
@@ -302,7 +303,7 @@
 		} catch (final RuntimeException e) {
 			if (e.getCause() != null) {
 				final String message = e.getMessage() != null
 					? String.format(
 						"%s\n%s: %s",
 						e.getMessage(),
 						e.getCause().getClass().getSimpleName(),
@@ -352,10 +353,12 @@
 			}
 
 			if (cmdClient != null) {
+				shutdownConduitFactory(cmdClient);
 				cmdClient.destroy();
 			}
 
 			if (wqlClient != null) {
+				shutdownConduitFactory(wqlClient);
 				wqlClient.destroy();
 			}
 
@@ -365,6 +368,20 @@
 		}
 	}
 
+	/**
+	 * Retrieves the {@link AsyncHTTPConduitFactory} registered on the given client's endpoint and calls
+	 * {@link AsyncHTTPConduitFactory#shutdown()} on it to stop any background threads (e.g. the idle-connection
+	 * reaper thread). This must be done before destroying the client to prevent thread leaks.
+	 *
+	 * @param client the CXF {@link Client} whose conduit factory should be shut down
+	 */
+	private void shutdownConduitFactory(final Client client) {
+		final Object factory = client.getEndpoint().getEndpointInfo().getProperty(HTTPConduitFactory.class.getName());
+		if (factory instanceof AsyncHTTPConduitFactory) {
+			((AsyncHTTPConduitFactory) factory).shutdown();
+		}
+	}
+
 	@Override
 	public WindowsRemoteCommandResult executeCommand(
 		final String command,
@@ -629,7 +646,7 @@
 				}
 			}
 			throw soapFault;
 		} catch (final NullPointerException e) {
 			throw soapFault;
 		}
 	}
@@ -688,30 +705,30 @@
 				final JAXBElement<?> jaxbElement = (JAXBElement<?>) object;
 
 				if (WSEN_ITEMS_QNAME.equals(jaxbElement.getName()) || WSMAN_ITEMS_QNAME.equals(jaxbElement.getName())) {
 					if (jaxbElement.isNil()) {
 						// No items
 					} else if (jaxbElement.getValue() instanceof AnyListType) {
 						// some items
 						final AnyListType itemList = (AnyListType) jaxbElement.getValue();
 						for (final Object item : itemList.getAny()) {
 							final Node node = toNode(item)
 								.orElseThrow(() ->
 									new WinRMException(
 										"Unsupported element of type %s in EnumerateResponse: %s",
 										object.getClass(),
 										object
 									)
 								);
 
 							items.add(node);
 						}
 					} else {
 						throw new WinRMException(
 							"Unsupported value in EnumerateResponse Items: %s of type: %s",
 							jaxbElement.getValue(),
 							jaxbElement.getValue().getClass()
 						);
 					}
 				} else if (
 					WSEN_END_OF_SEQUENCE_QNAME.equals(jaxbElement.getName()) ||
 					WSMAN_END_OF_SEQUENCE_QNAME.equals(jaxbElement.getName())
@@ -790,19 +807,19 @@
 
 				final MixedDataType mixed = (MixedDataType) nestedElement.getValue();
 				for (final Object nestedItem : mixed.getContent()) {
 					if (nestedItem instanceof String) {
 						// Skip over whitespace
 					} else if (nestedItem instanceof Node) {
 						// Node's can't belong to two different documents, so we need to import it first
 						final Node nestedNode = document.importNode((Node) nestedItem, true);
 						rootElement.appendChild(nestedNode);
 					} else {
 						throw new WinRMException(
 							"Unsupported element of type %s in XmlFragment: %s",
 							nestedItem.getClass(),
 							nestedItem
 						);
 					}
 				}
 				return Optional.of(rootElement);
 			}
@@ -825,7 +842,7 @@
 	public String getContextIdFrom(final EnumerationContextType context) throws WinRMException {
 		// The content of the EnumerationContext should contain a single string, the context id
 		if (context == null || context.getContent() == null) {
 			throw new WinRMException("EnumerationContext %s has no content.", context);
 		}
 
 		if (context.getContent().isEmpty()) {

src/main/java/org/metricshub/winrm/service/client/WinRMInvocationHandler.java

@@ -50,6 +50,7 @@
 import org.apache.cxf.frontend.ClientProxy;
 import org.apache.cxf.jaxws.JaxWsProxyFactoryBean;
 import org.apache.cxf.message.Message;
+import org.apache.cxf.service.model.EndpointInfo;
 import org.apache.cxf.service.model.ServiceInfo;
 import org.apache.cxf.transport.http.HTTPConduitFactory;
 import org.apache.cxf.transport.http.asyncclient.AsyncHTTPConduit;
@@ -179,7 +180,7 @@
 	}
 
 	public Client getClient() {
 		return wsClient;
 	}
 
 	@Override
@@ -403,10 +404,18 @@
 		client.getOutInterceptors().add(new SignAndEncryptOutInterceptor());
 
 		// this is different to endpoint properties
-		client
-			.getEndpoint()
-			.getEndpointInfo()
-			.setProperty(HTTPConduitFactory.class.getName(), new AsyncHttpEncryptionAwareConduitFactory());
+		// Register the conduit factory only once: on authentication retries this method is re-invoked on the
+		// same client, whose cached conduit keeps using the factory it was created with. Replacing the property
+		// would orphan factory instances, and shutting down the in-use factory would silently downgrade the
+		// conduit to the synchronous transport (AsyncHTTPConduit.setupConnection checks factory.isShutdown()).
+		// Reusing the factory also guarantees WinRMService.close() shuts down the instance that owns the
+		// background threads.
+		final EndpointInfo endpointInfo = client.getEndpoint().getEndpointInfo();
+		if (
+			!(endpointInfo.getProperty(HTTPConduitFactory.class.getName()) instanceof AsyncHttpEncryptionAwareConduitFactory)
+		) {
+			endpointInfo.setProperty(HTTPConduitFactory.class.getName(), new AsyncHttpEncryptionAwareConduitFactory());
+		}
 
 		final ServiceInfo serviceInfo = client.getEndpoint().getEndpointInfo().getService();
 		serviceInfo.setProperty("soap.force.doclit.bare", true);
@@ -541,12 +550,12 @@
 				return false;
 			}
 			final CredentialsMapKey other = (CredentialsMapKey) obj;
 			return (
 				authentication == other.authentication &&
 				Objects.equals(canonizedRawUsername, other.canonizedRawUsername) &&
 				Arrays.equals(password, other.password) &&
 				Objects.equals(ticketCache, other.ticketCache)
 			);
 		}
 	}
 }

src/test/java/org/metricshub/winrm/service/WinRMServiceTest.java

@@ -31,6 +31,11 @@
 import java.nio.file.Paths;
 import java.util.List;
 import org.apache.cxf.Bus;
+import org.apache.cxf.endpoint.Client;
+import org.apache.cxf.endpoint.Endpoint;
+import org.apache.cxf.service.model.EndpointInfo;
+import org.apache.cxf.transport.http.HTTPConduitFactory;
+import org.apache.cxf.transport.http.asyncclient.AsyncHTTPConduitFactory;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
@@ -268,4 +273,76 @@ void testExecuteWql() throws Exception {
 			assertEquals(emptyList(), winRMService.executeWql(wqlQuery, timeout));
 		}
 	}
+
+	@Test
+	void testCloseShutdownsConduitFactories() throws Exception {
+		// Use a unique endpoint to avoid interference with other test stubs
+		final WinRMEndpoint endpointForFactoryTest = new WinRMEndpoint(
+			null,
+			"factory-test-host",
+			null,
+			"user",
+			"pwd".toCharArray(),
+			null
+		);
+
+		// Set up a mock factory that will be stored in the client's endpoint info
+		final AsyncHTTPConduitFactory mockFactory = mock(AsyncHTTPConduitFactory.class);
+
+		// Set up mock endpoint info containing the factory
+		final EndpointInfo mockEndpointInfo = mock(EndpointInfo.class);
+		doReturn(mockFactory).when(mockEndpointInfo).getProperty(HTTPConduitFactory.class.getName());
+
+		// Set up mock endpoint
+		final Endpoint mockEndpoint = mock(Endpoint.class);
+		doReturn(mockEndpointInfo).when(mockEndpoint).getEndpointInfo();
+
+		// Set up mock client
+		final Client mockClient = mock(Client.class);
+		doReturn(mockEndpoint).when(mockClient).getEndpoint();
+
+		// Set up mock invocation handlers that expose the configured client
+		final WinRMInvocationHandler cmdHandler = mock(WinRMInvocationHandler.class);
+		doReturn(mockClient).when(cmdHandler).getClient();
+
+		final WinRMInvocationHandler wqlHandler = mock(WinRMInvocationHandler.class);
+		doReturn(mockClient).when(wqlHandler).getClient();
+
+		// Override the default stubs for this specific endpoint (registered later, so they take precedence)
+		MOCKED_WIN_RM_SERVICE
+			.when(() ->
+				WinRMService.createWinRMInvocationHandlerInstance(
+					eq(endpointForFactoryTest),
+					any(Bus.class),
+					anyLong(),
+					isNull(),
+					isNull(),
+					anyList()
+				)
+			)
+			.thenReturn(cmdHandler);
+
+		MOCKED_WIN_RM_SERVICE
+			.when(() ->
+				WinRMService.createWinRMInvocationHandlerInstance(
+					eq(endpointForFactoryTest),
+					any(Bus.class),
+					anyLong(),
+					anyString(),
+					isNull(),
+					anyList()
+				)
+			)
+			.thenReturn(wqlHandler);
+
+		// Create the service and immediately close it
+		// (null ticketCache and authentications: the class-level createInstance stub calling the real
+		// method only matches isNull() for both)
+		final WinRMService winRMService = createInstance(endpointForFactoryTest, 30000L, null, null);
+		assertNotNull(winRMService);
+		winRMService.close();
+
+		// Verify that shutdown() was called on the factory for both the cmd and wql clients
+		verify(mockFactory, times(2)).shutdown();
+	}
 }