SOLR-17707: release the content-writing thread when an async request fails

changelog/unreleased/SOLR-17707-httpjdk-async-thread-leak.yml

@@ -0,0 +1,11 @@
+# See https://github.com/apache/solr/blob/main/dev-docs/changelog.adoc
+
+title: >
+  Fixed HttpJdkSolrClient leaking an executor thread when an async request's connection failed while
+  its body was still being written; enough such failures could exhaust the client's thread pool.
+type: fixed
+authors:
+  - name: Serhiy Bzhezytskyy
+links:
+  - name: SOLR-17707
+    url: https://issues.apache.org/jira/browse/SOLR-17707

solr/solrj/src/java/org/apache/solr/client/solrj/impl/HttpJdkSolrClient.java

@@ -144,8 +144,10 @@ protected HttpJdkSolrClient(String serverBaseUrl, HttpJdkSolrClient.Builder buil
   protected CompletableFuture<HttpResponse<InputStream>> requestInputStreamAsync(
       String baseUrl, final SolrRequest<?> solrRequest, String collection) {
     try {
-      HttpRequest httpRequest = prepareRequest(baseUrl, solrRequest, collection).reqb.build();
-      return httpClient.sendAsync(httpRequest, HttpResponse.BodyHandlers.ofInputStream());
+      PreparedRequest pReq = prepareRequest(baseUrl, solrRequest, collection);
+      return httpClient
+          .sendAsync(pReq.reqb.build(), HttpResponse.BodyHandlers.ofInputStream())
+          .whenComplete((httpResponse, throwable) -> releaseContentWriting(pReq));
     } catch (Exception e) {
       CompletableFuture<HttpResponse<InputStream>> cf = new CompletableFuture<>();
       cf.completeExceptionally(e);
@@ -160,6 +162,7 @@ public CompletableFuture<NamedList<Object>> requestAsync(
       PreparedRequest pReq = prepareRequest(null, solrRequest, collection);
       return httpClient
           .sendAsync(pReq.reqb.build(), HttpResponse.BodyHandlers.ofInputStream())
+          .whenComplete((httpResponse, throwable) -> releaseContentWriting(pReq))
           .thenApply(
               httpResponse -> {
                 try {
@@ -176,6 +179,21 @@ public CompletableFuture<NamedList<Object>> requestAsync(
     }
   }
 
+  private void releaseContentWriting(PreparedRequest pReq) {
+    if (pReq.contentWritingFuture != null) {
+      pReq.contentWritingFuture.cancel(true);
+    }
+    // Closing the sink is what unblocks a writer already stuck in the pipe; cancel() alone does
+    // not.
+    if (pReq.contentWritingSink != null) {
+      try {
+        pReq.contentWritingSink.close();
+      } catch (IOException e) {
+        log.warn("Could not close content-writing pipe", e);
+      }
+    }
+  }
+
   @Override
   public NamedList<Object> requestWithBaseUrl(
       String baseUrl, SolrRequest<?> solrRequest, String collection)
@@ -271,7 +289,7 @@ private PreparedRequest prepareGet(
     reqb.GET();
     decorateRequest(reqb, solrRequest);
     reqb.uri(new URI(url + queryParams.toQueryString()));
-    return new PreparedRequest(reqb, null);
+    return new PreparedRequest(reqb, null, null);
   }
 
   private PreparedRequest preparePutOrPost(
@@ -303,14 +321,16 @@ private PreparedRequest preparePutOrPost(
 
     HttpRequest.BodyPublisher bodyPublisher;
     Future<?> contentWritingFuture = null;
+    PipedInputStream contentWritingSink = null;
     if (contentWriter != null) {
       boolean success = maybeTryHeadRequest(url);
       if (!success) {
         reqb.version(HttpClient.Version.HTTP_1_1);
       }
 
       final PipedOutputStream source = new PipedOutputStream();
-      final PipedInputStream sink = new PipedInputStream(source);
+      contentWritingSink = new PipedInputStream(source);
+      final PipedInputStream sink = contentWritingSink;
       bodyPublisher = HttpRequest.BodyPublishers.ofInputStream(() -> sink);
 
       contentWritingFuture =
@@ -352,20 +372,25 @@ private PreparedRequest preparePutOrPost(
     URI uriWithQueryParams = new URI(url + queryParams.toQueryString());
     reqb.uri(uriWithQueryParams);
 
-    return new PreparedRequest(reqb, contentWritingFuture);
+    return new PreparedRequest(reqb, contentWritingFuture, contentWritingSink);
   }
 
   protected static class PreparedRequest {
     Future<?> contentWritingFuture;
+    PipedInputStream contentWritingSink;
     HttpRequest.Builder reqb;
 
     ResponseParser parserToUse;
 
     String url;
 
-    PreparedRequest(HttpRequest.Builder reqb, Future<?> contentWritingFuture) {
+    PreparedRequest(
+        HttpRequest.Builder reqb,
+        Future<?> contentWritingFuture,
+        PipedInputStream contentWritingSink) {
       this.reqb = reqb;
       this.contentWritingFuture = contentWritingFuture;
+      this.contentWritingSink = contentWritingSink;
     }
   }
 

solr/solrj/src/test/org/apache/solr/client/solrj/impl/HttpJdkSolrClientTest.java

@@ -20,15 +20,23 @@
 import java.io.IOException;
 import java.net.CookieHandler;
 import java.net.CookieManager;
+import java.net.ServerSocket;
+import java.net.Socket;
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.net.http.HttpClient;
+import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;
 import org.apache.lucene.util.NamedThreadFactory;
 import org.apache.solr.client.api.util.SolrVersion;
@@ -38,6 +46,7 @@
 import org.apache.solr.client.solrj.request.JavaBinRequestWriter;
 import org.apache.solr.client.solrj.request.QueryRequest;
 import org.apache.solr.client.solrj.request.SolrQuery;
+import org.apache.solr.client.solrj.request.UpdateRequest;
 import org.apache.solr.client.solrj.request.XMLRequestWriter;
 import org.apache.solr.client.solrj.response.JavaBinResponseParser;
 import org.apache.solr.client.solrj.response.ResponseParser;
@@ -132,6 +141,132 @@ public void testDeleteXml() throws Exception {
     }
   }
 
+  /**
+   * A large content-writing request whose connection drops while the body is still being written
+   * must not leave the body-writing thread blocked forever in {@code
+   * PipedInputStream.awaitSpace()}. The server accepts the connection but never reads the body, so
+   * the pipe buffer fills and the writer blocks; when the connection is then reset the writer must
+   * be released (SOLR-17707).
+   */
+  @Test
+  public void testStuckContentWritingThreadIsReleasedOnFailure() throws Exception {
+    // A body far larger than the PipedInputStream buffer and the socket buffers, so the writer is
+    // still blocked in awaitSpace() when the connection drops.
+    StringBuilder big = new StringBuilder();
+    while (big.length() < 8 * 1024 * 1024) {
+      big.append("id_").append(big.length()).append(' ');
+    }
+    UpdateRequest req = new UpdateRequest();
+    req.deleteByQuery(big.toString());
+
+    ExecutorService executor =
+        ExecutorUtil.newMDCAwareCachedThreadPool(new NamedThreadFactory("solr-17707-writer"));
+    try (StallThenResetServer server = new StallThenResetServer();
+        HttpJdkSolrClient client =
+            builder(server.baseUrl()).useHttp1_1(true).withExecutor(executor).build()) {
+
+      CompletableFuture<?> cf = client.requestAsync(req, null);
+      server.awaitConnected(30, TimeUnit.SECONDS);
+      Thread.sleep(1000); // let the writer fill the pipe and block in awaitSpace()
+      server.resetAll(); // drop the connection under the in-flight write
+
+      // The request settles (normally exceptionally, from the dropped connection); either way the
+      // point of the test is the thread check below, so just wait for it to complete.
+      cf.handle((r, t) -> null).get(30, TimeUnit.SECONDS);
+
+      // The writer thread must not remain blocked in the pipe after the request has failed.
+      assertTrue(
+          "content-writing thread leaked, still blocked after failure",
+          waitForNoBlockedWriter(15, TimeUnit.SECONDS));
+    } finally {
+      ExecutorUtil.shutdownAndAwaitTermination(executor);
+    }
+  }
+
+  private static boolean waitForNoBlockedWriter(long timeout, TimeUnit unit)
+      throws InterruptedException {
+    long deadline = System.nanoTime() + unit.toNanos(timeout);
+    while (System.nanoTime() < deadline) {
+      if (!hasBlockedWriterThread()) {
+        return true;
+      }
+      Thread.sleep(100);
+    }
+    return !hasBlockedWriterThread();
+  }
+
+  private static boolean hasBlockedWriterThread() {
+    Thread[] threads = new Thread[Thread.activeCount() * 2];
+    int n = Thread.enumerate(threads);
+    for (int i = 0; i < n; i++) {
+      Thread t = threads[i];
+      if (t != null && t.getName().startsWith("solr-17707-writer")) {
+        for (StackTraceElement el : t.getStackTrace()) {
+          if ("awaitSpace".equals(el.getMethodName())) {
+            return true;
+          }
+        }
+      }
+    }
+    return false;
+  }
+
+  /** A TCP server that accepts a connection, never reads the body, then resets on demand. */
+  private static class StallThenResetServer implements AutoCloseable {
+    private final ServerSocket serverSocket;
+    private final List<Socket> accepted = Collections.synchronizedList(new ArrayList<>());
+    private final CountDownLatch connected = new CountDownLatch(1);
+    private final AtomicBoolean closed = new AtomicBoolean(false);
+
+    StallThenResetServer() throws IOException {
+      this.serverSocket = new ServerSocket(0);
+      Thread acceptThread =
+          new Thread(
+              () -> {
+                while (!serverSocket.isClosed()) {
+                  try {
+                    Socket s = serverSocket.accept();
+                    s.setSoLinger(true, 0); // close() sends RST rather than FIN
+                    accepted.add(s);
+                    connected.countDown();
+                    // never read the body: the client's send buffer and the pipe fill up
+                  } catch (IOException ignored) {
+                    return;
+                  }
+                }
+              },
+              "solr-17707-stall-server");
+      acceptThread.setDaemon(true);
+      acceptThread.start();
+    }
+
+    String baseUrl() {
+      return "http://127.0.0.1:" + serverSocket.getLocalPort() + "/solr";
+    }
+
+    void awaitConnected(long timeout, TimeUnit unit) throws InterruptedException {
+      connected.await(timeout, unit);
+    }
+
+    void resetAll() {
+      for (Socket s : accepted) {
+        try {
+          s.close();
+        } catch (IOException ignored) {
+          // ignore
+        }
+      }
+    }
+
+    @Override
+    public void close() throws IOException {
+      if (closed.compareAndSet(false, true)) {
+        resetAll();
+        serverSocket.close();
+      }
+    }
+  }
+
   @Override
   protected void testQuerySetup(SolrRequest.METHOD method, ResponseParser rp) throws Exception {
     DebugServlet.clear();