From a65bff9afbeec2217c33476d32e61ab3e0049b8f Mon Sep 17 00:00:00 2001 From: Serhiy Bzhezytskyy Date: Wed, 15 Jul 2026 15:03:08 +0300 Subject: [PATCH 1/2] SOLR-17707: release the content-writing thread when an async request fails HttpJdkSolrClient's async paths never cleaned up the content-writing task, so a connection failure while the body was being written left that thread blocked forever in PipedInputStream.awaitSpace(). Since the writer shares the client's executor, enough such failures exhaust the pool. The synchronous path already cancels the writer in its finally block; the async paths did not. Both async paths now release the writer on completion. Closing the pipe's sink is what unblocks a writer already stuck in awaitSpace(); cancelling the future alone does not, since the interrupt does not reliably reach the blocked thread. --- .../SOLR-17707-httpjdk-async-thread-leak.yml | 11 ++ .../client/solrj/impl/HttpJdkSolrClient.java | 37 ++++- .../solrj/impl/HttpJdkSolrClientTest.java | 138 ++++++++++++++++++ 3 files changed, 180 insertions(+), 6 deletions(-) create mode 100644 changelog/unreleased/SOLR-17707-httpjdk-async-thread-leak.yml diff --git a/changelog/unreleased/SOLR-17707-httpjdk-async-thread-leak.yml b/changelog/unreleased/SOLR-17707-httpjdk-async-thread-leak.yml new file mode 100644 index 000000000000..f015a99b5f3c --- /dev/null +++ b/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 diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/HttpJdkSolrClient.java b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/HttpJdkSolrClient.java index 1f88adc519fd..111bfd1bc92e 100644 --- a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/HttpJdkSolrClient.java +++ b/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> 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> cf = new CompletableFuture<>(); cf.completeExceptionally(e); @@ -160,6 +162,7 @@ public CompletableFuture> 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> 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 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,6 +321,7 @@ private PreparedRequest preparePutOrPost( HttpRequest.BodyPublisher bodyPublisher; Future contentWritingFuture = null; + PipedInputStream contentWritingSink = null; if (contentWriter != null) { boolean success = maybeTryHeadRequest(url); if (!success) { @@ -310,7 +329,8 @@ private PreparedRequest preparePutOrPost( } 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; } } diff --git a/solr/solrj/src/test/org/apache/solr/client/solrj/impl/HttpJdkSolrClientTest.java b/solr/solrj/src/test/org/apache/solr/client/solrj/impl/HttpJdkSolrClientTest.java index 73ff00474bb1..e56686ec87ef 100644 --- a/solr/solrj/src/test/org/apache/solr/client/solrj/impl/HttpJdkSolrClientTest.java +++ b/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.ExecutionException; 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,135 @@ 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 + + try { + cf.get(30, TimeUnit.SECONDS); + } catch (ExecutionException expected) { + // the dropped connection is expected to fail the request + } + + // 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 accepted = Collections.synchronizedList(new ArrayList<>()); + private final java.util.concurrent.CountDownLatch connected = + new java.util.concurrent.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(); From 7ff1e29d943f6c3218078851b3426525660ddeff Mon Sep 17 00:00:00 2001 From: Serhiy Bzhezytskyy Date: Wed, 15 Jul 2026 19:34:42 +0300 Subject: [PATCH 2/2] SOLR-17707: fix errorprone warnings in the new test Import CountDownLatch instead of fully-qualifying it, and wait for the request to settle via handle().get() rather than a catch that errorprone flags as MissingFail. --- .../client/solrj/impl/HttpJdkSolrClientTest.java | 13 +++++-------- 1 file changed, 5 insertions(+), 8 deletions(-) diff --git a/solr/solrj/src/test/org/apache/solr/client/solrj/impl/HttpJdkSolrClientTest.java b/solr/solrj/src/test/org/apache/solr/client/solrj/impl/HttpJdkSolrClientTest.java index e56686ec87ef..79bfdb63e9ca 100644 --- a/solr/solrj/src/test/org/apache/solr/client/solrj/impl/HttpJdkSolrClientTest.java +++ b/solr/solrj/src/test/org/apache/solr/client/solrj/impl/HttpJdkSolrClientTest.java @@ -33,7 +33,7 @@ import java.util.Objects; import java.util.Set; import java.util.concurrent.CompletableFuture; -import java.util.concurrent.ExecutionException; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; @@ -170,11 +170,9 @@ public void testStuckContentWritingThreadIsReleasedOnFailure() throws Exception Thread.sleep(1000); // let the writer fill the pipe and block in awaitSpace() server.resetAll(); // drop the connection under the in-flight write - try { - cf.get(30, TimeUnit.SECONDS); - } catch (ExecutionException expected) { - // the dropped connection is expected to fail the request - } + // 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( @@ -217,8 +215,7 @@ private static boolean hasBlockedWriterThread() { private static class StallThenResetServer implements AutoCloseable { private final ServerSocket serverSocket; private final List accepted = Collections.synchronizedList(new ArrayList<>()); - private final java.util.concurrent.CountDownLatch connected = - new java.util.concurrent.CountDownLatch(1); + private final CountDownLatch connected = new CountDownLatch(1); private final AtomicBoolean closed = new AtomicBoolean(false); StallThenResetServer() throws IOException {