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 1031b8ee4394df89aa7122e5be0b343099d12c79 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) --- .../SOLR-18406-replication-stale-generation.yml | 10 ++++ .../java/org/apache/solr/handler/IndexFetcher.java | 65 +++++++++++++++++----- .../solr/handler/admin/api/ReplicationAPIBase.java | 11 +++- .../solr/handler/TestReplicationHandler.java | 62 +++++++++++++++++++++ 4 files changed, 133 insertions(+), 15 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 4a9f14834cd..b6fa1146959 100644 --- a/solr/core/src/java/org/apache/solr/handler/IndexFetcher.java +++ b/solr/core/src/java/org/apache/solr/handler/IndexFetcher.java @@ -405,6 +405,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 { boolean cleanupDone = false; boolean successfulInstall = false; @@ -556,6 +568,7 @@ public class IndexFetcher { log.info("Starting replication process"); // get the list of files first fetchFileList(latestGeneration); + assert testWait.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; @@ -761,6 +774,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) { @@ -1593,6 +1608,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); @@ -1605,6 +1622,7 @@ public class IndexFetcher { } private void fetch() throws Exception { + boolean invalidIndexGeneration = false; try { while (true) { try (FastInputStream fis = getStream()) { @@ -1617,17 +1635,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; + } + }); + } } } @@ -1744,7 +1767,7 @@ public class IndexFetcher { } /** cleanup everything */ - private void cleanup() { + private void cleanup(boolean invalidIndexGeneration) { try { file.close(); } catch (Exception e) { @@ -1760,7 +1783,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 " @@ -1806,6 +1829,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( @@ -1813,13 +1840,16 @@ public class IndexFetcher { "Unexpected status code [%d] when downloading file [%s].", responseStatus, fileName); - closeStreamAndBuildIOE(is, errorMsg, null); + throw closeStreamAndBuildIOE(is, errorMsg, null); } if (useInternalCompression) { is = new InflaterInputStream(is); } return new FastInputStream(is); + } catch (InvalidIndexGenerationException e) { + IOUtils.closeQuietly(is); + throw e; } catch (Exception e) { final var ioe = closeStreamAndBuildIOE(is, "Could not download file '" + fileName + "'", e); throw ioe; @@ -1836,6 +1866,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 3b7170494c5..8dbf4afa911 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 @@ -254,6 +254,7 @@ public abstract class ReplicationAPIBase extends JerseyResource { protected Long indexGen; protected IndexDeletionPolicyWrapper delPolicy; + private boolean commitPointSaved; protected String fileName; protected String cfileName; @@ -342,7 +343,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); + } } } @@ -360,7 +367,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 326e0ac5059..cc719d054c4 100644 --- a/solr/core/src/test/org/apache/solr/handler/TestReplicationHandler.java +++ b/solr/core/src/test/org/apache/solr/handler/TestReplicationHandler.java @@ -34,6 +34,9 @@ import java.util.Arrays; 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 java.util.stream.Stream; import org.apache.lucene.index.DirectoryReader; @@ -65,7 +68,9 @@ import org.apache.solr.common.SolrDocument; import org.apache.solr.common.SolrDocumentList; import org.apache.solr.common.SolrException; import org.apache.solr.common.params.ModifiableSolrParams; +import org.apache.solr.common.util.ExecutorUtil; import org.apache.solr.common.util.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; @@ -1493,6 +1498,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.testWait = + () -> { + 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.testWait = () -> true; + continueDownload.countDown(); + followerFetch.get(TIMEOUT, TimeUnit.MILLISECONDS); + + assertEquals(1, numFound(rQuery(1, "name:generation-g-plus-one", followerClient))); + } finally { + IndexFetcher.testWait = () -> true; + continueDownload.countDown(); + workload.shutdownNow(); + } + } + @Test public void testFetchIndexShouldReportErrorsWhenTheyOccur() throws Exception { int leaderPort = leaderJetty.getLocalPort();
