This is an automated email from the ASF dual-hosted git repository. dsmiley pushed a commit to branch branch_10x in repository https://gitbox.apache.org/repos/asf/solr.git
commit 3810d0537ee5d2f5818a6ad4220e4e5bdc70a0d8 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) (cherry picked from commit 489237e71c0fe0c91756be701d4b4c354c1c32b1) --- 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 64ad2dadb2e..8db52035c27 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 af159f58a6a..056218cef9c 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 @@ -27,8 +27,11 @@ import java.util.Collection; 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; @@ -122,6 +125,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; @@ -133,6 +138,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 @@ -663,9 +669,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()); + } + }); } } @@ -835,6 +864,7 @@ public class HttpJettySolrClient extends HttpSolrClient { super(baseUrl, new HttpJettySolrClient.Builder(baseUrl).withHttpClient(parentClient)); this.asyncTracker = parentClient.asyncTracker; + this.abortTracker = parentClient.abortTracker; } @Override @@ -927,6 +957,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(); @@ -949,6 +1041,8 @@ public class HttpJettySolrClient extends HttpSolrClient { private List<HttpListenerFactory> listenerFactories; + private boolean abortableRequests; + public Builder() { super(); } @@ -1039,6 +1133,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 11bf3e3c85a..a47e981be5b 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; @@ -48,9 +53,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; @@ -667,6 +677,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) { } }
