This is an automated email from the ASF dual-hosted git repository.
epugh 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 489237e71c0 SOLR-18480 prevent TLOG+PULL leader election stall by
aborting fetcher (#4974)
489237e71c0 is described below
commit 489237e71c0fe0c91756be701d4b4c354c1c32b1
Author: Luke Kot-Zaniewski <[email protected]>
AuthorDate: Wed Oct 14 21:52:42 2026 +0100
SOLR-18480 prevent TLOG+PULL leader election stall by aborting fetcher
(#4974)
---
AGENTS.md | 4 +
changelog/unreleased/SOLR-18480-abort-fetcher.yml | 12 +
.../java/org/apache/solr/handler/IndexFetcher.java | 163 +++++++---
.../apache/solr/handler/ReplicationHandler.java | 2 +-
.../cloud/TlogLeaderElectionFrozenLeaderTest.java | 343 +++++++++++++++++++++
.../handler/IndexFetcherPacketProtocolTest.java | 28 +-
.../client/solrj/jetty/HttpJettySolrClient.java | 113 ++++++-
.../solrj/jetty/HttpJettySolrClientTest.java | 158 ++++++++++
.../java/org/apache/solr/util/ServletFixtures.java | 28 +-
9 files changed, 788 insertions(+), 63 deletions(-)
diff --git a/AGENTS.md b/AGENTS.md
index dab06fac42b..a7153a9d79a 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -43,6 +43,10 @@ While README.md and CONTRIBUTING.md are mainly written for
humans, this file is
- For BATS shell integration tests in `solr/packaging/test/`:
- Always use `run <command>` followed by `assert_output --partial "..."`
or `refute_output --partial "..."` instead of capturing output into local
variables and using `[[ ]]` comparisons
- Avoid patterns like `local var=$(cmd | grep ...); [[ "$var" == *"..."*
]]` — use `run cmd` + `assert_output`/`refute_output` instead
+- Avoid `Thread.sleep`; fixed sleeps waste build time on every run and can
flake under load.
+ - Have the other thread signal when it reaches a point, e.g. complete a
`CompletableFuture` that the test thread `get`s with a timeout.
+ - If you must poll, use `org.apache.solr.util.TimeOut#waitFor` (solr-core)
rather than a hand-rolled sleep loop.
+ - Derive timeouts from what they bound and define them as constants, e.g.
`ABORT_DEADLINE_MS = SlowServlet.DELAY_MS / 2`, not a bare `2, SECONDS`.
## Documentation
diff --git a/changelog/unreleased/SOLR-18480-abort-fetcher.yml
b/changelog/unreleased/SOLR-18480-abort-fetcher.yml
new file mode 100644
index 00000000000..4bb212ef87b
--- /dev/null
+++ b/changelog/unreleased/SOLR-18480-abort-fetcher.yml
@@ -0,0 +1,12 @@
+title: >-
+ A TLOG replica taking over leadership no longer stalls for up to 60 seconds
when the previous
+ leader has stopped responding without going down: the in-progress index
fetch from the old
+ leader is now aborted instead of waited out. To support this,
HttpJettySolrClient gains
+ abort(Throwable) and isAborted(), enabled with
Builder.withAbortableRequests(true); abort fails
+ the client's in-flight and subsequent requests without waiting for them.
+type: fixed
+authors:
+ - name: Luke Kot-Zaniewski
+links:
+ - name: SOLR-18480
+ url: https://issues.apache.org/jira/browse/SOLR-18480
diff --git a/solr/core/src/java/org/apache/solr/handler/IndexFetcher.java
b/solr/core/src/java/org/apache/solr/handler/IndexFetcher.java
index 2652f2add4c..a71366d7dbb 100644
--- a/solr/core/src/java/org/apache/solr/handler/IndexFetcher.java
+++ b/solr/core/src/java/org/apache/solr/handler/IndexFetcher.java
@@ -71,6 +71,7 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BooleanSupplier;
import java.util.stream.Collectors;
import java.util.zip.Adler32;
@@ -171,7 +172,14 @@ public class IndexFetcher {
private volatile ExecutorService fsyncService;
- private volatile boolean stop = false;
+ /**
+ * The client of the fetch in progress, if any. Each fetch gets its own
lightweight copy of {@link
+ * #solrClient}, so aborting it affects that fetch alone and nothing later.
+ */
+ private final AtomicReference<HttpJettySolrClient> currentFetchClient = new
AtomicReference<>();
+
+ /** Set by {@link #destroy()}; a destroyed fetcher starts no further
fetches. */
+ private volatile boolean destroyed;
private boolean useInternalCompression = false;
@@ -206,9 +214,8 @@ public class IndexFetcher {
new IndexFetchResult("Fetching latest index is successful", true,
null);
public static final IndexFetchResult LOCK_OBTAIN_FAILED =
new IndexFetchResult("Obtaining SnapPuller lock failed", false, null);
- public static final IndexFetchResult CONTAINER_IS_SHUTTING_DOWN =
- new IndexFetchResult(
- "I was asked to replicate but CoreContainer is shutting down",
false, null);
+ public static final IndexFetchResult REPLICATION_SHUTTING_DOWN =
+ new IndexFetchResult("Index fetch not started: replication is shutting
down", false, null);
public static final IndexFetchResult LEADER_VERSION_ZERO =
new IndexFetchResult("Index in peer is empty and never committed yet",
true, null);
public static final IndexFetchResult NO_INDEX_COMMIT_EXIST =
@@ -351,12 +358,16 @@ public class IndexFetcher {
/** Gets the latest commit version and generation from the leader */
public NamedList<Object> getLatestVersion() throws IOException {
+ return getLatestVersion(solrClient);
+ }
+
+ private NamedList<Object> getLatestVersion(HttpJettySolrClient client)
throws IOException {
ModifiableSolrParams params = new ModifiableSolrParams();
params.set(COMMAND, CMD_INDEX_VERSION);
params.set(CommonParams.WT, JAVABIN);
var req = createReplicationHandlerRequest(params);
try {
- return solrClient.requestWithBaseUrl(leaderBaseUrl, req, leaderCoreName);
+ return client.requestWithBaseUrl(leaderBaseUrl, req, leaderCoreName);
} catch (SolrServerException e) {
throw new SolrException(ErrorCode.SERVER_ERROR, e.getMessage(), e);
}
@@ -367,20 +378,20 @@ public class IndexFetcher {
* download.
*/
@SuppressWarnings({"unchecked"})
- private void fetchFileList(long gen) throws IOException {
+ private void fetchFileList(Fetch fetch) throws IOException {
ModifiableSolrParams params = new ModifiableSolrParams();
params.set(COMMAND, CMD_GET_FILE_LIST);
- params.set(GENERATION, String.valueOf(gen));
+ params.set(GENERATION, String.valueOf(fetch.generation()));
params.set(CommonParams.WT, JAVABIN);
var req = createReplicationHandlerRequest(params);
try {
- NamedList<?> response = solrClient.requestWithBaseUrl(leaderBaseUrl,
req, leaderCoreName);
+ NamedList<?> response = fetch.client().requestWithBaseUrl(leaderBaseUrl,
req, leaderCoreName);
List<Map<String, Object>> files = (List<Map<String, Object>>)
response.get(CMD_GET_FILE_LIST);
if (files != null) filesToDownload = Collections.synchronizedList(files);
else {
filesToDownload = List.of();
- log.error("No files to download for index generation: {}", gen);
+ log.error("No files to download for index generation: {}",
fetch.generation());
}
files = (List<Map<String, Object>>) response.get(CONF_FILES);
@@ -406,17 +417,52 @@ public class IndexFetcher {
*/
IndexFetchResult fetchLatestIndex(boolean forceReplication, boolean
forceCoreReload)
throws IOException, InterruptedException {
+ // A lightweight copy of solrClient, so that aborting it (abortFetch,
destroy) fails this
+ // fetch's requests and nothing else.
+ HttpJettySolrClient fetchClient =
+ new HttpJettySolrClient.Builder()
+ .withHttpClient(solrClient)
+ .withAbortableRequests(true)
+ .build();
+ // doFetch serializes fetches with indexFetchLock, so a fetch already in
progress means that
+ // invariant broke. Fail loudly rather than silently abort or orphan the
other fetch.
+ if (!currentFetchClient.compareAndSet(null, fetchClient)) {
+ IOUtils.closeQuietly(fetchClient);
+ throw new IllegalStateException("A fetch is already in progress on " +
this);
+ }
+ try {
+ // Pairs with destroy(): publish-then-check here and mark-then-read
there, so either this
+ // sees the fetcher destroyed, or destroy() sees this fetch and aborts
it.
+ if (destroyed) {
+ return IndexFetchResult.REPLICATION_SHUTTING_DOWN;
+ }
+ return fetchLatestIndexWithRetry(forceReplication, forceCoreReload,
fetchClient);
+ } finally {
+ currentFetchClient.compareAndSet(fetchClient, null);
+ IOUtils.closeQuietly(fetchClient);
+ }
+ }
+
+ /**
+ * Runs one fetch, restarting once if the leader discards the generation
being fetched. Recursive
+ * attempts call this rather than {@link #fetchLatestIndex(boolean,
boolean)}, so they use the
+ * caller's per-fetch client and an abort reaches them too.
+ */
+ private IndexFetchResult fetchLatestIndexWithRetry(
+ boolean forceReplication, boolean forceCoreReload, HttpJettySolrClient
fetchClient)
+ throws IOException, InterruptedException {
try {
- return fetchLatestIndexOnce(forceReplication, forceCoreReload);
+ return fetchLatestIndexOnce(forceReplication, forceCoreReload,
fetchClient);
} catch (InvalidIndexGenerationException e) {
log.info(
"Leader no longer has index generation {}; restarting replication
from its latest generation",
e.generation);
- return fetchLatestIndexOnce(forceReplication, forceCoreReload);
+ return fetchLatestIndexOnce(forceReplication, forceCoreReload,
fetchClient);
}
}
- private IndexFetchResult fetchLatestIndexOnce(boolean forceReplication,
boolean forceCoreReload)
+ private IndexFetchResult fetchLatestIndexOnce(
+ boolean forceReplication, boolean forceCoreReload, HttpJettySolrClient
fetchClient)
throws IOException, InterruptedException {
boolean cleanupDone = false;
@@ -477,7 +523,7 @@ public class IndexFetcher {
// get the current 'replicateable' index version in the leader
NamedList<?> response;
try {
- response = getLatestVersion();
+ response = getLatestVersion(fetchClient);
} catch (Exception e) {
final String errorMsg = e.toString();
if (StrUtils.isNotNullOrEmpty(errorMsg) &&
errorMsg.contains(INTERRUPT_RESPONSE_MESSAGE)) {
@@ -499,6 +545,7 @@ public class IndexFetcher {
long latestGeneration = (Long) response.get(GENERATION);
log.info("Leader's generation: {}", latestGeneration);
+ Fetch fetch = new Fetch(fetchClient, latestGeneration);
log.info("Leader's version: {}", latestVersion);
Span.current()
.setAttribute("replication.leader", leaderCoreUrl)
@@ -571,7 +618,7 @@ public class IndexFetcher {
}
log.info("Starting replication process");
// get the list of files first
- fetchFileList(latestGeneration);
+ fetchFileList(fetch);
assert testWait.getAsBoolean();
// this can happen if the commit point is deleted before we fetch the
file list.
if (filesToDownload.isEmpty()) {
@@ -676,12 +723,7 @@ public class IndexFetcher {
long bytesDownloaded =
downloadIndexFiles(
- isFullCopyNeeded,
- indexDir,
- tmpIndexDir,
- indexDirPath,
- tmpIndexDirPath,
- latestGeneration);
+ isFullCopyNeeded, indexDir, tmpIndexDir, indexDirPath,
tmpIndexDirPath, fetch);
final long timeTakenSeconds = getReplicationTimeElapsed();
final Long bytesDownloadedPerSecond =
(timeTakenSeconds != 0 ? bytesDownloaded / timeTakenSeconds :
null);
@@ -700,7 +742,7 @@ public class IndexFetcher {
getModifiedConfFiles(confFilesToDownload);
if (!modifiedConfFiles.isEmpty()) {
reloadCore = true;
- downloadConfFiles(confFilesToDownload, latestGeneration);
+ downloadConfFiles(confFilesToDownload, fetch);
if (isFullCopyNeeded) {
successfulInstall = solrCore.modifyIndexProps(tmpIdxDirName);
if (successfulInstall) deleteTmpIdxDir = false;
@@ -772,7 +814,8 @@ public class IndexFetcher {
log.warn(
"Replication attempt was not successful - trying a full index
replication reloadCore={}",
reloadCore);
- successfulInstall = fetchLatestIndex(true,
reloadCore).getSuccessful();
+ successfulInstall =
+ fetchLatestIndexWithRetry(true, reloadCore,
fetchClient).getSuccessful();
}
markReplicationStop();
@@ -837,7 +880,6 @@ public class IndexFetcher {
localFileFetcher = null;
if (fsyncService != null && !ExecutorUtil.isShutdown(fsyncService))
fsyncService.shutdown();
fsyncService = null;
- stop = false;
fsyncException = null;
} finally {
// order below is important
@@ -1079,8 +1121,7 @@ public class IndexFetcher {
}
}
- private void downloadConfFiles(
- List<Map<String, Object>> confFilesToDownload, long latestGeneration) {
+ private void downloadConfFiles(List<Map<String, Object>>
confFilesToDownload, Fetch fetch) {
log.info("Starting download of configuration files from leader: {}",
confFilesToDownload);
confFilesDownloaded = Collections.synchronizedList(new ArrayList<>());
Path tmpConfDir =
@@ -1095,8 +1136,7 @@ public class IndexFetcher {
}
for (Map<String, Object> file : confFilesToDownload) {
String saveAs = (String) (file.get(ALIAS) == null ? file.get(NAME) :
file.get(ALIAS));
- localFileFetcher =
- new LocalFsFileFetcher(tmpConfDir, file, saveAs, CONF_FILE_SHORT,
latestGeneration);
+ localFileFetcher = new LocalFsFileFetcher(tmpConfDir, file, saveAs,
CONF_FILE_SHORT, fetch);
currentFile = file;
localFileFetcher.fetchFile();
confFilesDownloaded.add(new HashMap<>(file));
@@ -1121,7 +1161,7 @@ public class IndexFetcher {
* @param indexDir the indexDir to be merged to
* @param tmpIndexDir the directory to which files need to be downloaded to
* @param indexDirPath the path of indexDir
- * @param latestGeneration the version number
+ * @param fetch the fetch, including the generation being fetched
* @return number of bytes downloaded
*/
private long downloadIndexFiles(
@@ -1130,7 +1170,7 @@ public class IndexFetcher {
Directory tmpIndexDir,
String indexDirPath,
String tmpIndexDirPath,
- long latestGeneration)
+ Fetch fetch)
throws Exception {
if (log.isDebugEnabled()) {
log.debug("Download files to dir: {}",
Arrays.asList(indexDir.listAll()));
@@ -1185,8 +1225,7 @@ public class IndexFetcher {
bytesSkippedCopying += Files.size(localFile);
} else {
dirFileFetcher =
- new DirectoryFileFetcher(
- tmpIndexDir, file, (String) file.get(NAME), FILE,
latestGeneration);
+ new DirectoryFileFetcher(tmpIndexDir, file, (String)
file.get(NAME), FILE, fetch);
currentFile = file;
dirFileFetcher.fetchFile();
bytesDownloaded += dirFileFetcher.getBytesDownloaded();
@@ -1487,9 +1526,12 @@ public class IndexFetcher {
}
}
- /** Stops the ongoing fetch */
+ /** Aborts the fetch in progress, if any. A no-op when idle; later fetches
are unaffected. */
void abortFetch() {
- stop = true;
+ HttpJettySolrClient fetchClient = currentFetchClient.get();
+ if (fetchClient != null) {
+ fetchClient.abort(new IOException("Index fetch aborted"));
+ }
}
@SuppressForbidden(reason = "Need currentTimeMillis for debugging/stats")
@@ -1586,20 +1628,22 @@ public class IndexFetcher {
private final Checksum checksum;
private int errorCount = 0;
private boolean aborted = false;
+ private final HttpJettySolrClient fetchClient;
FileFetcher(
FileInterface file,
Map<String, Object> fileDetails,
String saveAs,
String solrParamOutput,
- long latestGen) {
+ Fetch fetch) {
this.file = file;
+ this.fetchClient = fetch.client();
this.fileName = (String) fileDetails.get(NAME);
this.size = (Long) fileDetails.get(SIZE);
buf = new byte[(int) Math.min(this.size, ReplicationAPIBase.PACKET_SZ)];
this.solrParamOutput = solrParamOutput;
this.saveAs = saveAs;
- indexGen = latestGen;
+ indexGen = fetch.generation();
if (includeChecksum) {
checksum = new Adler32();
} else {
@@ -1674,11 +1718,6 @@ public class IndexFetcher {
}
return 0;
}
- if (stop) {
- stop = false;
- aborted = true;
- throw new ReplicationHandlerException("User aborted replication");
- }
long checkSumServer = -1;
fis.readFully(intbytes);
@@ -1720,9 +1759,11 @@ public class IndexFetcher {
// errorCount is always set to zero after a successful packet
errorCount = 0;
}
- } catch (ReplicationHandlerException e) {
- throw e;
} catch (Exception e) {
+ // An abort fails the exchange under us; that is not a transient error
worth retrying.
+ if (fetchClient.isAborted()) {
+ throw abortedException();
+ }
log.warn(
"Error in fetching file: {} (downloaded {} of {} bytes)",
fileName,
@@ -1753,6 +1794,14 @@ public class IndexFetcher {
}
}
+ /**
+ * Marks this download as aborted, which suppresses retries, and returns
the exception to throw.
+ */
+ private ReplicationHandlerException abortedException() {
+ aborted = true;
+ return new ReplicationHandlerException("User aborted replication");
+ }
+
/**
* The web container flushes the data only after it fills the buffer size.
So, all data has to
* be read as readFully() otherwise it fails. So read everything as bytes
and then extract an
@@ -1804,7 +1853,7 @@ public class IndexFetcher {
}
/** Open a new stream using HttpClient */
- private FastInputStream getStream() throws IOException {
+ private FastInputStream getStream() throws IOException,
ReplicationHandlerException {
ModifiableSolrParams params = new ModifiableSolrParams();
// the method is command=filecontent
@@ -1833,7 +1882,7 @@ public class IndexFetcher {
var req = createReplicationHandlerRequest(params);
req.setResponseParser(new InputStreamResponseParser(FILE_STREAM));
if (useExternalCompression) req.addHeader("Accept-Encoding", "gzip");
- response = solrClient.requestWithBaseUrl(leaderBaseUrl, req,
leaderCoreName);
+ response = fetchClient.requestWithBaseUrl(leaderBaseUrl, req,
leaderCoreName);
final var responseStatus = (Integer) response.get("responseStatus");
is = (InputStream) response.get("stream");
@@ -1859,8 +1908,11 @@ public class IndexFetcher {
IOUtils.closeQuietly(is);
throw e;
} catch (Exception e) {
- final var ioe = closeStreamAndBuildIOE(is, "Could not download file '"
+ fileName + "'", e);
- throw ioe;
+ if (fetchClient.isAborted()) {
+ IOUtils.closeQuietly(is);
+ throw abortedException();
+ }
+ throw closeStreamAndBuildIOE(is, "Could not download file '" +
fileName + "'", e);
}
}
@@ -1874,6 +1926,13 @@ public class IndexFetcher {
}
}
+ /**
+ * A fetch of one leader generation: the fetch's client plus the generation
being fetched. A retry
+ * (when the leader discards the generation, or the full-copy retry) reads a
fresh generation and
+ * starts a new {@code Fetch} with the same client.
+ */
+ private record Fetch(HttpJettySolrClient client, long generation) {}
+
private static class InvalidIndexGenerationException extends IOException {
private final long generation;
@@ -1921,10 +1980,9 @@ public class IndexFetcher {
Map<String, Object> fileDetails,
String saveAs,
String solrParamOutput,
- long latestGen)
+ Fetch fetch)
throws IOException {
- super(
- new DirectoryFile(tmpIndexDir, saveAs), fileDetails, saveAs,
solrParamOutput, latestGen);
+ super(new DirectoryFile(tmpIndexDir, saveAs), fileDetails, saveAs,
solrParamOutput, fetch);
}
}
@@ -1980,9 +2038,9 @@ public class IndexFetcher {
Map<String, Object> fileDetails,
String saveAs,
String solrParamOutput,
- long latestGen)
+ Fetch fetch)
throws IOException {
- super(new LocalFsFile(dir, saveAs), fileDetails, saveAs,
solrParamOutput, latestGen);
+ super(new LocalFsFile(dir, saveAs), fileDetails, saveAs,
solrParamOutput, fetch);
}
}
@@ -1996,6 +2054,9 @@ public class IndexFetcher {
}
public void destroy() {
+ // Marks this fetcher destroyed, which fetchLatestIndex checks, then
aborts the fetch in
+ // progress. See fetchLatestIndex for why this order closes the race with
a starting fetch.
+ destroyed = true;
abortFetch();
IOUtils.closeQuietly(solrClient);
}
diff --git a/solr/core/src/java/org/apache/solr/handler/ReplicationHandler.java
b/solr/core/src/java/org/apache/solr/handler/ReplicationHandler.java
index 6278a33fbfb..01058b59c9b 100644
--- a/solr/core/src/java/org/apache/solr/handler/ReplicationHandler.java
+++ b/solr/core/src/java/org/apache/solr/handler/ReplicationHandler.java
@@ -526,7 +526,7 @@ public class ReplicationHandler extends RequestHandlerBase
if (!indexFetchLock.tryLock()) return IndexFetchResult.LOCK_OBTAIN_FAILED;
if (core.getCoreContainer().isShutDown()) {
log.warn("I was asked to replicate but CoreContainer is shutting down");
- return IndexFetchResult.CONTAINER_IS_SHUTTING_DOWN;
+ return IndexFetchResult.REPLICATION_SHUTTING_DOWN;
}
try {
if (leaderUrl != null) {
diff --git
a/solr/core/src/test/org/apache/solr/cloud/TlogLeaderElectionFrozenLeaderTest.java
b/solr/core/src/test/org/apache/solr/cloud/TlogLeaderElectionFrozenLeaderTest.java
new file mode 100644
index 00000000000..5856cc6c177
--- /dev/null
+++
b/solr/core/src/test/org/apache/solr/cloud/TlogLeaderElectionFrozenLeaderTest.java
@@ -0,0 +1,343 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.solr.cloud;
+
+import com.carrotsearch.randomizedtesting.annotations.Name;
+import com.carrotsearch.randomizedtesting.annotations.ParametersFactory;
+import jakarta.servlet.Filter;
+import jakarta.servlet.FilterChain;
+import jakarta.servlet.ServletException;
+import jakarta.servlet.ServletOutputStream;
+import jakarta.servlet.ServletRequest;
+import jakarta.servlet.ServletResponse;
+import jakarta.servlet.http.HttpServletRequest;
+import jakarta.servlet.http.HttpServletResponse;
+import jakarta.servlet.http.HttpServletResponseWrapper;
+import java.io.IOException;
+import java.io.InterruptedIOException;
+import java.lang.invoke.MethodHandles;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.solr.client.solrj.request.CollectionAdminRequest;
+import org.apache.solr.common.SolrInputDocument;
+import org.apache.solr.common.cloud.Replica;
+import org.apache.solr.common.cloud.Slice;
+import org.apache.solr.common.util.SuppressForbidden;
+import org.apache.solr.embedded.JettySolrRunner;
+import org.apache.solr.handler.ReplicationHandler;
+import org.apache.solr.servlet.ServletOutputStreamWrapper;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Reproduces a TLOG leader election that stalls because the outgoing leader
froze rather than died.
+ *
+ * <p>{@link ShardLeaderElectionContext#runLeaderProcess} calls {@link
+ * ZkController#stopReplicationFromLeader} inline on the election thread. That
tears down the
+ * follower's replication process, which blocks in {@code
ExecutorUtil.shutdownAndAwaitTermination}
+ * until the in-flight index fetch finishes. {@code IndexFetcher.abortFetch}
only sets a flag that
+ * is polled while streaming file packets, so a fetch parked in the network
phase is not cut short:
+ * the election parks for the 60s executor wait before {@code shutdownNow()}
finally interrupts the
+ * poll thread.
+ */
+public class TlogLeaderElectionFrozenLeaderTest extends SolrCloudTestCase {
+
+ private static final Logger log =
LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ private static final String COLLECTION = "tlog_frozen_leader";
+ private static final String SHARD = "shard1";
+
+ /**
+ * A replacement election sleeps a fixed 2.5s for in-flight updates to settle
+ * (ShardLeaderElectionContext), then syncs, replays its tlog and publishes
in well under a
+ * second. The bug parks it for the 60s {@code
ExecutorUtil.awaitTermination} wait instead.
+ */
+ private static final long MAX_ACCEPTABLE_ELECTION_MS = 10_000;
+
+ /** How long to wait for the follower's next poll to reach the stalled
leader. */
+ private static final long STALL_ARRIVAL_TIMEOUT_MS = 30_000;
+
+ /** Where a follower's fetch can park against a frozen leader. */
+ public enum TestStallPoint {
+ INDEX_VERSION(ReplicationHandler.CMD_INDEX_VERSION, false),
+ FILE_LIST(ReplicationHandler.CMD_GET_FILE_LIST, false),
+ FILE_CONTENT(ReplicationHandler.CMD_GET_FILE, false),
+ /**
+ * Headers and part of the first packet arrive, then the body stops: the
common production case.
+ */
+ FILE_CONTENT_BODY(ReplicationHandler.CMD_GET_FILE, true);
+
+ final String command;
+ final boolean midBody;
+
+ TestStallPoint(String command, boolean midBody) {
+ this.command = command;
+ this.midBody = midBody;
+ }
+ }
+
+ /**
+ * Nightly runs every stall point. Otherwise only the mid-body file
download, the most likely
+ * failure in production, so every CI run covers it.
+ */
+ @ParametersFactory
+ public static Iterable<Object[]> parameters() {
+ List<TestStallPoint> stallPoints =
+ TEST_NIGHTLY ? List.of(TestStallPoint.values()) :
List.of(TestStallPoint.FILE_CONTENT_BODY);
+ return stallPoints.stream().map(stallPoint -> new Object[]
{stallPoint}).toList();
+ }
+
+ private final TestStallPoint stallPoint;
+
+ public TlogLeaderElectionFrozenLeaderTest(@Name("stallPoint") TestStallPoint
stallPoint) {
+ this.stallPoint = stallPoint;
+ }
+
+ @Before
+ public void setupCluster() throws Exception {
+ System.setProperty("solr.directoryFactory",
"solr.StandardDirectoryFactory");
+
+ // extraFilters are installed ahead of SolrServlet and its filters
(JettySolrRunner:336-338), so
+ // ours sees the request first. It is installed on every node but only
acts on the armed core.
+ configureCluster(2)
+ .withJettyConfig(b -> b.withFilter(TestStallReplicationFilter.class,
"/*"))
+ .addConfig("conf", configset("cloud-minimal"))
+ .configure();
+ }
+
+ @After
+ public void tearDownCluster() throws Exception {
+ TestStallChannel stallChannel =
TestStallReplicationFilter.STALL_CHANNEL.getAndSet(null);
+ if (stallChannel != null) {
+ stallChannel.released().complete(null);
+ }
+ shutdownCluster();
+ System.clearProperty("solr.directoryFactory");
+ }
+
+ @Test
+ public void testElectionIsNotBlockedByFrozenOldLeader() throws Exception {
+ CollectionAdminRequest.createCollection(COLLECTION, "conf", 1, 0, 2, 0)
+ .process(cluster.getSolrClient());
+ cluster.waitForActiveCollection(COLLECTION, 1, 2);
+
+ // Index something so the follower has a real index to poll against.
+ indexAndCommit(0, 10);
+
+ Slice shard = getCollectionState(COLLECTION).getSlice(SHARD);
+ Replica oldLeader = shard.getLeader();
+ Replica follower =
+ shard.getReplicas().stream()
+ .filter(r -> !r.getName().equals(oldLeader.getName()))
+ .findFirst()
+ .orElseThrow();
+ JettySolrRunner leaderJetty = cluster.getReplicaJetty(oldLeader);
+ JettySolrRunner followerJetty = cluster.getReplicaJetty(follower);
+
+ // Stall one of the leader's replication commands. The follower polls
every second under
+ // jetty.testMode, so its next request for that command lands in the
filter and never returns.
+ if (log.isInfoEnabled()) {
+ log.info("Stalling {} responses from leader core {}", stallPoint,
oldLeader.getCoreName());
+ }
+ TestStallChannel channel = new TestStallChannel(oldLeader.getCoreName(),
stallPoint);
+ assertTrue(
+ "the replication filter was already armed; a previous test did not
release it",
+ TestStallReplicationFilter.STALL_CHANNEL.compareAndSet(null, channel));
+
+ // Give the follower something new to fetch. Every poll sends
indexversion, but filelist and
+ // filecontent are only requested once the leader has a newer commit than
the follower.
+ indexAndCommit(10, 20);
+
+ // The ordering here is important: once the leader leaves live_nodes,
later polls bail out
+ // early (LEADER_IS_NOT_ACTIVE) without making an HTTP call. Only a fetch
that is *already* in
+ // the network phase reproduces the bug.
+ try {
+ channel.arrived().get(STALL_ARRIVAL_TIMEOUT_MS, TimeUnit.MILLISECONDS);
+ } catch (TimeoutException e) {
+ fail("the follower never issued a " + stallPoint.command + " request to
the leader");
+ }
+ log.info("Follower's index fetch is parked in the leader's replication
handler");
+
+ if (log.isInfoEnabled()) {
+ log.info("Expiring the ZooKeeper session of the frozen leader {}",
leaderJetty.getNodeName());
+ }
+ long start = System.nanoTime();
+ cluster.expireZkSession(leaderJetty);
+
+ waitForState(
+ "the surviving TLOG replica never became leader",
+ COLLECTION,
+ 90,
+ TimeUnit.SECONDS,
+ (liveNodes, collectionState) -> {
+ Replica leader = collectionState.getLeader(SHARD);
+ return leader != null
+ && leader.isActive(liveNodes)
+ && leader.getNodeName().equals(followerJetty.getNodeName());
+ });
+ long elapsedMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start);
+
+ log.info("New leader elected after {}ms", elapsedMs);
+ assertTrue(
+ "Leader election took "
+ + elapsedMs
+ + "ms with a frozen old leader; expected under "
+ + MAX_ACCEPTABLE_ELECTION_MS
+ + "ms. The election thread is parked in stopReplicationFromLeader
waiting on an index "
+ + "fetch that abortFetch() cannot cancel.",
+ elapsedMs < MAX_ACCEPTABLE_ELECTION_MS);
+ }
+
+ private static void indexAndCommit(int fromId, int toId) throws Exception {
+ for (int i = fromId; i < toId; i++) {
+ SolrInputDocument doc = new SolrInputDocument();
+ doc.addField("id", String.valueOf(i));
+ cluster.getSolrClient(COLLECTION).add(doc);
+ }
+ cluster.getSolrClient(COLLECTION).commit();
+ }
+
+ /**
+ * Stalls {@code /replication} requests for one command addressed to one
particular core,
+ * simulating a leader whose process is alive but which has stopped
answering. Depending on the
+ * {@link TestStallPoint}, the stall happens either before any response is
sent, or after the
+ * headers and the first half of the body's first write have been flushed to
the wire.
+ */
+ public static class TestStallReplicationFilter implements Filter {
+
+ private static final AtomicReference<TestStallChannel> STALL_CHANNEL = new
AtomicReference<>();
+
+ @Override
+ public void doFilter(ServletRequest request, ServletResponse response,
FilterChain chain)
+ throws IOException, ServletException {
+ // Capture once: the test nulls STALL_CHANNEL out in tearDownCluster().
+ TestStallChannel stallChannel = STALL_CHANNEL.get();
+ if (stallChannel == null || !stallChannel.matches(request)) {
+ chain.doFilter(request, response);
+ } else if (stallChannel.stallPoint().midBody) {
+ chain.doFilter(request, new TestMidBodyStall((HttpServletResponse)
response, stallChannel));
+ } else {
+ stallChannel.stall();
+ chain.doFilter(request, response);
+ }
+ }
+ }
+
+ /**
+ * Lets the headers and the first half of the body's first write reach the
client, then stalls.
+ */
+ private static class TestMidBodyStall extends HttpServletResponseWrapper {
+ private final TestStallChannel stallChannel;
+ private ServletOutputStream stream;
+ private boolean stalled;
+
+ TestMidBodyStall(HttpServletResponse response, TestStallChannel
stallChannel) {
+ super(response);
+ this.stallChannel = stallChannel;
+ }
+
+ @Override
+ public ServletOutputStream getOutputStream() throws IOException {
+ if (stream == null) {
+ // Each write overload delegates independently, so all three need the
hook.
+ stream =
+ new ServletOutputStreamWrapper(super.getOutputStream()) {
+ @Override
+ public void write(int b) throws IOException {
+ super.write(b);
+ stallOnce(this);
+ }
+
+ @Override
+ public void write(byte[] b) throws IOException {
+ write(b, 0, b.length);
+ }
+
+ @Override
+ public void write(byte[] b, int off, int len) throws IOException
{
+ if (!stalled && len > 1) {
+ int head = len / 2;
+ super.write(b, off, head);
+ stallOnce(this);
+ super.write(b, off + head, len - head);
+ } else {
+ super.write(b, off, len);
+ stallOnce(this);
+ }
+ }
+ };
+ }
+ return stream;
+ }
+
+ private void stallOnce(ServletOutputStream out) throws IOException {
+ if (!stalled) {
+ stalled = true;
+ // Need to flush Jetty buffer so client gets header bytes.
+ out.flush();
+ stallChannel.stall();
+ }
+ }
+ }
+
+ private record TestStallChannel(
+ String coreToStall,
+ TestStallPoint stallPoint,
+ CompletableFuture<Void> arrived,
+ CompletableFuture<Void> released) {
+ TestStallChannel(String coreToStall, TestStallPoint stallPoint) {
+ this(coreToStall, stallPoint, new CompletableFuture<>(), new
CompletableFuture<>());
+ }
+
+ @SuppressForbidden(reason = "forbiddenApis: getParameter is fine in tests")
+ boolean matches(ServletRequest request) {
+ if (!(request instanceof HttpServletRequest http)) {
+ return false;
+ }
+ String uri = http.getRequestURI();
+ return uri != null
+ && uri.endsWith("/" + coreToStall + ReplicationHandler.PATH)
+ && stallPoint.command.equals(http.getParameter("command"));
+ }
+
+ /**
+ * Tells the test the fetch has arrived and cannot complete, then holds
the calling Jetty thread
+ * for longer than the acceptable leader election duration, or until the
test releases it.
+ */
+ void stall() throws IOException {
+ arrived.complete(null);
+ try {
+ released.get(MAX_ACCEPTABLE_ELECTION_MS * 2, TimeUnit.MILLISECONDS);
+ } catch (TimeoutException e) {
+ // Hold cap reached without a release; carry on and let teardown
proceed.
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new InterruptedIOException(e.toString());
+ } catch (ExecutionException e) {
+ throw new IOException(e);
+ }
+ }
+ }
+}
diff --git
a/solr/core/src/test/org/apache/solr/handler/IndexFetcherPacketProtocolTest.java
b/solr/core/src/test/org/apache/solr/handler/IndexFetcherPacketProtocolTest.java
index 36d47427a4f..5ffd898c690 100644
---
a/solr/core/src/test/org/apache/solr/handler/IndexFetcherPacketProtocolTest.java
+++
b/solr/core/src/test/org/apache/solr/handler/IndexFetcherPacketProtocolTest.java
@@ -30,6 +30,7 @@ import org.apache.lucene.store.Directory;
import org.apache.lucene.store.IOContext;
import org.apache.lucene.store.IndexOutput;
import org.apache.solr.SolrTestCaseJ4;
+import org.apache.solr.client.solrj.jetty.HttpJettySolrClient;
import org.apache.solr.common.params.ModifiableSolrParams;
import org.apache.solr.common.util.FastInputStream;
import org.apache.solr.common.util.NamedList;
@@ -313,7 +314,12 @@ public class IndexFetcherPacketProtocolTest extends
SolrTestCaseJ4 {
IndexFetcher indexFetcher = createIndexFetcher();
- try {
+ try (HttpJettySolrClient fetchClient =
+ new HttpJettySolrClient.Builder()
+ .withHttpClient(
+
h.getCoreContainer().getUpdateShardHandler().getRecoveryOnlyHttpClient())
+ .withAbortableRequests(true)
+ .build()) {
Map<String, Object> fileDetails = new HashMap<>();
fileDetails.put("name", fileName);
fileDetails.put("size", expectedSize);
@@ -322,6 +328,12 @@ public class IndexFetcherPacketProtocolTest extends
SolrTestCaseJ4 {
MethodHandles.Lookup lookup =
MethodHandles.privateLookupIn(fileFetcherClass,
MethodHandles.lookup());
+ Class<?> fetchClass = getInnerClass("Fetch");
+ MethodHandle fetchCtorHandle =
+ MethodHandles.privateLookupIn(fetchClass, MethodHandles.lookup())
+ .findConstructor(
+ fetchClass,
+ MethodType.methodType(void.class, HttpJettySolrClient.class,
long.class));
MethodHandle ctorHandle =
lookup.findConstructor(
fileFetcherClass,
@@ -332,12 +344,14 @@ public class IndexFetcherPacketProtocolTest extends
SolrTestCaseJ4 {
Map.class,
String.class,
String.class,
- long.class));
+ fetchClass));
Object fileFetcher;
try {
+ Object fetch = fetchCtorHandle.invoke(fetchClient, 0L);
fileFetcher =
- ctorHandle.invoke(indexFetcher, mockFileInterface, fileDetails,
fileName, "file", 0L);
+ ctorHandle.invoke(
+ indexFetcher, mockFileInterface, fileDetails, fileName,
"file", fetch);
} catch (Exception e) {
throw e;
} catch (Throwable t) {
@@ -370,12 +384,16 @@ public class IndexFetcherPacketProtocolTest extends
SolrTestCaseJ4 {
}
private Class<?> getFileInterfaceClass() throws Exception {
+ return getInnerClass("FileInterface");
+ }
+
+ private static Class<?> getInnerClass(String simpleName) {
for (Class<?> innerClass : IndexFetcher.class.getDeclaredClasses()) {
- if (innerClass.getSimpleName().equals("FileInterface")) {
+ if (innerClass.getSimpleName().equals(simpleName)) {
return innerClass;
}
}
- throw new AssertionError("FileInterface not found");
+ throw new AssertionError(simpleName + " not found");
}
/** Creates a mock FileInterface that captures written bytes to the provided
output stream. */
diff --git
a/solr/solrj-jetty/src/java/org/apache/solr/client/solrj/jetty/HttpJettySolrClient.java
b/solr/solrj-jetty/src/java/org/apache/solr/client/solrj/jetty/HttpJettySolrClient.java
index cf05fdac07e..f6f2a0718a1 100644
---
a/solr/solrj-jetty/src/java/org/apache/solr/client/solrj/jetty/HttpJettySolrClient.java
+++
b/solr/solrj-jetty/src/java/org/apache/solr/client/solrj/jetty/HttpJettySolrClient.java
@@ -26,8 +26,11 @@ import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Phaser;
@@ -119,6 +122,8 @@ public class HttpJettySolrClient extends HttpSolrClient {
private final List<HttpListenerFactory> listenerFactory;
protected AsyncTracker asyncTracker = new AsyncTracker();
+ // null unless built withAbortableRequests(true)
+ protected AbortTracker abortTracker;
private final boolean closeClient;
private ExecutorService executor;
@@ -130,6 +135,7 @@ public class HttpJettySolrClient extends HttpSolrClient {
protected HttpJettySolrClient(String serverBaseUrl, Builder builder) {
super(serverBaseUrl, builder);
+ this.abortTracker = builder.abortableRequests ? new AbortTracker() : null;
if (builder.httpClient != null) {
// Validate that no conflicting options are provided when using an
existing HttpClient
@@ -660,9 +666,32 @@ public class HttpJettySolrClient extends HttpSolrClient {
req.onComplete(listener);
}
- if (isAsync) {
- req.onRequestQueued(asyncTracker.queuedListener);
- req.onComplete(asyncTracker.completeListener);
+ AsyncTracker asyncTracker = isAsync ? this.asyncTracker : null;
+ AbortTracker abortTracker = this.abortTracker;
+ if (asyncTracker != null || abortTracker != null) {
+ // One listener each, so this order doesn't depend on how Jetty orders
separate listeners.
+ // The async tracker must take its permit before the request can be
aborted: aborting a
+ // queued request completes it synchronously, and a permit taken after
that is never
+ // released. Requests still waiting for a permit get one as aborted
in-flight requests
+ // complete, and are then aborted here.
+ req.onRequestQueued(
+ request -> {
+ if (asyncTracker != null) {
+ asyncTracker.queuedListener.onQueued(request);
+ }
+ if (abortTracker != null) {
+ abortTracker.track(request);
+ }
+ });
+ req.onComplete(
+ result -> {
+ if (asyncTracker != null) {
+ asyncTracker.completeListener.onComplete(result);
+ }
+ if (abortTracker != null) {
+ abortTracker.untrack(result.getRequest());
+ }
+ });
}
}
@@ -839,6 +868,7 @@ public class HttpJettySolrClient extends HttpSolrClient {
super(baseUrl, new
HttpJettySolrClient.Builder(baseUrl).withHttpClient(parentClient));
this.asyncTracker = parentClient.asyncTracker;
+ this.abortTracker = parentClient.abortTracker;
}
@Override
@@ -931,6 +961,68 @@ public class HttpJettySolrClient extends HttpSolrClient {
}
}
+ /**
+ * Tracks a client's requests, synchronous and asynchronous, from queued to
complete, so that
+ * {@link #abort(Throwable)} can reach them.
+ */
+ private static class AbortTracker {
+ // requests queued and not yet complete
+ private final Set<Request> outstanding = ConcurrentHashMap.newKeySet();
+ // non-null once abort() has been called; fails any request queued
afterwards
+ private volatile Throwable abortCause;
+
+ void track(Request request) {
+ // Pairs with abort(): add-then-check here and set-then-scan there, so a
request queued
+ // concurrently with abort() is aborted by at least one of the two.
+ outstanding.add(request);
+ Throwable cause = abortCause;
+ if (cause != null) {
+ request.abort(cause);
+ }
+ }
+
+ void untrack(Request request) {
+ outstanding.remove(request);
+ }
+
+ void abort(Throwable cause) {
+ abortCause = Objects.requireNonNull(cause);
+ outstanding.forEach(request -> request.abort(cause));
+ }
+
+ boolean isAborted() {
+ return abortCause != null;
+ }
+ }
+
+ /**
+ * Aborts every request in flight on this client, and fails any request made
afterwards. Unlike
+ * {@link #close()}, this does not wait: each aborted request completes
exceptionally straight
+ * away, so a subsequent {@link #close()} returns promptly.
+ *
+ * <p>Scoped to this client instance and to the clients derived from it by
{@link
+ * #requestWithBaseUrl}, which share its tracking. Other clients wrapping
the same Jetty {@link
+ * HttpClient} are unaffected, so a lightweight copy made with {@link
+ * Builder#withHttpClient(HttpJettySolrClient)} can be aborted without
disturbing anyone else.
+ *
+ * @param cause reported as the failure of each aborted request, and
rethrown as-is to anyone
+ * reading a streamed response body, so pass an {@link IOException} to
keep {@link
+ * InputStream} callers on their usual error path; must not be null
+ * @throws IllegalStateException if the client was not built with {@link
+ * Builder#withAbortableRequests(boolean)}
+ */
+ public void abort(Throwable cause) {
+ if (abortTracker == null) {
+ throw new IllegalStateException("Client was not built with
withAbortableRequests(true)");
+ }
+ abortTracker.abort(cause);
+ }
+
+ /** Whether {@link #abort(Throwable)} has been called on this client. */
+ public boolean isAborted() {
+ return abortTracker != null && abortTracker.isAborted();
+ }
+
/** Returns the configured maximum number of outstanding async requests. */
public int asyncTrackerMaxPermits() {
return asyncTracker.maxPermits();
@@ -953,6 +1045,8 @@ public class HttpJettySolrClient extends HttpSolrClient {
private List<HttpListenerFactory> listenerFactories;
+ private boolean abortableRequests;
+
public Builder() {
super();
}
@@ -1043,6 +1137,19 @@ public class HttpJettySolrClient extends HttpSolrClient {
return type.cast(build());
}
+ /**
+ * Tracks every request, so that {@link
HttpJettySolrClient#abort(Throwable)} can fail them. Off
+ * by default, since it adds bookkeeping to every request. Can be combined
with {@link
+ * #withHttpClient(HttpJettySolrClient)}; it is not inherited from the
client passed there.
+ *
+ * @param abortableRequests whether this client's requests can be aborted
+ * @return This Builder
+ */
+ public Builder withAbortableRequests(boolean abortableRequests) {
+ this.abortableRequests = abortableRequests;
+ return this;
+ }
+
@Override
public HttpJettySolrClient build() {
return new HttpJettySolrClient(baseSolrUrl, this);
diff --git
a/solr/solrj-jetty/src/test/org/apache/solr/client/solrj/jetty/HttpJettySolrClientTest.java
b/solr/solrj-jetty/src/test/org/apache/solr/client/solrj/jetty/HttpJettySolrClientTest.java
index 6a3be472677..365f9a2b098 100644
---
a/solr/solrj-jetty/src/test/org/apache/solr/client/solrj/jetty/HttpJettySolrClientTest.java
+++
b/solr/solrj-jetty/src/test/org/apache/solr/client/solrj/jetty/HttpJettySolrClientTest.java
@@ -24,6 +24,11 @@ import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.util.Base64;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
@@ -50,9 +55,14 @@ import org.apache.solr.common.SolrException;
import org.apache.solr.common.params.CommonParams;
import org.apache.solr.common.params.ModifiableSolrParams;
import org.apache.solr.common.params.SolrParams;
+import org.apache.solr.common.util.ExecutorUtil;
import org.apache.solr.common.util.NamedList;
+import org.apache.solr.common.util.SolrNamedThreadFactory;
+import org.apache.solr.common.util.TimeSource;
import org.apache.solr.util.ServletFixtures;
import org.apache.solr.util.ServletFixtures.DebugServlet;
+import org.apache.solr.util.ServletFixtures.SlowServlet;
+import org.apache.solr.util.TimeOut;
import org.eclipse.jetty.client.WWWAuthenticationProtocolHandler;
import org.eclipse.jetty.http.HttpStatus;
import org.eclipse.jetty.util.ssl.SslContextFactory;
@@ -698,6 +708,154 @@ public class HttpJettySolrClientTest extends
HttpSolrClientTestBase {
}
}
+ /**
+ * How soon an aborted request must fail: well before the slow servlet would
have answered on its
+ * own, so passing proves the abort, not the servlet, ended it.
+ */
+ private static final long ABORT_DEADLINE_MS = SlowServlet.DELAY_MS / 2;
+
+ @Test
+ public void testAbortFailsAsyncRequestInFlight() throws Exception {
+ try (var client =
+ new HttpJettySolrClient.Builder(solrTestRule.getBaseUrl() +
SLOW_SERVLET_PATH)
+ .withAbortableRequests(true)
+ .build()) {
+ CompletableFuture<Void> arrived = new CompletableFuture<>();
+ CompletableFuture<NamedList<Object>> future =
client.requestAsync(stallRequest(arrived));
+ arrived.get(SlowServlet.DELAY_MS, TimeUnit.MILLISECONDS);
+ client.abort(new IOException("test abort"));
+ expectThrows(
+ ExecutionException.class, () -> future.get(ABORT_DEADLINE_MS,
TimeUnit.MILLISECONDS));
+ assertAllRequestsComplete(client);
+ }
+ }
+
+ @Test
+ public void testAbortFailsResponseMidBody() throws Exception {
+ try (var client =
+ new HttpJettySolrClient.Builder(solrTestRule.getBaseUrl() +
SLOW_SERVLET_PATH)
+ .withAbortableRequests(true)
+ .build()) {
+ QueryRequest req = new
QueryRequest(SolrParams.of(SlowServlet.FIRST_BYTE_PARAM, "true"));
+ req.setResponseParser(new InputStreamResponseParser(FILE_STREAM));
+ NamedList<Object> response = client.requestAsync(req).get(10,
TimeUnit.SECONDS);
+ try (InputStream is = (InputStream) response.get("stream")) {
+ assertEquals('0', is.read());
+ // The servlet now sends nothing more; the abort must wake the read
rather than wait it out.
+ client.abort(new IOException("test abort"));
+ long start = System.nanoTime();
+ IOException e = expectThrows(IOException.class, is::read);
+ long elapsedMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() -
start);
+ assertEquals("test abort", e.getMessage());
+ assertTrue("read took " + elapsedMs + "ms after abort", elapsedMs <
ABORT_DEADLINE_MS);
+ }
+ assertAllRequestsComplete(client);
+ }
+ }
+
+ @Test
+ public void testAbortFailsLaterAsyncRequests() throws Exception {
+ try (var client =
+ new HttpJettySolrClient.Builder(solrTestRule.getBaseUrl() +
SLOW_STREAM_SERVLET_PATH)
+ .withAbortableRequests(true)
+ .build()) {
+ client.abort(new IOException("test abort"));
+ assertTrue(client.isAborted());
+ QueryRequest req = new QueryRequest(SolrParams.of("count", "1"));
+ req.setResponseParser(new InputStreamResponseParser(FILE_STREAM));
+ CompletableFuture<NamedList<Object>> future = client.requestAsync(req);
+ expectThrows(
+ ExecutionException.class, () -> future.get(ABORT_DEADLINE_MS,
TimeUnit.MILLISECONDS));
+ assertAllRequestsComplete(client);
+ }
+ }
+
+ @Test
+ public void testAbortIsScopedToOneClient() throws Exception {
+ String url = solrTestRule.getBaseUrl() + SLOW_STREAM_SERVLET_PATH;
+ try (var base = new HttpJettySolrClient.Builder(url).build();
+ var copy =
+ new HttpJettySolrClient.Builder(url)
+ .withHttpClient(base)
+ .withAbortableRequests(true)
+ .build()) {
+ copy.abort(new IOException("test abort"));
+ assertTrue(copy.isAborted());
+ assertFalse(base.isAborted());
+ // The copy shares base's Jetty client, but not its tracker, so base
still works.
+ QueryRequest req = new QueryRequest(SolrParams.of("count", "1"));
+ req.setResponseParser(new InputStreamResponseParser(FILE_STREAM));
+ NamedList<Object> response = base.requestAsync(req).get(10,
TimeUnit.SECONDS);
+ try (InputStream is = (InputStream) response.get("stream")) {
+ assertEquals("0", new String(is.readAllBytes(),
StandardCharsets.UTF_8));
+ }
+ }
+ }
+
+ @Test
+ public void testAbortFailsSyncRequestInFlight() throws Exception {
+ ExecutorService executor =
+ ExecutorUtil.newMDCAwareSingleThreadExecutor(new
SolrNamedThreadFactory("syncRequest"));
+ try (var client =
+ new HttpJettySolrClient.Builder(solrTestRule.getBaseUrl() +
SLOW_SERVLET_PATH)
+ .withAbortableRequests(true)
+ .build()) {
+ CompletableFuture<Void> arrived = new CompletableFuture<>();
+ QueryRequest req = stallRequest(arrived);
+ Future<NamedList<Object>> call = executor.submit(() ->
client.request(req));
+ arrived.get(SlowServlet.DELAY_MS, TimeUnit.MILLISECONDS);
+ client.abort(new IOException("test abort"));
+ ExecutionException e =
+ expectThrows(
+ ExecutionException.class, () -> call.get(ABORT_DEADLINE_MS,
TimeUnit.MILLISECONDS));
+ assertTrue(e.getCause().toString(), e.getCause() instanceof
SolrServerException);
+ } finally {
+ ExecutorUtil.shutdownAndAwaitTermination(executor);
+ }
+ }
+
+ @Test
+ public void testAbortFailsLaterSyncRequests() {
+ try (var client =
+ new HttpJettySolrClient.Builder(solrTestRule.getBaseUrl() +
SLOW_STREAM_SERVLET_PATH)
+ .withAbortableRequests(true)
+ .build()) {
+ client.abort(new IOException("test abort"));
+ QueryRequest req = new QueryRequest(SolrParams.of("count", "1"));
+ req.setResponseParser(new InputStreamResponseParser(FILE_STREAM));
+ expectThrows(SolrServerException.class, () -> client.request(req));
+ }
+ }
+
+ @Test
+ public void testAbortRequiresOptIn() {
+ try (var client = new
HttpJettySolrClient.Builder(solrTestRule.getBaseUrl()).build()) {
+ expectThrows(IllegalStateException.class, () -> client.abort(new
IOException("test abort")));
+ assertFalse(client.isAborted());
+ }
+ }
+
+ /**
+ * Asserts every async request has completed, i.e. close() would not wait.
Aborted requests
+ * complete straight away; without the abort these would stay open until the
servlet answers.
+ */
+ private static void assertAllRequestsComplete(HttpJettySolrClient client)
throws Exception {
+ new TimeOut(ABORT_DEADLINE_MS, TimeUnit.MILLISECONDS, TimeSource.NANO_TIME)
+ .waitFor(
+ "async requests still outstanding after abort",
+ () -> client.asyncTrackerAvailablePermits() ==
client.asyncTrackerMaxPermits());
+ }
+
+ /**
+ * A request to the stall servlet that completes {@code arrived} once it is
parked there, so the
+ * test can abort it in flight without guessing how long it takes to get
there.
+ */
+ private static QueryRequest stallRequest(CompletableFuture<Void> arrived) {
+ String arrivalId = UUID.randomUUID().toString();
+ SlowServlet.ARRIVALS.put(arrivalId, arrived);
+ return new QueryRequest(SolrParams.of(SlowServlet.ARRIVAL_PARAM,
arrivalId));
+ }
+
@Test
public void testIdleTimeoutWithHttpClient() throws Exception {
String url = solrTestRule.getBaseUrl() + SLOW_STREAM_SERVLET_PATH;
diff --git
a/solr/test-framework/src/java/org/apache/solr/util/ServletFixtures.java
b/solr/test-framework/src/java/org/apache/solr/util/ServletFixtures.java
index cb1b66f6eee..97d6af67c27 100644
--- a/solr/test-framework/src/java/org/apache/solr/util/ServletFixtures.java
+++ b/solr/test-framework/src/java/org/apache/solr/util/ServletFixtures.java
@@ -35,6 +35,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
@@ -119,12 +120,33 @@ public class ServletFixtures {
}
}
+ /**
+ * Sends nothing for {@link #DELAY_MS}: a server gone quiet. With {@link
#FIRST_BYTE_PARAM} it
+ * sends and flushes one byte of body first, so the silence is mid-response.
A request carrying
+ * {@link #ARRIVAL_PARAM} completes the matching future in {@link #ARRIVALS}
once it is parked
+ * here, so a test can act on it without guessing how long it takes to
arrive.
+ */
public static class SlowServlet extends HttpServlet {
+
+ public static final int DELAY_MS = 5000;
+ public static final String FIRST_BYTE_PARAM = "firstByte";
+ public static final String ARRIVAL_PARAM = "arrivalId";
+ public static final Map<String, CompletableFuture<Void>> ARRIVALS = new
ConcurrentHashMap<>();
+
+ @SuppressForbidden(reason = "forbiddenApis: getParameter is fine in tests")
@Override
- protected void doGet(HttpServletRequest req, HttpServletResponse resp)
- throws ServletException, IOException {
+ protected void doGet(HttpServletRequest req, HttpServletResponse resp)
throws IOException {
+ if (Boolean.parseBoolean(req.getParameter(FIRST_BYTE_PARAM))) {
+ resp.getOutputStream().write('0');
+ resp.getOutputStream().flush();
+ }
+ String arrivalId = req.getParameter(ARRIVAL_PARAM);
+ CompletableFuture<Void> arrival = arrivalId == null ? null :
ARRIVALS.remove(arrivalId);
+ if (arrival != null) {
+ arrival.complete(null);
+ }
try {
- Thread.sleep(5000);
+ Thread.sleep(DELAY_MS);
} catch (InterruptedException ignored) {
}
}
