This is an automated email from the ASF dual-hosted git repository.
dsmiley pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/solr.git
The following commit(s) were added to refs/heads/main by this push:
new aa9e3f1a784 SOLR-17707: HttpJdkSolrClient fix thread leak when async
request fails (#4639)
aa9e3f1a784 is described below
commit aa9e3f1a7841e765d3806951e2d6f0460f6a057d
Author: Serhiy Bzhezytskyy <[email protected]>
AuthorDate: Thu Jul 16 21:28:05 2026 +0300
SOLR-17707: HttpJdkSolrClient fix thread leak when async request fails
(#4639)
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.
---
.../SOLR-17707-httpjdk-async-thread-leak.yml | 11 ++
.../solr/client/solrj/impl/HttpJdkSolrClient.java | 37 +++++-
.../client/solrj/impl/HttpJdkSolrClientTest.java | 135 +++++++++++++++++++++
3 files changed, 177 insertions(+), 6 deletions(-)
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 @@ public class HttpJdkSolrClient extends HttpSolrClient {
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 class HttpJdkSolrClient extends HttpSolrClient {
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 class HttpJdkSolrClient extends HttpSolrClient {
}
}
+ 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 @@ public class HttpJdkSolrClient extends HttpSolrClient {
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 @@ public class HttpJdkSolrClient extends HttpSolrClient {
HttpRequest.BodyPublisher bodyPublisher;
Future<?> contentWritingFuture = null;
+ PipedInputStream contentWritingSink = null;
if (contentWriter != null) {
boolean success = maybeTryHeadRequest(url);
if (!success) {
@@ -310,7 +329,8 @@ public class HttpJdkSolrClient extends HttpSolrClient {
}
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 @@ public class HttpJdkSolrClient extends HttpSolrClient {
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 @@ package org.apache.solr.client.solrj.impl;
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.SolrServerException;
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 class HttpJdkSolrClientTest extends
HttpSolrClientTestBase {
}
}
+ /**
+ * 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();
