This is an automated email from the ASF dual-hosted git repository.
markrmiller pushed a commit to branch branch_10x
in repository https://gitbox.apache.org/repos/asf/solr.git
The following commit(s) were added to refs/heads/branch_10x by this push:
new bfd808924e0 SOLR-18244: Fix several concurrency bugs in
ParallelHttpShardHandler (#4548)
bfd808924e0 is described below
commit bfd808924e098b8f7340969297b5cfe1ab45ccfb
Author: Mark Robert Miller <[email protected]>
AuthorDate: Wed Jun 24 11:29:32 2026 -0500
SOLR-18244: Fix several concurrency bugs in ParallelHttpShardHandler (#4548)
---
...parallel-http-shard-handler-lost-wakeup-fix.yml | 10 +
.../solr/handler/component/HttpShardHandler.java | 104 +++++--
.../component/ParallelHttpShardHandler.java | 198 ++++++++++---
.../component/ParallelHttpShardHandlerTest.java | 307 ++++++++++++++++++++-
4 files changed, 563 insertions(+), 56 deletions(-)
diff --git
a/changelog/unreleased/SOLR-18244-parallel-http-shard-handler-lost-wakeup-fix.yml
b/changelog/unreleased/SOLR-18244-parallel-http-shard-handler-lost-wakeup-fix.yml
new file mode 100644
index 00000000000..96706027bc1
--- /dev/null
+++
b/changelog/unreleased/SOLR-18244-parallel-http-shard-handler-lost-wakeup-fix.yml
@@ -0,0 +1,10 @@
+# See https://github.com/apache/solr/blob/main/dev-docs/changelog.adoc
+title: Fix several concurrency bugs in HttpShardHandler /
ParallelHttpShardHandler that
+ could cause search threads to hang in take() or return HTTP 500 instead of
honoring
+ shards.tolerant under thread-pool saturation
+type: fixed
+authors:
+ - name: Mark Miller
+links:
+ - name: SOLR-18244
+ url: https://issues.apache.org/jira/browse/SOLR-18244
diff --git
a/solr/core/src/java/org/apache/solr/handler/component/HttpShardHandler.java
b/solr/core/src/java/org/apache/solr/handler/component/HttpShardHandler.java
index 43165272027..34f9a022d70 100644
--- a/solr/core/src/java/org/apache/solr/handler/component/HttpShardHandler.java
+++ b/solr/core/src/java/org/apache/solr/handler/component/HttpShardHandler.java
@@ -96,6 +96,23 @@ public class HttpShardHandler extends ShardHandler {
protected final BlockingQueue<ShardResponse> responses;
private final AtomicBoolean canceled = new AtomicBoolean(false);
+ // One monitor guards every cancellation-related state transition: the
canceled flag, the
+ // responseFutureMap, and queueing the responses queue's
CANCELLATION_NOTIFICATION. Holding it
+ // makes "has everything been canceled / is anything still outstanding?" a
single atomic question
+ // rather than a set of separately-observed flags. It is a dedicated object
(not the canceled
+ // flag itself) so the lock's identity does not depend on how cancellation
state happens to be
+ // stored. Subclasses with extra cancellable bookkeeping must read and
mutate it under this same
+ // monitor; see ParallelHttpShardHandler.
+ private final Object cancellationLock = new Object();
+
+ protected final Object cancellationLock() {
+ return cancellationLock;
+ }
+
+ protected final boolean isCanceled() {
+ return canceled.get();
+ }
+
private final Map<String, List<String>> shardToURLs;
protected LBAsyncSolrClient lbClient;
@@ -214,7 +231,7 @@ public class HttpShardHandler extends ShardHandler {
srsp.setException(exception);
srsp.setResponseCode(exception.code());
- synchronized (canceled) {
+ synchronized (cancellationLock) {
if (!canceled.get()) {
responses.add(srsp);
}
@@ -266,9 +283,10 @@ public class HttpShardHandler extends ShardHandler {
ShardResponse srsp,
long startTimeNS) {
CompletableFuture<LBSolrClient.Rsp> future =
this.lbClient.requestAsync(lbReq);
- // Synchronize on canceled, so that we know precisely whether to add it to
the responseFutureMap
- // or not.
- synchronized (canceled) {
+ // Hold the cancellation lock so the canceled check and the
responseFutureMap put happen as one
+ // step: either we register this future for later cancellation, or (if
cancelAll already ran) we
+ // cancel it now and never track it.
+ synchronized (cancellationLock) {
if (canceled.get() && !future.isDone()) {
future.cancel(true);
return;
@@ -280,27 +298,49 @@ public class HttpShardHandler extends ShardHandler {
// on the map already having the future.
future.whenComplete(
(LBSolrClient.Rsp rsp, Throwable throwable) -> {
- if (rsp != null) {
- ssr.nl = rsp.getResponse();
- srsp.setShardAddress(rsp.getServer());
- } else if (throwable != null) {
- srsp.setException(throwable);
- if (throwable instanceof SolrException) {
- srsp.setResponseCode(((SolrException) throwable).code());
+ try {
+ if (rsp != null) {
+ ssr.nl = rsp.getResponse();
+ srsp.setShardAddress(rsp.getServer());
+ } else if (throwable != null) {
+ srsp.setException(throwable);
+ if (throwable instanceof SolrException) {
+ srsp.setResponseCode(((SolrException) throwable).code());
+ }
}
- }
- ssr.elapsedTime =
- TimeUnit.MILLISECONDS.convert(System.nanoTime() - startTimeNS,
TimeUnit.NANOSECONDS);
- // Synchronize on cancelled so this code and cancelAll() cannot
happen at the same time
- synchronized (canceled) {
- // We don't want to add responses after the requests have been
canceled
- if (responseFutureMap.containsKey(srsp)) {
- responses.add(HttpShardHandler.this.transformResponse(sreq,
srsp, shard));
+ ssr.elapsedTime =
+ TimeUnit.MILLISECONDS.convert(
+ System.nanoTime() - startTimeNS, TimeUnit.NANOSECONDS);
+ enqueueIfTracked(srsp,
HttpShardHandler.this.transformResponse(sreq, srsp, shard));
+ } catch (Exception e) {
+ // If response processing throws (a subclass transformResponse, a
malformed Rsp, etc.)
+ // the response would never be enqueued — yet responseFutureMap
still tracks srsp, so a
+ // consumer in take() would park forever. Turn the failure into
the shard's response and
+ // enqueue it raw, bypassing transformResponse (which may itself
be the thrower). We
+ // deliberately catch Exception, not Throwable, so a JVM Error
(e.g. OutOfMemoryError)
+ // propagates instead of being silently downgraded to a shard
error.
+ srsp.setException(e);
+ if (e instanceof SolrException) {
+ srsp.setResponseCode(((SolrException) e).code());
}
+ enqueueIfTracked(srsp, srsp);
}
});
}
+ /**
+ * Enqueue {@code value} into the {@link #responses} queue iff {@code key}
is still tracked in
+ * {@link #responseFutureMap}, holding the cancellation monitor so this
stays atomic with {@link
+ * #cancelAll()}'s clear.
+ */
+ private void enqueueIfTracked(ShardResponse key, ShardResponse value) {
+ synchronized (cancellationLock) {
+ if (responseFutureMap.containsKey(key)) {
+ responses.add(value);
+ }
+ }
+ }
+
/** Subclasses could modify the request based on the shard */
@SuppressWarnings("unused")
protected QueryRequest createQueryRequest(
@@ -330,7 +370,12 @@ public class HttpShardHandler extends ShardHandler {
ShardResponse previousResponse = null;
try {
while (responsesPending()) {
- ShardResponse rsp = responses.take();
+ ShardResponse rsp = awaitNextResponse();
+ if (rsp == null) {
+ // awaitNextResponse() returned without a response — only happens
for subclasses that
+ // override with a timed poll. Re-evaluate responsesPending() and
either re-wait or exit.
+ continue;
+ }
if (rsp == CANCELLATION_NOTIFICATION) {
// This is only queued in cancelAll(), so all outstanding futures
have already been
// canceled.
@@ -378,6 +423,23 @@ public class HttpShardHandler extends ShardHandler {
return !responseFutureMap.isEmpty() || !responses.isEmpty();
}
+ /**
+ * Wait for the next response from the {@link #responses} queue. Defaults to
a blocking {@link
+ * BlockingQueue#take()}.
+ *
+ * <p>Subclasses that gate {@link #responsesPending()} on an async tracker
outside the {@link
+ * #responses} queue's lifecycle (e.g. {@link
ParallelHttpShardHandler#submitFutures}) MUST
+ * override this with a timed poll. The cancellation lock can serialize
{@link
+ * #responsesPending()} reads with state mutations, but it cannot signal the
queue's internal
+ * {@code Condition}: if the tracker drains without anything being enqueued
to {@link #responses},
+ * a thread parked in {@link BlockingQueue#take()} would never wake up.
Returning {@code null}
+ * from this method instructs {@link #take(boolean)} to re-check {@link
#responsesPending()} and
+ * either re-wait or exit cleanly.
+ */
+ protected ShardResponse awaitNextResponse() throws InterruptedException {
+ return responses.take();
+ }
+
@Override
public void cancelAll() {
// Canceled must be set to true before calling the cancellation code, to
ensure that new tasks
@@ -388,7 +450,7 @@ public class HttpShardHandler extends ShardHandler {
// responses will not be recorded.
// Queue a fake response to notify take() that it should no longer wait on
responses as the
// outstanding requests have been canceled
- synchronized (canceled) {
+ synchronized (cancellationLock) {
boolean alreadyCanceled = canceled.getAndSet(true);
if (!alreadyCanceled) {
// We don't want to queue this multiple times if we are already
canceled
diff --git
a/solr/core/src/java/org/apache/solr/handler/component/ParallelHttpShardHandler.java
b/solr/core/src/java/org/apache/solr/handler/component/ParallelHttpShardHandler.java
index 5eb9c992ed9..705f19d13bf 100644
---
a/solr/core/src/java/org/apache/solr/handler/component/ParallelHttpShardHandler.java
+++
b/solr/core/src/java/org/apache/solr/handler/component/ParallelHttpShardHandler.java
@@ -16,19 +16,20 @@
*/
package org.apache.solr.handler.component;
-import java.lang.invoke.MethodHandles;
import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
import net.jcip.annotations.NotThreadSafe;
import org.apache.solr.client.solrj.impl.LBSolrClient;
import org.apache.solr.common.SolrException;
import org.apache.solr.common.params.ModifiableSolrParams;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
/**
* A version of {@link HttpShardHandler} optimized for massively-sharded
collections.
@@ -39,23 +40,62 @@ import org.slf4j.LoggerFactory;
*
* <p>The additional focus on parallelization makes this an ideal
implementation for collections
* with many shards.
+ *
+ * <h2>Concurrency model</h2>
+ *
+ * <p>Each shard submit runs as two chained async stages. This class uses the
words <b>outer</b> and
+ * <b>inner</b> throughout to keep them straight:
+ *
+ * <ul>
+ * <li><b>outer future</b> — the {@link CompletableFuture#runAsync} we
schedule on {@link
+ * #commExecutor}. All it does is call {@code super.makeShardRequest},
which kicks off the
+ * actual HTTP request. Doing this on commExecutor is the whole point of
the class: {@code
+ * submit} returns immediately instead of blocking the caller once per
shard.
+ * <li><b>inner future</b> — the HTTP future that {@code
super.makeShardRequest} obtains from
+ * {@code lbClient.requestAsync}. It completes later, on a Jetty IO
thread, and its callback
+ * enqueues the finished {@link ShardResponse} onto {@link #responses}.
+ * </ul>
+ *
+ * <p>Meanwhile the consumer thread sits in {@link #take(boolean)}, looping
while {@link
+ * #responsesPending()} is true and draining finished responses off {@link
#responses}.
+ *
+ * <p><b>The core invariant:</b> {@code responsesPending()} must stay true as
long as any response
+ * might still arrive. If it ever reports false too early, the consumer stops
waiting and a response
+ * that lands a moment later is lost — and the consumer can park forever. To
answer "could anything
+ * still arrive?" it has to consult three pieces of state at once:
+ *
+ * <ul>
+ * <li>{@link #inFlightSubmits} — outer futures not yet finished (this
class).
+ * <li>{@code responseFutureMap} — inner futures not yet finished (base
class).
+ * <li>{@link #responses} — finished responses not yet consumed (base class).
+ * </ul>
+ *
+ * <p>The risky instant is the hand-off between the two stages: an outer
future finishes (so {@link
+ * #inFlightSubmits} drops) a hair before its inner future's callback puts the
response on the
+ * queue. A consumer that samples the three trackers in that gap,
unsynchronized, can see all three
+ * empty even though a response is moments away. Every {@code synchronized}
block in this class
+ * exists to close that gap: reads and mutations of all three trackers share
the one monitor
+ * returned by {@link #cancellationLock()}, so the consumer can never observe
a torn, in-between
+ * view.
*/
@NotThreadSafe
public class ParallelHttpShardHandler extends HttpShardHandler {
- @SuppressWarnings("unused")
- private static final Logger log =
LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
-
private final ExecutorService commExecutor;
/*
- * Unlike the basic HttpShardHandler, this class allows us to exit submit
before
- * the responseFutureMap is updated. If the runnables that
- * do that are slow to execute the calling code could attempt to
takeCompleted(),
- * while pending is still zero. In this condition, the code would assume
that all
- * requests are processed (despite the runnables created by this class still
- * waiting). Thus, we need to track that there are attempts still in flight.
+ * Number of outer futures (see class javadoc) not yet finished. This is the
authoritative
+ * "is a submit still in flight?" signal that responsesPending() consults.
+ *
+ * Why a counter and not submitFutures.isEmpty():
ConcurrentHashMap.size()/isEmpty() are
+ * documented as estimates. Under concurrent put/remove the internal counter
cells can settle at
+ * a non-zero sum while the table is physically empty, so isEmpty() can
return false for a
+ * logically empty map — which would leave responsesPending() stuck at true
and park the consumer
+ * forever. AtomicInteger.get() is exact under any concurrency.
*/
+ private final AtomicInteger inFlightSubmits = new AtomicInteger();
+
+ /* The outer futures themselves, kept only so cancelAll() has something to
iterate and cancel. */
private final ConcurrentMap<ShardResponse, CompletableFuture<Void>>
submitFutures;
public ParallelHttpShardHandler(ParallelHttpShardHandlerFactory
httpShardHandlerFactory) {
@@ -66,11 +106,39 @@ public class ParallelHttpShardHandler extends
HttpShardHandler {
@Override
protected boolean responsesPending() {
- // ensure we can't exit while loop in HttpShardHandler.take(boolean) until
we've completed
- // submitting all of the shard requests
- return super.responsesPending() || !submitFutures.isEmpty();
+ // Read all three trackers under the shared monitor so the consumer sees a
consistent view (see
+ // "core invariant" in the class javadoc). Without the lock it could catch
the brief gap where
+ // an outer future has finished (inFlightSubmits == 0) but its inner
future's callback has not
+ // yet added the response to the queue, conclude nothing is pending, and
lose that response.
+ synchronized (cancellationLock()) {
+ return super.responsesPending() || inFlightSubmits.get() > 0;
+ }
}
+ /**
+ * Wait for the next response by polling instead of blocking.
+ *
+ * <p>The lock fixes torn reads, but it cannot wake a thread blocked in
{@code responses.take()}:
+ * when the last outer future finishes without the inner callback enqueuing
anything (e.g. the
+ * trackers simply drain to empty), nothing is ever added to the queue to
signal its internal
+ * condition, so {@code take()} would block forever. Polling instead lets
{@link #take(boolean)}
+ * periodically re-check {@link #responsesPending()} and exit cleanly — or
pick up the inner
+ * callback's response once it lands.
+ */
+ @Override
+ protected ShardResponse awaitNextResponse() throws InterruptedException {
+ return responses.poll(50, TimeUnit.MILLISECONDS);
+ }
+
+ /**
+ * Schedules {@code super.makeShardRequest} onto {@link #commExecutor} as an
outer future (see
+ * class javadoc) instead of running it on the caller's thread.
+ *
+ * <p>The {@link #inFlightSubmits} counter is the thread-safety backbone
here. It is bumped up
+ * before the outer future is created and back down exactly once when that
future settles, so it
+ * always reflects the true number of submits still in flight — every exit
path below maintains
+ * that pairing.
+ */
@Override
protected void makeShardRequest(
ShardRequest sreq,
@@ -80,11 +148,65 @@ public class ParallelHttpShardHandler extends
HttpShardHandler {
SimpleSolrResponse ssr,
ShardResponse srsp,
long startTimeNS) {
- CompletableFuture<Void> completableFuture =
- CompletableFuture.runAsync(
- () -> super.makeShardRequest(sreq, shard, params, lbReq, ssr,
srsp, startTimeNS),
- commExecutor);
- submitFutures.put(srsp, completableFuture);
+ // The runnable below needs a reference to its own outer future to check
isCancelled(), but that
+ // future doesn't exist until runAsync returns. We can't close over a
not-yet-assigned local, so
+ // we hand the runnable an AtomicReference and fill it in once runAsync
returns the future. That
+ // AtomicReference also gives the runnable's thread a safe, visible read
of the late assignment.
+ AtomicReference<CompletableFuture<Void>> selfRef = new AtomicReference<>();
+ CompletableFuture<Void> completableFuture;
+ // Count this submit as in flight before it can possibly start, so
responsesPending() never sees
+ // a zero count while a submit exists. Each branch below pairs this with
exactly one decrement.
+ inFlightSubmits.incrementAndGet();
+ try {
+ completableFuture =
+ CompletableFuture.runAsync(
+ () -> {
+ // If cancelAll already cancelled this specific outer future
before it got to run,
+ // skip the HTTP call that super would otherwise start and
immediately cancel.
+ // selfRef can still be null if this runnable beats the
selfRef.set() below; that's
+ // harmless, since super.makeShardRequest re-checks the
canceled flag itself.
+ CompletableFuture<Void> self = selfRef.get();
+ if (self != null && self.isCancelled()) {
+ return;
+ }
+ super.makeShardRequest(sreq, shard, params, lbReq, ssr, srsp,
startTimeNS);
+ },
+ commExecutor);
+ } catch (RejectedExecutionException ree) {
+ // commExecutor is saturated or shutting down. If we let this propagate
it would blow up
+ // SearchHandler's distributed loop before cancelAll() runs, stranding
shard requests already
+ // submitted and turning a transient overload into an HTTP 500 even
under shards.tolerant.
+ // Instead record it as an ordinary shard failure (503, i.e. transient)
so the responses queue
+ // stays consistent. No outer future was created, so there is no
whenComplete to balance the
+ // increment — undo it here. (If the rejection happens because we are
already canceling,
+ // recordShardSubmitError intentionally drops the response: the consumer
is tearing down and
+ // is woken by cancelAll's CANCELLATION_NOTIFICATION, not by this shard.)
+ inFlightSubmits.decrementAndGet();
+ recordShardSubmitError(
+ srsp,
+ new SolrException(
+ SolrException.ErrorCode.SERVICE_UNAVAILABLE,
+ "Comm executor thread pool is full, unable to send request to
shard: " + shard,
+ ree));
+ return;
+ }
+ // Publish the future before the cancellation check below: if that block
cancels it, the
+ // runnable is then guaranteed to observe the cancellation through
selfRef.get().isCancelled().
+ selfRef.set(completableFuture);
+
+ // Register the outer future for cancelAll to find — but only if we
haven't already been
+ // canceled. This mirrors super.makeShardRequest's check-and-put on
responseFutureMap and runs
+ // under the same monitor, so registration and cancellation can't
interleave: if cancelAll has
+ // already run, we cancel this future and skip tracking it (the runnable
will short-circuit).
+ synchronized (cancellationLock()) {
+ if (isCanceled()) {
+ completableFuture.cancel(true);
+ // Early return: no whenComplete will be attached, so balance the
increment here.
+ inFlightSubmits.decrementAndGet();
+ return;
+ }
+ submitFutures.put(srsp, completableFuture);
+ }
completableFuture.whenComplete(
(r, t) -> {
try {
@@ -104,23 +226,35 @@ public class ParallelHttpShardHandler extends
HttpShardHandler {
}
}
} finally {
- // Remove so that we keep track of in-flight submits only
- submitFutures.remove(srsp);
+ // Remove from submitFutures first (under the shared monitor, so
cancelAll always sees a
+ // consistent set), then drop the count. Decrementing last
guarantees responsesPending()
+ // — which reads both under that monitor — never sees
inFlightSubmits == 0 while this
+ // entry is still in submitFutures.
+ synchronized (cancellationLock()) {
+ submitFutures.remove(srsp);
+ }
+ inFlightSubmits.decrementAndGet();
}
});
}
@Override
public void cancelAll() {
- super.cancelAll();
- submitFutures
- .values()
- .forEach(
- future -> {
- if (!future.isDone()) {
- future.cancel(true);
- }
- });
- submitFutures.clear();
+ // Cancel base-class state and our outer futures together under the one
monitor, so a concurrent
+ // makeShardRequest can't slip a new outer future into submitFutures while
we are sweeping it.
+ // Same monitor as super.cancelAll(), so the canceled flag,
responseFutureMap, and submitFutures
+ // all flip to the canceled state atomically.
+ synchronized (cancellationLock()) {
+ super.cancelAll();
+ submitFutures
+ .values()
+ .forEach(
+ future -> {
+ if (!future.isDone()) {
+ future.cancel(true);
+ }
+ });
+ submitFutures.clear();
+ }
}
}
diff --git
a/solr/core/src/test/org/apache/solr/handler/component/ParallelHttpShardHandlerTest.java
b/solr/core/src/test/org/apache/solr/handler/component/ParallelHttpShardHandlerTest.java
index 4eb0db8858f..454f1a8f9fa 100644
---
a/solr/core/src/test/org/apache/solr/handler/component/ParallelHttpShardHandlerTest.java
+++
b/solr/core/src/test/org/apache/solr/handler/component/ParallelHttpShardHandlerTest.java
@@ -18,16 +18,35 @@ package org.apache.solr.handler.component;
import java.util.List;
import java.util.concurrent.AbstractExecutorService;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.SynchronousQueue;
+import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import org.apache.solr.SolrTestCaseJ4;
+import org.apache.solr.client.solrj.impl.LBAsyncSolrClient;
import org.apache.solr.client.solrj.impl.LBSolrClient;
import org.apache.solr.client.solrj.request.QueryRequest;
import org.apache.solr.common.SolrException;
import org.apache.solr.common.params.ModifiableSolrParams;
+import org.apache.solr.common.util.ExecutorUtil;
+import org.apache.solr.common.util.SolrNamedThreadFactory;
+import org.junit.BeforeClass;
import org.junit.Test;
+import org.mockito.Mockito;
public class ParallelHttpShardHandlerTest extends SolrTestCaseJ4 {
+ @BeforeClass
+ public static void ensureWorkingMockito() {
+ assumeWorkingMockito();
+ }
+
private static class DirectExecutorService extends AbstractExecutorService {
private volatile boolean shutdown;
@@ -72,9 +91,7 @@ public class ParallelHttpShardHandlerTest extends
SolrTestCaseJ4 {
// Force super.makeShardRequest to throw before it enqueues the response
future.
handler.lbClient = null;
- ShardRequest shardRequest = new ShardRequest();
- shardRequest.params = new ModifiableSolrParams();
- shardRequest.actualShards = new String[] {"shardA"};
+ ShardRequest shardRequest = buildShardRequest("shardA");
ShardResponse shardResponse = new ShardResponse();
shardResponse.setShardRequest(shardRequest);
@@ -105,4 +122,288 @@ public class ParallelHttpShardHandlerTest extends
SolrTestCaseJ4 {
recorded.getException());
assertTrue(recorded.getException() instanceof SolrException);
}
+
+ /**
+ * Verifies the contract that when the commExecutor rejects the runnable,
the failure is recorded
+ * via recordShardSubmitError (i.e., shows up in the responses queue) rather
than being propagated
+ * synchronously to the caller.
+ *
+ * <p>This exercises issue #1 from the ParallelHttpShardHandler review: with
a single-thread
+ * ThreadPoolExecutor backed by a SynchronousQueue, once the worker is busy,
the next
+ * CompletableFuture.runAsync(...) call throws RejectedExecutionException
synchronously out of
+ * makeShardRequest. The expected (post-fix) behavior is that the error is
routed through
+ * recordShardSubmitError instead.
+ */
+ @Test
+ public void testRejectedExecutorRecordsErrorInsteadOfThrowing() throws
Exception {
+ CountDownLatch holdWorker = new CountDownLatch(1);
+ CountDownLatch workerStarted = new CountDownLatch(1);
+ ThreadPoolExecutor busyExecutor =
+ new ExecutorUtil.MDCAwareThreadPoolExecutor(
+ 1, 1, 0L, TimeUnit.MILLISECONDS, new SynchronousQueue<>()); //
default AbortPolicy
+ try {
+ // Occupy the single worker thread so the next submission has nowhere to
go.
+ busyExecutor.execute(
+ () -> {
+ workerStarted.countDown();
+ try {
+ holdWorker.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ });
+ assertTrue("worker did not start within timeout", workerStarted.await(5,
TimeUnit.SECONDS));
+
+ ParallelHttpShardHandlerFactory factory = new
ParallelHttpShardHandlerFactory();
+ factory.commExecutor = busyExecutor;
+ ParallelHttpShardHandler handler = new ParallelHttpShardHandler(factory);
+
+ ShardRequest shardRequest = buildShardRequest("shardA");
+
+ ShardResponse shardResponse = new ShardResponse();
+ shardResponse.setShardRequest(shardRequest);
+ shardResponse.setShard("shardA");
+
+ HttpShardHandler.SimpleSolrResponse simpleResponse =
+ new HttpShardHandler.SimpleSolrResponse();
+ shardResponse.setSolrResponse(simpleResponse);
+
+ ModifiableSolrParams params = new ModifiableSolrParams();
+ QueryRequest queryRequest = new QueryRequest(params);
+ LBSolrClient.Endpoint endpoint = new
LBSolrClient.Endpoint("http://ignored:8983/solr";);
+ LBSolrClient.Req lbReq = new LBSolrClient.Req(queryRequest,
List.of(endpoint));
+
+ // The desired contract: rejection is captured and surfaced through the
responses queue
+ // (i.e., this call should not throw RejectedExecutionException).
+ try {
+ handler.makeShardRequest(
+ shardRequest,
+ "shardA",
+ params,
+ lbReq,
+ simpleResponse,
+ shardResponse,
+ System.nanoTime());
+ } catch (RejectedExecutionException ree) {
+ fail(
+ "makeShardRequest should not propagate RejectedExecutionException;
the failure "
+ + "should be recorded via recordShardSubmitError. Got: "
+ + ree);
+ }
+
+ ShardResponse recorded = handler.responses.poll(2, TimeUnit.SECONDS);
+ assertNotNull(
+ "Expected the executor rejection to be recorded as a shard failure
in the responses"
+ + " queue, but no response arrived",
+ recorded);
+ assertSame(
+ "The recorded shard response should be the same instance passed in",
+ shardResponse,
+ recorded);
+ assertNotNull(
+ "Expected an exception to be attached to the recorded shard
response",
+ recorded.getException());
+ } finally {
+ holdWorker.countDown();
+ busyExecutor.shutdownNow();
+ busyExecutor.awaitTermination(5, TimeUnit.SECONDS);
+ }
+ }
+
+ private ShardRequest buildShardRequest(String shard) {
+ ShardRequest sreq = new ShardRequest();
+ sreq.params = new ModifiableSolrParams();
+ sreq.actualShards = new String[] {shard};
+ return sreq;
+ }
+
+ /**
+ * Runs handler.takeCompletedIncludingErrors() on a worker thread with a
timeout. If take() does
+ * not return within timeoutMs, fails the test with a clear message naming
the iteration and phase
+ * — this is the signal for the lost-wakeup bug.
+ */
+ private ShardResponse runTakeWithTimeout(
+ ParallelHttpShardHandler handler,
+ ExecutorService takeExecutor,
+ int iteration,
+ String phaseLabel,
+ long timeoutMs)
+ throws Exception {
+ Future<ShardResponse> future =
takeExecutor.submit(handler::takeCompletedIncludingErrors);
+ try {
+ return future.get(timeoutMs, TimeUnit.MILLISECONDS);
+ } catch (TimeoutException te) {
+ future.cancel(true);
+ throw new AssertionError(
+ "take() hung in iteration "
+ + iteration
+ + " "
+ + phaseLabel
+ + ": did not return within "
+ + timeoutMs
+ + "ms. The worker thread is parked in LinkedBlockingQueue.take()
waiting for"
+ + " an element that will never arrive because the handler's
state transitioned"
+ + " to empty without anything being enqueued on the responses
queue.");
+ } catch (ExecutionException ee) {
+ throw new AssertionError(
+ "take() threw unexpectedly in iteration " + iteration + " " +
phaseLabel, ee.getCause());
+ }
+ }
+
+ /**
+ * More aggressive variant of the lost-wakeup stress test that uses
asynchronous inner-future
+ * completion on a dedicated scheduler. In production the inner future (from
{@code
+ * lbClient.requestAsync}) completes on a Jetty IO thread, not synchronously
at the registration
+ * site. That timing gap between {@code super.makeShardRequest} returning
(and the outer {@code
+ * whenComplete} firing to remove {@code submitFutures}) and the inner
{@code whenComplete} firing
+ * (to add to {@code responses}) is exactly where the observed 930-handler
hang lives. This test
+ * matches that timing.
+ */
+ @Test
+ public void testTakeDoesNotHangUnderAsyncInnerFutureCompletion() throws
Exception {
+ // The race window is wide, so a modest count reliably catches a
regression on routine CI; run
+ // many more under -Dtests.nightly to keep deep coverage without bounding
every build by the
+ // worst-case (iterations * perIterationTimeoutMs) hang time.
+ final int iterations = TEST_NIGHTLY ? 1000 : 100;
+ final long perIterationTimeoutMs = 3_000;
+
+ ExecutorService commExecutor =
+ new ExecutorUtil.MDCAwareThreadPoolExecutor(
+ 0,
+ Integer.MAX_VALUE,
+ 5L,
+ TimeUnit.SECONDS,
+ new SynchronousQueue<>(),
+ new SolrNamedThreadFactory("testCommExecutor"));
+
+ // Simulates Jetty IO threads: a small pool that completes the inner
future asynchronously
+ // some tiny amount of time after requestAsync() returns, exposing the
race window.
+ ExecutorService mockIoThreads =
+ ExecutorUtil.newMDCAwareFixedThreadPool(2, new
SolrNamedThreadFactory("testMockIo"));
+
+ ExecutorService takeExecutor =
+ ExecutorUtil.newMDCAwareCachedThreadPool(new
SolrNamedThreadFactory("testTakeRunner"));
+
+ try {
+ for (int i = 0; i < iterations; i++) {
+ runAsyncRaceCycle(commExecutor, mockIoThreads, takeExecutor, i,
perIterationTimeoutMs);
+ }
+ } finally {
+ takeExecutor.shutdownNow();
+ takeExecutor.awaitTermination(5, TimeUnit.SECONDS);
+ mockIoThreads.shutdownNow();
+ mockIoThreads.awaitTermination(5, TimeUnit.SECONDS);
+ commExecutor.shutdown();
+ if (!commExecutor.awaitTermination(15, TimeUnit.SECONDS)) {
+ commExecutor.shutdownNow();
+ commExecutor.awaitTermination(5, TimeUnit.SECONDS);
+ }
+ }
+ }
+
+ private void runAsyncRaceCycle(
+ ExecutorService commExecutor,
+ ExecutorService mockIoThreads,
+ ExecutorService takeExecutor,
+ int iteration,
+ long timeoutMs)
+ throws Exception {
+
+ ParallelHttpShardHandlerFactory factory = new
ParallelHttpShardHandlerFactory();
+ factory.commExecutor = commExecutor;
+ ParallelHttpShardHandler handler = new ParallelHttpShardHandler(factory);
+
+ // LB client that returns a future which completes asynchronously on a
separate thread —
+ // mimicking the Jetty IO thread model. This creates a real race between:
+ // (a) the outer runAsync future completing + its whenComplete removing
submitFutures,
+ // (b) the inner future completing + its whenComplete adding to
responses.
+ LBAsyncSolrClient mockLb = Mockito.mock(LBAsyncSolrClient.class);
+ Mockito.when(mockLb.requestAsync(Mockito.any(LBSolrClient.Req.class)))
+ .thenAnswer(
+ inv -> {
+ CompletableFuture<LBSolrClient.Rsp> f = new
CompletableFuture<>();
+ mockIoThreads.execute(() -> f.complete(new LBSolrClient.Rsp()));
+ return f;
+ });
+ handler.lbClient = mockLb;
+
+ // Single-shard submit → take. This is the simplest real workload. Under
async inner-future
+ // completion, the outer whenComplete (removing submitFutures) and inner
whenComplete
+ // (adding to responses) race. If there's a window where
responsesPending() transitions to
+ // false without the responses queue getting an entry, take() parks
forever.
+ ShardRequest sreq = buildShardRequest("shard-" + iteration);
+ handler.submit(sreq, "shard-" + iteration, sreq.params);
+
+ ShardResponse rsp =
+ runTakeWithTimeout(handler, takeExecutor, iteration, "async-race",
timeoutMs);
+
+ assertNotNull(
+ "async-race iteration " + iteration + " take() returned null —
response was never enqueued",
+ rsp);
+ }
+
+ /**
+ * Invariant test for the cancellation synchronization contract in {@link
+ * ParallelHttpShardHandler}: when {@code makeShardRequest} is invoked while
{@code canceled} is
+ * already {@code true}, the outer future must be cancelled and NOT tracked
in {@code
+ * submitFutures}. This keeps {@code submitFutures} consistent with the
cancellation state —
+ * mirroring {@link HttpShardHandler#makeShardRequest}'s check-and-put
pattern on {@code
+ * responseFutureMap}.
+ *
+ * <p>Without this invariant, a runnable could observe {@code canceled=true}
(and early-return in
+ * super) while {@code submitFutures} still tracks its outer future, leaving
the outer
+ * whenComplete's bookkeeping racing against {@code cancelAll}'s own
submitFutures sweep.
+ */
+ @Test
+ public void testCanceledMakeShardRequestDoesNotTrackSubmitFutures() throws
Exception {
+ ExecutorService commExecutor =
+ new ExecutorUtil.MDCAwareThreadPoolExecutor(
+ 0,
+ Integer.MAX_VALUE,
+ 5L,
+ TimeUnit.SECONDS,
+ new SynchronousQueue<>(),
+ new SolrNamedThreadFactory("invariantTestComm"));
+
+ try {
+ ParallelHttpShardHandlerFactory factory = new
ParallelHttpShardHandlerFactory();
+ factory.commExecutor = commExecutor;
+ ParallelHttpShardHandler handler = new ParallelHttpShardHandler(factory);
+
+ LBAsyncSolrClient mockLb = Mockito.mock(LBAsyncSolrClient.class);
+ Mockito.when(mockLb.requestAsync(Mockito.any(LBSolrClient.Req.class)))
+ .thenAnswer(inv -> new CompletableFuture<LBSolrClient.Rsp>());
+ handler.lbClient = mockLb;
+
+ // Force canceled=true and drain the CANCELLATION_NOTIFICATION so we can
observe the
+ // post-cancel state cleanly.
+ handler.cancelAll();
+ assertNotNull(
+ "CANCELLATION_NOTIFICATION should be queued by cancelAll",
+ handler.responses.poll(2, TimeUnit.SECONDS));
+
+ ShardRequest sreq = buildShardRequest("shardA");
+ ShardResponse srsp = new ShardResponse();
+ srsp.setShardRequest(sreq);
+ srsp.setShard("shardA");
+ HttpShardHandler.SimpleSolrResponse ssr = new
HttpShardHandler.SimpleSolrResponse();
+ srsp.setSolrResponse(ssr);
+
+ ModifiableSolrParams params = new ModifiableSolrParams();
+ QueryRequest queryRequest = new QueryRequest(params);
+ LBSolrClient.Endpoint endpoint = new
LBSolrClient.Endpoint("http://ignored:8983/solr";);
+ LBSolrClient.Req lbReq = new LBSolrClient.Req(queryRequest,
List.of(endpoint));
+
+ // Invoke makeShardRequest while canceled=true. Expected: outer is
cancelled, nothing is
+ // tracked in submitFutures, responsesPending() stays false.
+ handler.makeShardRequest(sreq, "shardA", params, lbReq, ssr, srsp,
System.nanoTime());
+
+ assertFalse(
+ "submitFutures must not track requests submitted while
canceled=true",
+ handler.responsesPending());
+ } finally {
+ commExecutor.shutdownNow();
+ commExecutor.awaitTermination(5, TimeUnit.SECONDS);
+ }
+ }
}
