[dao-jun]

What changes were proposed in this pull request?

gate stale read when snapshots installing

What is the link to the Apache JIRA

https://issues.apache.org/jira/projects/RATIS/issues/RATIS-2795

How was this patch tested?

unit tests

[amaliujia]

Can we also add a config to guard this improvement and by default turn it off?

[szetszwo]

@dao-jun , thanks for working on this!

• ReadException currently has readIndex in the message. Let's change it.
• Let's also add an op parameter to getReadExceptionIfInstallingSnapshot. Then we can use it for stale read.

diff --git a/ratis-server/src/main/java/org/apache/ratis/server/impl/ReadRequestHandler.java b/ratis-server/src/main/java/org/apache/ratis/server/impl/ReadRequestHandler.java
index 9b961c577..7d93d9ce3 100644
--- a/ratis-server/src/main/java/org/apache/ratis/server/impl/ReadRequestHandler.java
+++ b/ratis-server/src/main/java/org/apache/ratis/server/impl/ReadRequestHandler.java
@@ -81,6 +81,11 @@ class ReadRequestHandler {
   }
 
   CompletableFuture<RaftClientReply> staleReadAsync(RaftClientRequest request) {
+    final ReadException ex = getReadExceptionIfInstallingSnapshot("stale read");
+    if (ex != null) {
+      return CompletableFuture.completedFuture(
+          server.newExceptionReply(request, new StateMachineException(getMemberId(), ex)));
+    }
     final long minIndex = request.getType().getStaleRead().getMinIndex();
     final long commitIndex = state.getLog().getLastCommittedIndex();
     LOG.debug("{}: minIndex={}, commitIndex={} from {}", getMemberId(), minIndex, commitIndex, request.getClientId());
@@ -94,19 +99,20 @@ class ReadRequestHandler {
   }
 
   ReadException getReadException(String op, long installSnapshot, boolean started) {
-    return installSnapshot == RaftLog.INVALID_LOG_INDEX ? null : new ReadException(getMemberId() + ": Failed to " + op
-        + " readIndex as snapshot (" + installSnapshot + ") installation is " + (started ? "started" : "in progress"));
+    return installSnapshot == RaftLog.INVALID_LOG_INDEX ? null : new ReadException(getMemberId() + ": Failed " + op
+        + " as snapshot (" + installSnapshot + ") installation is " + (started ? "started" : "in progress"));
   }
 
-  private ReadException getReadExceptionIfInstallingSnapshot() {
-    return getReadException("add", server.getSnapshotInstallationHandler().getInProgressInstallSnapshotIndex(), false);
+  private ReadException getReadExceptionIfInstallingSnapshot(String op) {
+    final long installSnapshotIndex = server.getSnapshotInstallationHandler().getInProgressInstallSnapshotIndex();
+    return getReadException(op, installSnapshotIndex, false);
   }
 
   private CompletableFuture<ReadIndexReplyProto> sendReadIndexAsync(
       ClientId clientId, ReadRequestTypeProto readRequestType) {
-    final long installSnapshot = server.getSnapshotInstallationHandler().getInProgressInstallSnapshotIndex();
-    if (installSnapshot != RaftLog.INVALID_LOG_INDEX) {
-      return JavaUtils.completeExceptionally(getReadException("get", installSnapshot, false));
+    final ReadException readException = getReadExceptionIfInstallingSnapshot("to send readIndex request");
+    if (readException != null) {
+      return JavaUtils.completeExceptionally(readException);
     }
     final RaftPeerId leaderId = server.getInfo().getLeaderId();
     if (leaderId == null) {
@@ -154,7 +160,7 @@ class ReadRequestHandler {
       final LeaderStateImpl leader = server.getRole().getLeaderState().orElse(null);
       return getReadIndex(clientId, readRequestType, leader)
           .thenCompose(readIndex -> state.getReadRequests().waitToAdvance(readIndex,
-              this::getReadExceptionIfInstallingSnapshot))
+              () -> getReadExceptionIfInstallingSnapshot("local read")))
           .thenCompose(readIndex -> stateMachine.query(queryMessage));
     } else {
       throw new IllegalStateException("Unexpected read option: " + readOption);
@@ -177,7 +183,7 @@ class ReadRequestHandler {
 
       return replyFuture
           .thenCompose(readIndex -> state.getReadRequests().waitToAdvance(readIndex,
-              this::getReadExceptionIfInstallingSnapshot))
+              () -> getReadExceptionIfInstallingSnapshot("read")))
           .thenCompose(readIndex -> isDummyRead(request)
               ? CompletableFuture.completedFuture(server.newSuccessReply(request)) : queryStateMachine(request))
           .exceptionally(e -> readException2Reply(request, e));
diff --git a/ratis-server/src/main/java/org/apache/ratis/server/impl/SnapshotInstallationHandler.java b/ratis-server/src/main/java/org/apache/ratis/server/impl/SnapshotInstallationHandler.java
index d34e84c3f..4e23d30c7 100644
--- a/ratis-server/src/main/java/org/apache/ratis/server/impl/SnapshotInstallationHandler.java
+++ b/ratis-server/src/main/java/org/apache/ratis/server/impl/SnapshotInstallationHandler.java
@@ -43,6 +43,7 @@ import org.apache.ratis.util.BatchLogger;
 import org.apache.ratis.util.CodeInjectionForTesting;
 import org.apache.ratis.util.LifeCycle;
 import org.apache.ratis.util.Preconditions;
+import org.apache.ratis.thirdparty.com.google.common.annotations.VisibleForTesting;
 import org.apache.ratis.util.ProtoUtils;
 import org.apache.ratis.util.Timestamp;
 import org.slf4j.Logger;
@@ -106,6 +107,11 @@ class SnapshotInstallationHandler {
     return inProgressInstallSnapshotIndex.get();
   }
 
+  @VisibleForTesting
+  void setInProgressInstallSnapshotIndex(long index) {
+    inProgressInstallSnapshotIndex.set(index);
+  }
+
   InstallSnapshotReplyProto installSnapshot(InstallSnapshotRequestProto request) throws IOException {
     BatchLogger.print(BatchLogKey.INSTALL_SNAPSHOT_REQUEST, getMemberId(),
         suffix -> LOG.info("{}: receive installSnapshot: {} {}",
@@ -289,7 +295,7 @@ class SnapshotInstallationHandler {
           return future.thenApply(dummy -> reply);
         }
         server.getState().getReadRequests().fail(
-            server.getReadRequestHandler().getReadException("wait for", firstAvailableLogIndex, true));
+            server.getReadRequestHandler().getReadException("to wait for readIndex", firstAvailableLogIndex, true));
 
         final RaftPeerProto leaderProto;
         if (!request.hasLastRaftConfigurationLogEntryProto()) {

[dao-jun]

@amaliujia @szetszwo Thanks for your review! comments addressed

[szetszwo]

@dao-jun , thanks for the update!

+1 the change looks good.