SOLR-18406: Restart replication after index generation expires

solr/core/src/java/org/apache/solr/handler/IndexFetcher.java

@@ -405,6 +405,18 @@ IndexFetchResult fetchLatestIndex(boolean forceReplication)
    */
   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 @@ IndexFetchResult fetchLatestIndex(boolean forceReplication, boolean forceCoreRel
       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 @@ IndexFetchResult fetchLatestIndex(boolean forceReplication, boolean forceCoreRel
       } 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 void fetchFile() throws Exception {
       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 void fetchFile() throws Exception {
     }
 
     private void fetch() throws Exception {
+      boolean invalidIndexGeneration = false;
       try {
         while (true) {
           try (FastInputStream fis = getStream()) {
@@ -1617,17 +1635,22 @@ private void fetch() throws Exception {
             // 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 @@ private long readLong(byte[] b) {
     }
 
     /** cleanup everything */
-    private void cleanup() {
+    private void cleanup(boolean invalidIndexGeneration) {
       try {
         file.close();
       } catch (Exception e) {
@@ -1760,7 +1783,7 @@ private void cleanup() {
           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,20 +1829,27 @@ private FastInputStream getStream() throws IOException {
         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(
                   Locale.ROOT,
                   "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 @@ private IOException closeStreamAndBuildIOE(
     }
   }
 
+  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;

solr/core/src/java/org/apache/solr/handler/admin/api/ReplicationAPIBase.java

@@ -254,6 +254,7 @@ protected class DirectoryFileStream implements SolrCore.RawWriter, StreamingOutp
 
     protected Long indexGen;
     protected IndexDeletionPolicyWrapper delPolicy;
+    private boolean commitPointSaved;
 
     protected String fileName;
     protected String cfileName;
@@ -342,7 +343,13 @@ protected void initWrite() throws IOException {
 
       // 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 @@ protected void extendReserveAndReleaseCommitPoint() {
       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.

solr/core/src/test/org/apache/solr/handler/TestReplicationHandler.java

@@ -34,6 +34,9 @@
 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.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;
@@ -1494,6 +1499,63 @@ public void testFileListShouldReportErrorsWhenTheyOccur() throws Exception {
     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();