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 00000000000..f015a99b5f3 --- /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 1f88adc519f..111bfd1bc92 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 73ff00474bb..79bfdb63e9c 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.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 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();