This is an automated email from the ASF dual-hosted git repository. dsmiley pushed a commit to branch branch_9x in repository https://gitbox.apache.org/repos/asf/solr.git
commit 30b3cddd8ed6df6f4d13372ef2312605beeb32e8 Author: Zhenyu Li <[email protected]> AuthorDate: Mon Sep 7 23:04:40 2026 -0400 SOLR-18406: Restart replication after index generation expires (#4820) Leader/follower replication now restarts with the latest index generation when the selected generation expires during download, avoiding retries against an unavailable generation. (cherry picked from commit 0cc72b8e326d9702958e9603b420af4c60d24c19) (cherry picked from commit 1031b8ee4394df89aa7122e5be0b343099d12c79) --- .../SOLR-18406-replication-stale-generation.yml | 10 ++++ .../java/org/apache/solr/handler/IndexFetcher.java | 64 ++++++++++++++++++---- .../solr/handler/admin/api/ReplicationAPIBase.java | 11 +++- .../solr/handler/TestReplicationHandler.java | 62 +++++++++++++++++++++ 4 files changed, 133 insertions(+), 14 deletions(-) diff --git a/changelog/unreleased/SOLR-18406-replication-stale-generation.yml b/changelog/unreleased/SOLR-18406-replication-stale-generation.yml new file mode 100644 index 00000000000..694dd5c4b08 --- /dev/null +++ b/changelog/unreleased/SOLR-18406-replication-stale-generation.yml @@ -0,0 +1,10 @@ +title: > + Leader/follower replication now restarts with the latest index generation when the selected + generation expires during download, avoiding retries against an unavailable generation. +type: changed +authors: + - name: ZhenyuLi + nick: JHSUYU +links: + - name: SOLR-18406 + url: https://issues.apache.org/jira/browse/SOLR-18406 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 eae68ceac39..7c77a599492 100644 --- a/solr/core/src/java/org/apache/solr/handler/IndexFetcher.java +++ b/solr/core/src/java/org/apache/solr/handler/IndexFetcher.java @@ -413,6 +413,18 @@ public class IndexFetcher { */ IndexFetchResult fetchLatestIndex(boolean forceReplication, boolean forceCoreReload) throws IOException, InterruptedException { + try { + return fetchLatestIndexOnce(forceReplication, forceCoreReload); + } catch (InvalidIndexGenerationException e) { + log.info( + "Leader no longer has index generation {}; restarting replication from its latest generation", + e.generation); + return fetchLatestIndexOnce(forceReplication, forceCoreReload); + } + } + + private IndexFetchResult fetchLatestIndexOnce(boolean forceReplication, boolean forceCoreReload) + throws IOException, InterruptedException { this.clearLocalIndexFirst = false; boolean cleanupDone = false; @@ -567,6 +579,7 @@ public class IndexFetcher { log.info("Starting replication process"); // get the list of files first fetchFileList(latestGeneration); + assert testWaitAfterFileList.getAsBoolean(); // this can happen if the commit point is deleted before we fetch the file list. if (filesToDownload.isEmpty()) { return IndexFetchResult.PEER_INDEX_COMMIT_DELETED; @@ -776,6 +789,8 @@ public class IndexFetcher { } catch (ReplicationHandlerException e) { log.error("User aborted Replication", e); return new IndexFetchResult(IndexFetchResult.FAILED_BY_EXCEPTION_MESSAGE, false, e); + } catch (InvalidIndexGenerationException e) { + throw e; } catch (SolrException e) { throw e; } catch (InterruptedException e) { @@ -1208,6 +1223,7 @@ public class IndexFetcher { // only for testing purposes. do not use this anywhere else // -----------START---------------------- static BooleanSupplier testWait = () -> true; + static BooleanSupplier testWaitAfterFileList = () -> true; static Function<String, Long> usableDiskSpaceProvider = dir -> getUsableSpace(dir); // ------------ END--------------------- @@ -1751,6 +1767,8 @@ public class IndexFetcher { bytesDownloaded = 0; try { fetch(); + } catch (InvalidIndexGenerationException e) { + throw e; } catch (Exception e) { if (!aborted) { IndexFetcher.log.error("Error fetching file, doing one retry...", e); @@ -1763,6 +1781,7 @@ public class IndexFetcher { } private void fetch() throws Exception { + boolean invalidIndexGeneration = false; try { while (true) { try (FastInputStream fis = getStream()) { @@ -1775,17 +1794,22 @@ public class IndexFetcher { // if there is an error continue. But continue from the point where it got broken } } + } catch (InvalidIndexGenerationException e) { + invalidIndexGeneration = true; + throw e; } finally { - cleanup(); - // if cleanup succeeds, and the file is downloaded fully, then do a fsync. - fsyncService.execute( - () -> { - try { - file.sync(); - } catch (IOException | AlreadyClosedException e) { - fsyncException = e; - } - }); + cleanup(invalidIndexGeneration); + if (!invalidIndexGeneration) { + // if cleanup succeeds, and the file is downloaded fully, then do a fsync. + fsyncService.execute( + () -> { + try { + file.sync(); + } catch (IOException | AlreadyClosedException e) { + fsyncException = e; + } + }); + } } } @@ -1902,7 +1926,7 @@ public class IndexFetcher { } /** cleanup everything */ - private void cleanup() { + private void cleanup(boolean invalidIndexGeneration) { try { file.close(); } catch (Exception e) { @@ -1918,7 +1942,7 @@ public class IndexFetcher { log.error("Error deleting file: {}", this.saveAs, e); } // if the failure is due to a user abort it is returned normally else an exception is thrown - if (!aborted) + if (!aborted && !invalidIndexGeneration) throw new SolrException( SolrException.ErrorCode.SERVER_ERROR, "Unable to download " @@ -1966,6 +1990,10 @@ public class IndexFetcher { final var responseStatus = (Integer) response.get("responseStatus"); is = (InputStream) response.get("stream"); + if (responseStatus == ErrorCode.CONFLICT.code) { + throw new InvalidIndexGenerationException(indexGen); + } + if (responseStatus != 200) { final var errorMsg = String.format( @@ -1980,6 +2008,9 @@ public class IndexFetcher { is = new InflaterInputStream(is); } return new FastInputStream(is); + } catch (InvalidIndexGenerationException e) { + IOUtils.closeQuietly(is); + throw e; } catch (Exception e) { closeStreamAndThrowIOE(is, "Could not download file '" + fileName + "'", Optional.of(e)); } @@ -1998,6 +2029,15 @@ public class IndexFetcher { } } + private static class InvalidIndexGenerationException extends IOException { + private final long generation; + + InvalidIndexGenerationException(long generation) { + super("Leader no longer has index generation " + generation); + this.generation = generation; + } + } + private static class DirectoryFile implements FileInterface { private final String saveAs; private Directory copy2Dir; diff --git a/solr/core/src/java/org/apache/solr/handler/admin/api/ReplicationAPIBase.java b/solr/core/src/java/org/apache/solr/handler/admin/api/ReplicationAPIBase.java index 132448d1c27..69aab484ec0 100644 --- a/solr/core/src/java/org/apache/solr/handler/admin/api/ReplicationAPIBase.java +++ b/solr/core/src/java/org/apache/solr/handler/admin/api/ReplicationAPIBase.java @@ -256,6 +256,7 @@ public abstract class ReplicationAPIBase extends JerseyResource { protected Long indexGen; protected IndexDeletionPolicyWrapper delPolicy; + private boolean commitPointSaved; protected String fileName; protected String cfileName; @@ -344,7 +345,13 @@ public abstract class ReplicationAPIBase extends JerseyResource { // reserve commit point till write is complete if (indexGen != null) { - delPolicy.saveCommitPoint(indexGen); + try { + delPolicy.saveCommitPoint(indexGen); + commitPointSaved = true; + } catch (IllegalStateException e) { + throw new SolrException( + SolrException.ErrorCode.CONFLICT, "invalid index generation: " + indexGen, e); + } } } @@ -362,7 +369,7 @@ public abstract class ReplicationAPIBase extends JerseyResource { ReplicationHandler replicationHandler = (ReplicationHandler) solrCore.getRequestHandler(ReplicationHandler.PATH); - if (indexGen != null) { + if (commitPointSaved) { // Reserve the commit point for another 10s for the next file to be to fetched. // We need to keep extending the commit reservation between requests so that the replica can // fetch all the files correctly. diff --git a/solr/core/src/test/org/apache/solr/handler/TestReplicationHandler.java b/solr/core/src/test/org/apache/solr/handler/TestReplicationHandler.java index 49a21656ae6..5c197ec698a 100644 --- a/solr/core/src/test/org/apache/solr/handler/TestReplicationHandler.java +++ b/solr/core/src/test/org/apache/solr/handler/TestReplicationHandler.java @@ -36,6 +36,9 @@ import java.util.Collection; import java.util.Date; import java.util.List; import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import org.apache.lucene.index.DirectoryReader; import org.apache.lucene.index.IndexReader; @@ -64,7 +67,9 @@ 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.core.CachingDirectoryFactory; import org.apache.solr.core.CoreContainer; @@ -1547,6 +1552,63 @@ public class TestReplicationHandler extends SolrTestCaseJ4 { assertEquals("invalid index generation", resp.get("message")); } + @Test + public void testFollowerRestartsWhenCommitExpiresBeforeFileDownload() throws Exception { + invokeReplicationCommand( + buildUrl(followerJetty.getLocalPort()) + "/" + DEFAULT_TEST_CORENAME, "disablepoll"); + + index(leaderClient, "id", "1", "name", "generation-g"); + leaderClient.commit(); + index(leaderClient, "id", "1", "name", "generation-g-plus-one"); + + CountDownLatch fileListFetched = new CountDownLatch(1); + CountDownLatch continueDownload = new CountDownLatch(1); + IndexFetcher.testWaitAfterFileList = + () -> { + fileListFetched.countDown(); + try { + continueDownload.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException(e); + } + return true; + }; + + ExecutorService workload = + ExecutorUtil.newMDCAwareSingleThreadExecutor( + new SolrNamedThreadFactory("staleGenerationWorkload")); + try { + Future<?> followerFetch = + workload.submit( + () -> { + pullFromTo(leaderJetty, followerJetty); + return null; + }); + + assertTrue(fileListFetched.await(TIMEOUT, TimeUnit.MILLISECONDS)); + + long reserveDuration; + try (SolrCore core = leaderJetty.getCoreContainer().getCore(DEFAULT_TEST_CORENAME)) { + ReplicationHandler handler = + (ReplicationHandler) core.getRequestHandler(ReplicationHandler.PATH); + reserveDuration = handler.getReserveCommitDuration(); + } + Thread.sleep(reserveDuration + 1000); + leaderClient.commit(); + + IndexFetcher.testWaitAfterFileList = () -> true; + continueDownload.countDown(); + followerFetch.get(TIMEOUT, TimeUnit.MILLISECONDS); + + assertEquals(1, numFound(rQuery(1, "name:generation-g-plus-one", followerClient))); + } finally { + IndexFetcher.testWaitAfterFileList = () -> true; + continueDownload.countDown(); + workload.shutdownNow(); + } + } + @Test public void testFetchIndexShouldReportErrorsWhenTheyOccur() throws Exception { int leaderPort = leaderJetty.getLocalPort();
