TIKA-5064 Close HTTP clients when a fetcher is replaced.

tika-pipes/tika-pipes-plugins/tika-pipes-atlassian-jwt/src/main/java/org/apache/tika/pipes/fetcher/atlassianjwt/AtlassianJwtFetcher.java

@@ -16,6 +16,7 @@
  */
 package org.apache.tika.pipes.fetcher.atlassianjwt;
 
+import java.io.Closeable;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
@@ -69,7 +70,7 @@
 import org.apache.tika.plugins.ExtensionConfig;
 import org.apache.tika.utils.StringUtils;
 
-public class AtlassianJwtFetcher extends AbstractTikaExtension implements Fetcher {
+public class AtlassianJwtFetcher extends AbstractTikaExtension implements Fetcher, Closeable {
 
     private static final Logger LOG = LoggerFactory.getLogger(AtlassianJwtFetcher.class);
 
@@ -346,4 +347,14 @@ private String responseToString(HttpResponse response) {
             EntityUtils.consumeQuietly(response.getEntity());
         }
     }
+
+    /**
+     * Each client owns a connection pool and a "Connection evictor" thread; the fetcher manager
+     * calls this when a runtime save or delete retires this instance.
+     */
+    @Override
+    public void close() throws IOException {
+        IOUtils.close(httpClient instanceof Closeable c ? c : null,
+                noCompressHttpClient instanceof Closeable nc ? nc : null);
+    }
 }

tika-pipes/tika-pipes-plugins/tika-pipes-http/src/main/java/org/apache/tika/pipes/fetcher/http/HttpFetcher.java

@@ -17,6 +17,7 @@
 package org.apache.tika.pipes.fetcher.http;
 
 
+import java.io.Closeable;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
@@ -85,7 +86,7 @@
 /**
  * Based on Apache httpclient
  */
-public class HttpFetcher extends AbstractTikaExtension implements Fetcher, RangeFetcher {
+public class HttpFetcher extends AbstractTikaExtension implements Fetcher, RangeFetcher, Closeable {
 
     public static HttpFetcher build(ExtensionConfig pluginConfig) throws TikaConfigException, IOException {
         HttpFetcherConfig httpFetcherConfig = HttpFetcherConfig.load(pluginConfig.json());
@@ -576,4 +577,14 @@ public void setJwtGenerator(JwtGenerator jwtGenerator) {
     public JwtGenerator getJwtGenerator() {
         return jwtGenerator;
     }
+
+    /**
+     * Each client owns a connection pool and a "Connection evictor" thread; the fetcher manager
+     * calls this when a runtime save or delete retires this instance.
+     */
+    @Override
+    public void close() throws IOException {
+        IOUtils.close(httpClient instanceof Closeable c ? c : null,
+                noCompressHttpClient instanceof Closeable nc ? nc : null);
+    }
 }

tika-pipes/tika-pipes-plugins/tika-pipes-http/src/test/java/org/apache/tika/pipes/fetcher/http/HttpFetcherTest.java

@@ -39,6 +39,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.stream.Collectors;
 import java.util.zip.GZIPInputStream;
 
@@ -378,6 +379,31 @@ public void testVerifySslReachesClientFactory() throws Exception {
         assertFalse(factory.isVerifySsl(), "the explicit opt-out must reach the factory");
     }
 
+    /**
+     * A runtime save retires the old fetcher by closing it. Unclosed, each save leaked one
+     * "Connection evictor" thread per client until the process could not start threads.
+     */
+    @Test
+    public void testCloseStopsEvictorThreads() throws Exception {
+        Set<Thread> before = evictorThreads();
+        HttpFetcher fetcher = (HttpFetcher) new HttpFetcherFactory().buildExtension(
+                new ExtensionConfig("id", "http-fetcher", "{}"));
+        Set<Thread> started = evictorThreads();
+        started.removeAll(before);
+        assertEquals(2, started.size(), "one evictor per client");
+
+        fetcher.close();
+        for (Thread t : started) {
+            assertFalse(t.isAlive(), "close must stop the evictor");
+        }
+    }
+
+    private static Set<Thread> evictorThreads() {
+        return Thread.getAllStackTraces().keySet().stream()
+                .filter(t -> "Connection evictor".equals(t.getName()))
+                .collect(Collectors.toSet());
+    }
+
     private void mockClientResponse(final HttpResponse response) throws Exception {
         httpFetcher = (HttpFetcher) getFetcherManager("tika-config-http.json").getFetcher("http-fetcher-1").get();