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 282346f941f SOLR-18312: HttpJdkSolrClient: Fix Thread Pool Starvation
(#4655)
282346f941f is described below
commit 282346f941fda0987e7b2a0771696c18038d6395
Author: Renato Haeberli <[email protected]>
AuthorDate: Sun Aug 16 22:46:13 2026 +0200
SOLR-18312: HttpJdkSolrClient: Fix Thread Pool Starvation (#4655)
Fix Thread Pool Starvation in HttpJdkSolrClient.
HttpJdkSolrClient defaults would often create 32 threads, likely
under-utilizing them. and it would cap threads to 256. Now it does neither by
default but Executor customization (and other saturation controls) remain.
Co-authored-by: David Smiley <[email protected]>
---
...Thread-Pool-Starvation-in-HttpJdkSolrClient.yml | 11 ++
.../pages/major-changes-in-solr-10.adoc | 1 +
.../solr/client/solrj/impl/HttpJdkSolrClient.java | 159 ++++++++++++---------
.../client/solrj/impl/HttpJdkSolrClientTest.java | 49 +++++++
4 files changed, 152 insertions(+), 68 deletions(-)
diff --git
a/changelog/unreleased/SOLR-18312-fix-Thread-Pool-Starvation-in-HttpJdkSolrClient.yml
b/changelog/unreleased/SOLR-18312-fix-Thread-Pool-Starvation-in-HttpJdkSolrClient.yml
new file mode 100644
index 00000000000..3691824f823
--- /dev/null
+++
b/changelog/unreleased/SOLR-18312-fix-Thread-Pool-Starvation-in-HttpJdkSolrClient.yml
@@ -0,0 +1,11 @@
+title: >
+ Fix Thread Pool Starvation in HttpJdkSolrClient.
+ HttpJdkSolrClient defaults would often create 32 threads, likely
under-utilizing them. and it would cap threads to 256.
+ Now it does neither by default but Executor customization (and other
saturation controls) remain.
+type: fixed
+authors:
+ - name: Renato Haeberli
+ - name: David Smiley
+links:
+ - name: SOLR-18312
+ url: https://issues.apache.org/jira/browse/SOLR-18312
diff --git
a/solr/solr-ref-guide/modules/upgrade-notes/pages/major-changes-in-solr-10.adoc
b/solr/solr-ref-guide/modules/upgrade-notes/pages/major-changes-in-solr-10.adoc
index a065a7413e1..cfa96a93865 100644
---
a/solr/solr-ref-guide/modules/upgrade-notes/pages/major-changes-in-solr-10.adoc
+++
b/solr/solr-ref-guide/modules/upgrade-notes/pages/major-changes-in-solr-10.adoc
@@ -95,6 +95,7 @@ When enabled, this escape hatch relaxes both the login-time
check and the user c
It is intended as a temporary measure while you migrate the affected accounts
to stronger passwords, and should be removed once that is done.
=== SolrJ
+The `HttpJdkSolrClient` and `HttpJettySolrClient` no longer have default
thread/executor limits. Nonetheless the Executor is configurable.
HttpSolrClient returns; this time as a base class for HttpJettySolrClient and
HttpJdkSolrClient.
Its builder will dynamically detect if solr-jetty is available and use that,
otherwise it will use the JDK client.
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 111bfd1bc92..722b241cef6 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
@@ -21,6 +21,7 @@ import java.io.IOException;
import java.io.InputStream;
import java.io.PipedInputStream;
import java.io.PipedOutputStream;
+import java.io.UncheckedIOException;
import java.lang.invoke.MethodHandles;
import java.net.CookieHandler;
import java.net.InetSocketAddress;
@@ -37,12 +38,9 @@ import java.util.Collection;
import java.util.HashMap;
import java.util.Locale;
import java.util.Map;
-import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.TimeUnit;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import javax.net.ssl.SSLContext;
@@ -68,7 +66,10 @@ import org.slf4j.LoggerFactory;
* A SolrClient implementation that communicates to a Solr server using the
built-in Java 11+ Http
* Client. This client is targeted for those users who wish to minimize
application dependencies.
* This client will connect to solr using Http/2 but can seamlessly downgrade
to Http/1.1 when
- * connecting to Solr hosts running on older versions.
+ * connecting to Solr hosts running on older versions. Uses two {@link
+ * java.util.concurrent.ThreadPoolExecutor}, one for {@link HttpClient}
(consumer) and one for
+ * writing request bodies (producer). Both are unbounded cached thread pools
(maximumPoolSize =
+ * Integer.MAX_VALUE).
*/
public class HttpJdkSolrClient extends HttpSolrClient {
private static final Logger log =
LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
@@ -78,54 +79,63 @@ public class HttpJdkSolrClient extends HttpSolrClient {
protected HttpClient httpClient;
+ /**
+ * Executor used to stream (produce) request bodies into the pipe consumed
by the JDK HttpClient.
+ */
+ protected ExecutorService requestBodyExecutor;
+
+ /** Dedicated executor handed to the JDK HttpClient */
protected ExecutorService executor;
private boolean forceHttp11;
private final boolean shutdownExecutor;
+ /**
+ * {@link ExecutorService} on {@link HttpJdkSolrClient.Builder} is used for
{@link HttpClient}
+ * only.
+ */
protected HttpJdkSolrClient(String serverBaseUrl, HttpJdkSolrClient.Builder
builder) {
super(serverBaseUrl, builder);
- HttpClient.Builder b = HttpClient.newBuilder();
+ HttpClient.Builder httpClientBuilder = HttpClient.newBuilder();
HttpClient.Redirect followRedirects =
Boolean.TRUE.equals(builder.getFollowRedirects())
? HttpClient.Redirect.NORMAL
: HttpClient.Redirect.NEVER;
- b.followRedirects(followRedirects);
+ httpClientBuilder.followRedirects(followRedirects);
- b.connectTimeout(Duration.of(builder.getConnectionTimeoutMillis(),
ChronoUnit.MILLIS));
+ httpClientBuilder.connectTimeout(
+ Duration.of(builder.getConnectionTimeoutMillis(), ChronoUnit.MILLIS));
// note: idle timeout isn't used for the JDK client
// note: request timeout is set per request
if (builder.sslContext != null) {
- b.sslContext(builder.sslContext);
+ httpClientBuilder.sslContext(builder.sslContext);
}
if (builder.getExecutor() != null) {
this.executor = builder.getExecutor();
this.shutdownExecutor = false;
} else {
- BlockingQueue<Runnable> queue = new LinkedBlockingQueue<>(1024);
this.executor =
- new ExecutorUtil.MDCAwareThreadPoolExecutor(
- 4,
- 256,
- 60,
- TimeUnit.SECONDS,
- queue,
+ ExecutorUtil.newMDCAwareCachedThreadPool(
new SolrNamedThreadFactory(this.getClass().getSimpleName()));
this.shutdownExecutor = true;
}
- b.executor(this.executor);
+ httpClientBuilder.executor(this.executor);
+
+ this.requestBodyExecutor =
+ ExecutorUtil.newMDCAwareCachedThreadPool(
+ new SolrNamedThreadFactory(this.getClass().getSimpleName() +
"-reqBody"));
if (builder.shouldUseHttp1_1()) {
this.forceHttp11 = true;
- b.version(HttpClient.Version.HTTP_1_1);
+ httpClientBuilder.version(HttpClient.Version.HTTP_1_1);
}
if (builder.cookieHandler != null) {
- b.cookieHandler(builder.cookieHandler);
+ httpClientBuilder.cookieHandler(builder.cookieHandler);
}
if (builder.getProxyHost() != null) {
@@ -133,10 +143,10 @@ public class HttpJdkSolrClient extends HttpSolrClient {
log.warn(
"Socks4 is likely not supported by this client. See
https://bugs.openjdk.org/browse/JDK-8214516";);
}
- b.proxy(
+ httpClientBuilder.proxy(
ProxySelector.of(new InetSocketAddress(builder.getProxyHost(),
builder.getProxyPort())));
}
- this.httpClient = b.build();
+ this.httpClient = httpClientBuilder.build();
assert ObjectReleaseTracker.track(this);
}
@@ -147,7 +157,7 @@ public class HttpJdkSolrClient extends HttpSolrClient {
PreparedRequest pReq = prepareRequest(baseUrl, solrRequest, collection);
return httpClient
.sendAsync(pReq.reqb.build(),
HttpResponse.BodyHandlers.ofInputStream())
- .whenComplete((httpResponse, throwable) ->
releaseContentWriting(pReq));
+ .whenComplete((httpResponse, throwable) ->
pReq.releaseContentWriting());
} catch (Exception e) {
CompletableFuture<HttpResponse<InputStream>> cf = new
CompletableFuture<>();
cf.completeExceptionally(e);
@@ -162,7 +172,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))
+ .whenComplete((httpResponse, throwable) ->
pReq.releaseContentWriting())
.thenApply(
httpResponse -> {
try {
@@ -179,21 +189,6 @@ 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)
@@ -214,9 +209,7 @@ public class HttpJdkSolrClient extends HttpSolrClient {
} catch (RuntimeException e) {
throw new SolrServerException(e);
} finally {
- if (pReq.contentWritingFuture != null) {
- pReq.contentWritingFuture.cancel(true);
- }
+ pReq.releaseContentWriting();
// See
//
https://docs.oracle.com/en/java/javase/17/docs/api/java.net.http/java/net/http/HttpResponse.BodySubscribers.html#ofInputStream()
@@ -252,7 +245,7 @@ public class HttpJdkSolrClient extends HttpSolrClient {
ResponseParser parserToUse = responseParser(solrRequest);
ModifiableSolrParams queryParams = initializeSolrParams(solrRequest,
parserToUse);
var reqb = HttpRequest.newBuilder();
- PreparedRequest pReq = null;
+ PreparedRequest pReq;
try {
switch (solrRequest.getMethod()) {
case GET:
@@ -289,7 +282,7 @@ public class HttpJdkSolrClient extends HttpSolrClient {
reqb.GET();
decorateRequest(reqb, solrRequest);
reqb.uri(new URI(url + queryParams.toQueryString()));
- return new PreparedRequest(reqb, null, null);
+ return new PreparedRequest(reqb);
}
private PreparedRequest preparePutOrPost(
@@ -320,28 +313,16 @@ public class HttpJdkSolrClient extends HttpSolrClient {
}
HttpRequest.BodyPublisher bodyPublisher;
- Future<?> contentWritingFuture = null;
- PipedInputStream contentWritingSink = null;
+ PreparedRequest pReq = new PreparedRequest(reqb);
if (contentWriter != null) {
boolean success = maybeTryHeadRequest(url);
if (!success) {
reqb.version(HttpClient.Version.HTTP_1_1);
}
- final PipedOutputStream source = new PipedOutputStream();
- contentWritingSink = new PipedInputStream(source);
- final PipedInputStream sink = contentWritingSink;
- bodyPublisher = HttpRequest.BodyPublishers.ofInputStream(() -> sink);
-
- contentWritingFuture =
- executor.submit(
- () -> {
- try (source) {
- contentWriter.write(source);
- } catch (Exception e) {
- log.error("Cannot write Content Stream", e);
- }
- });
+ bodyPublisher =
+ HttpRequest.BodyPublishers.ofInputStream(
+ () -> pReq.beginContentWriting(contentWriter,
this.requestBodyExecutor));
} else if (streams != null && streams.size() == 1) {
boolean success = maybeTryHeadRequest(url);
if (!success) {
@@ -372,25 +353,62 @@ public class HttpJdkSolrClient extends HttpSolrClient {
URI uriWithQueryParams = new URI(url + queryParams.toQueryString());
reqb.uri(uriWithQueryParams);
- return new PreparedRequest(reqb, contentWritingFuture, contentWritingSink);
+ return pReq;
}
protected static class PreparedRequest {
- Future<?> contentWritingFuture;
- PipedInputStream contentWritingSink;
- HttpRequest.Builder reqb;
+ final HttpRequest.Builder reqb;
ResponseParser parserToUse;
String url;
- PreparedRequest(
- HttpRequest.Builder reqb,
- Future<?> contentWritingFuture,
- PipedInputStream contentWritingSink) {
+ // Both remain null if the request has no streamed content, or if the body
is never requested
+ // (e.g. the connection failed before sending it). Filled in lazily by
+ // beginContentWriting once the JDK HttpClient actually requests the body.
+ private PipedInputStream contentWritingSink;
+ private Future<?> contentWritingFuture;
+
+ PreparedRequest(HttpRequest.Builder reqb) {
this.reqb = reqb;
- this.contentWritingFuture = contentWritingFuture;
- this.contentWritingSink = contentWritingSink;
+ }
+
+ synchronized PipedInputStream beginContentWriting(
+ RequestWriter.ContentWriter contentWriter, ExecutorService
bodyExecutor) {
+ final PipedOutputStream source = new PipedOutputStream();
+ try {
+ contentWritingSink = new PipedInputStream(source);
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+
+ contentWritingFuture =
+ bodyExecutor.submit(
+ () -> {
+ // note: doesn't need to synchronize with PreparedRequest.this
+ try (source) {
+ contentWriter.write(source);
+ } catch (Exception e) {
+ log.error("Cannot write Content Stream", e);
+ }
+ });
+ return contentWritingSink;
+ }
+
+ synchronized void releaseContentWriting() {
+ if (contentWritingFuture != null) {
+ contentWritingFuture.cancel(true);
+ }
+
+ // Closing the sink is what unblocks a writer already stuck in the pipe;
cancel() alone does
+ // not.
+ if (contentWritingSink != null) {
+ try {
+ contentWritingSink.close();
+ } catch (IOException e) {
+ log.warn("Could not close content-writing pipe", e);
+ }
+ }
}
}
@@ -545,6 +563,11 @@ public class HttpJdkSolrClient extends HttpSolrClient {
}
executor = null;
+ if (requestBodyExecutor != null) {
+ ExecutorUtil.shutdownAndAwaitTermination(requestBodyExecutor);
+ requestBodyExecutor = null;
+ }
+
assert ObjectReleaseTracker.release(this);
}
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 79bfdb63e9c..9f9233f375e 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
@@ -48,6 +48,7 @@ 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.request.json.JsonQueryRequest;
import org.apache.solr.client.solrj.response.JavaBinResponseParser;
import org.apache.solr.client.solrj.response.ResponseParser;
import org.apache.solr.client.solrj.response.SolrPingResponse;
@@ -637,6 +638,54 @@ public class HttpJdkSolrClientTest extends
HttpSolrClientTestBase {
}
}
+ @Test(timeout = 30000)
+ public void testConcurrentStreamedBodiesDoNotDeadlockWithHttp1() throws
Exception {
+ DebugServlet.clear();
+ DebugServlet.addResponseHeader("Content-Type", "application/octet-stream");
+ DebugServlet.responseBodyByQueryFragment.put("", javabinResponse());
+ String url = solrTestRule.getBaseUrl() + DEBUG_SERVLET_PATH;
+
+ int concurrency = 8;
+ ExecutorService callers =
+ ExecutorUtil.newMDCAwareFixedThreadPool(concurrency, new
NamedThreadFactory("test-caller"));
+
+ try (HttpJdkSolrClient client = builder(url).useHttp1_1(true).build()) {
+ List<CompletableFuture<Void>> futures = new ArrayList<>(concurrency);
+ for (int i = 0; i < concurrency; i++) {
+ futures.add(
+ CompletableFuture.runAsync(
+ () -> {
+ JsonQueryRequest q = buildLargeBodyQuery();
+ try {
+ q.process(client);
+ } catch (SolrServerException | IOException e) {
+ throw new RuntimeException(e);
+ }
+ },
+ callers));
+ }
+ CompletableFuture.allOf(futures.toArray(new CompletableFuture<?>[0]))
+ .get(30, TimeUnit.SECONDS);
+ } finally {
+ ExecutorUtil.shutdownAndAwaitTermination(callers);
+ }
+ }
+
+ private static JsonQueryRequest buildLargeBodyQuery() {
+ StringBuilder filter = new StringBuilder("id:(");
+ for (int i = 0; i < 400; i++) {
+ if (i > 0) {
+ filter.append(" OR ");
+ }
+ filter.append("value_").append(i);
+ }
+ filter.append(')');
+ JsonQueryRequest q = new JsonQueryRequest();
+ q.setQuery("*:*");
+ q.withFilter(filter.toString());
+ return q;
+ }
+
/**
* This is not required for any test, but there appears to be a bug in the
JDK client where it
* does not release all threads if the client has not performed any queries,
even after a forced
