(crawler) Add logging to get more info about connection leak

code/processes/crawling-process/java/nu/marginalia/crawl/fetcher/HttpFetcherImpl.java

@@ -35,6 +35,7 @@ import org.apache.hc.core5.http.io.entity.EntityUtils;
 import org.apache.hc.core5.http.io.support.ClassicRequestBuilder;
 import org.apache.hc.core5.http.message.MessageSupport;
 import org.apache.hc.core5.http.protocol.HttpContext;
+import org.apache.hc.core5.pool.PoolStats;
 import org.apache.hc.core5.util.TimeValue;
 import org.apache.hc.core5.util.Timeout;
 import org.jsoup.Jsoup;
@@ -77,14 +78,20 @@ public class HttpFetcherImpl implements HttpFetcher, HttpRequestRetryStrategy {
     }
 
     private final CloseableHttpClient client;
+    private PoolingHttpClientConnectionManager connectionManager;
+
+    public PoolStats getPoolStats() {
+        return connectionManager.getTotalStats();
+    }
 
     private CloseableHttpClient createClient() throws NoSuchAlgorithmException {
         final ConnectionConfig connectionConfig = ConnectionConfig.custom()
                 .setSocketTimeout(10, TimeUnit.SECONDS)
                 .setConnectTimeout(30, TimeUnit.SECONDS)
+                .setValidateAfterInactivity(TimeValue.ofSeconds(5))
                 .build();
 
-        final PoolingHttpClientConnectionManager connectionManager = PoolingHttpClientConnectionManagerBuilder.create()
+        connectionManager = PoolingHttpClientConnectionManagerBuilder.create()
                 .setMaxConnPerRoute(2)
                 .setMaxConnTotal(5000)
                 .setDefaultConnectionConfig(connectionConfig)
@@ -97,6 +104,18 @@ public class HttpFetcherImpl implements HttpFetcher, HttpRequestRetryStrategy {
                 .build()
         );
 
+        Thread.ofPlatform().daemon(true).start(() -> {
+            try {
+                for (;;) {
+                    TimeUnit.SECONDS.sleep(15);
+                    logger.info("Connection pool stats: {}", connectionManager.getTotalStats());
+                }
+            }
+            catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+            }
+        });
+
         final RequestConfig defaultRequestConfig = RequestConfig.custom()
                 .setCookieSpec(StandardCookieSpec.RELAXED)
                 .setResponseTimeout(10, TimeUnit.SECONDS)

code/processes/crawling-process/java/nu/marginalia/crawl/fetcher/warc/WarcInputBuffer.java

@@ -57,10 +57,11 @@ public abstract class WarcInputBuffer implements AutoCloseable {
             return new ErrorBuffer();
         }
 
-        InputStream is = entity.getContent();
-        long length = entity.getContentLength();
-
+        InputStream is = null;
         try {
+            is = entity.getContent();
+            long length = entity.getContentLength();
+
             if (length > 0 && length < 8192) {
                 // If the content is small and not compressed, we can just read it into memory
                 return new MemoryBuffer(response.getHeaders(), request, timeLimit, is, (int) length);

code/processes/crawling-process/java/nu/marginalia/crawl/fetcher/warc/WarcRecorder.java

@@ -237,7 +237,6 @@ public class WarcRecorder implements AutoCloseable {
                             dataStart,
                             responseDataBuffer.length() - dataStart);
                 } catch (Exception ex) {
-                    ex.printStackTrace();
                     flagAsError(new EdgeUrl(requestUri), ex); // write a WARC record to indicate the error
                     logger.warn("Failed to fetch URL {}:  {}", requestUri, ex.getMessage());
                     return new HttpFetchResult.ResultException(ex);
@@ -250,7 +249,6 @@ public class WarcRecorder implements AutoCloseable {
             flagAsTimeout(new EdgeUrl(requestUri)); // write a WARC record to indicate the timeout
             return new HttpFetchResult.ResultException(ex);
         } catch (IOException ex) {
-            ex.printStackTrace();
             flagAsError(new EdgeUrl(requestUri), ex); // write a WARC record to indicate the error
             logger.warn("Failed to fetch URL {}:  {}", requestUri, ex.getMessage());
             return new HttpFetchResult.ResultException(ex);

code/processes/crawling-process/test/nu/marginalia/crawl/fetcher/HttpFetcherImplContentTypeProbeTest.java

@@ -11,6 +11,8 @@ import org.junit.jupiter.api.*;
 import java.io.IOException;
 import java.net.URISyntaxException;
 
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
 @Tag("slow")
 class HttpFetcherImplContentTypeProbeTest {
 
@@ -85,6 +87,10 @@ class HttpFetcherImplContentTypeProbeTest {
 
     @AfterEach
     public void tearDown() throws IOException {
+        var stats = fetcher.getPoolStats();
+        assertEquals(0, stats.getLeased());
+        assertEquals(0, stats.getPending());
+
         fetcher.close();
     }
 

code/processes/crawling-process/test/nu/marginalia/crawl/fetcher/HttpFetcherImplDomainProbeTest.java

@@ -12,6 +12,8 @@ import org.junit.jupiter.api.*;
 import java.io.IOException;
 import java.net.URISyntaxException;
 
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
 @Tag("slow")
 class HttpFetcherImplDomainProbeTest {
 
@@ -47,6 +49,10 @@ class HttpFetcherImplDomainProbeTest {
 
     @AfterEach
     public void tearDown() throws IOException {
+        var stats = fetcher.getPoolStats();
+        assertEquals(0, stats.getLeased());
+        assertEquals(0, stats.getPending());
+
         fetcher.close();
     }
 

code/processes/crawling-process/test/nu/marginalia/crawl/fetcher/HttpFetcherImplFetchTest.java

@@ -139,12 +139,23 @@ class HttpFetcherImplFetchTest {
 
     @AfterEach
     public void tearDown() throws IOException {
+        var stats = fetcher.getPoolStats();
+        assertEquals(0, stats.getLeased());
+        assertEquals(0, stats.getPending());
+
+        System.out.println(stats);
+
         fetcher.close();
         warcRecorder.close();
         Files.deleteIfExists(warcFile);
     }
 
 
+    @Test
+    public void testFoo() {
+        fetcher.fetchSitemapUrls("https://www.marginalia.nu/sitemap.xml", new CrawlDelayTimer(100));
+    }
+
     @Test
     public void testOk_NoProbe() throws IOException {
         var result = fetcher.fetchContent(okUrl, warcRecorder, new CrawlDelayTimer(1000), ContentTags.empty(), HttpFetcher.ProbeType.DISABLED);

code/processes/crawling-process/test/nu/marginalia/crawl/retreival/fetcher/WarcRecorderFakeServerTest.java

@@ -94,6 +94,7 @@ class WarcRecorderFakeServerTest {
 
     @AfterEach
     public void tearDown() throws Exception {
+
         client.close();
         Files.delete(fileNameWarc);
     }