RATIS-2379. Support returning applied index for ReadIndex

ratis-docs/src/site/markdown/configurations.md

@@ -218,6 +218,13 @@ if it fails to receive any RPC responses from this peer within this specified ti
 | **Type**        | TimeDuration                                                                   |
 | **Default**     | 60s                                                                            |
 
+### Read Index - Configurations related to ReadIndex used in linearizable read 
+
+| **Property**    | `raft.server.read.read-index.applied-index.enabled`                   |
+|:----------------|:----------------------------------------------------------------------|
+| **Description** | whether applied index (instead of commit index) is used for ReadIndex |
+| **Type**        | boolean                                                               |
+| **Default**     | false                                                                 |
 
 
 ### Write - Configurations related to write requests.

ratis-server-api/src/main/java/org/apache/ratis/server/RaftServerConfigKeys.java

@@ -265,6 +265,21 @@ static void setWriteIndexCacheExpiryTime(RaftProperties properties, TimeDuration
         setTimeDuration(properties::setTimeDuration, WRITE_INDEX_CACHE_EXPIRY_TIME_KEY, expiryTime);
       }
     }
+
+    interface ReadIndex {
+      String PREFIX = Read.PREFIX + ".read-index";
+
+      String APPLIED_INDEX_ENABLED_KEY = PREFIX + ".applied-index.enabled";
+      boolean APPLIED_INDEX_ENABLED_DEFAULT = false;
+      static boolean appliedIndexEnabled(RaftProperties properties) {
+        return getBoolean(properties::getBoolean, APPLIED_INDEX_ENABLED_KEY,
+            APPLIED_INDEX_ENABLED_DEFAULT, getDefaultLog());
+      }
+
+      static void setAppliedIndexEnabled(RaftProperties properties, boolean enabled) {
+        setBoolean(properties::setBoolean, APPLIED_INDEX_ENABLED_KEY, enabled);
+      }
+    }
   }
 
   interface Write {

ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderStateImpl.java

@@ -353,6 +353,7 @@ boolean isApplied() {
   private final PendingStepDown pendingStepDown;
 
   private final ReadIndexHeartbeats readIndexHeartbeats;
+  private final boolean readIndexAppliedIndexEnabled;
   private final LeaderLease lease;
 
   LeaderStateImpl(RaftServerImpl server) {
@@ -389,6 +390,8 @@ boolean isApplied() {
     } else {
       this.followerMaxGapThreshold = (long) (followerGapRatioMax * maxPendingRequests);
     }
+    this.readIndexAppliedIndexEnabled = RaftServerConfigKeys.Read.ReadIndex
+        .appliedIndexEnabled(properties);
 
     final RaftConfigurationImpl conf = state.getRaftConf();
     Collection<RaftPeer> others = conf.getOtherPeers(server.getId());
@@ -1134,21 +1137,23 @@ public boolean checkLeadership() {
   /**
    * Obtain the current readIndex for read only requests. See Raft paper section 6.4.
    * 1. Leader makes sure at least one log from current term is committed.
-   * 2. Leader record last committed index as readIndex.
+   * 2. Leader record last committed index or applied index (depending on configuration) as readIndex.
    * 3. Leader broadcast heartbeats to followers and waits for acknowledgements.
    * 4. If majority respond success, returns readIndex.
    * @return current readIndex.
    */
   CompletableFuture<Long> getReadIndex(Long readAfterWriteConsistentIndex) {
-    final long commitIndex = server.getRaftLog().getLastCommittedIndex();
+    final long index = readIndexAppliedIndexEnabled ?
+        server.getState().getLastAppliedIndex() : server.getRaftLog().getLastCommittedIndex();
     final long readIndex;
-    if (readAfterWriteConsistentIndex != null && readAfterWriteConsistentIndex > commitIndex) {
+    if (readAfterWriteConsistentIndex != null && readAfterWriteConsistentIndex > index) {
       readIndex = readAfterWriteConsistentIndex;
     } else {
-      readIndex = commitIndex;
+      readIndex = index;
     }
-    LOG.debug("readIndex={} (commitIndex={}, readAfterWriteConsistentIndex={})",
-        readIndex, commitIndex, readAfterWriteConsistentIndex);
+    LOG.debug("readIndex={} ({}Index={}, readAfterWriteConsistentIndex={})",
+        readIndex, readIndexAppliedIndexEnabled ? "applied" : "commit",
+        index, readAfterWriteConsistentIndex);
 
     // if group contains only one member, fast path
     if (server.getRaftConf().isSingleton()) {

ratis-server/src/test/java/org/apache/ratis/LinearizableReadTests.java

@@ -60,6 +60,8 @@ public abstract class LinearizableReadTests<CLUSTER extends MiniRaftCluster>
 
   public abstract boolean isLeaderLeaseEnabled();
 
+  public abstract boolean readIndexAppliedIndexEnabled();
+
   public abstract void assertRaftProperties(RaftProperties properties);
 
   void runWithNewCluster(CheckedConsumer<CLUSTER, Exception> testCase) throws Exception {
@@ -75,6 +77,7 @@ public void setup() {
     CounterStateMachine.setProperties(p);
     RaftServerConfigKeys.Read.setOption(p, LINEARIZABLE);
     RaftServerConfigKeys.Read.setLeaderLeaseEnabled(p, isLeaderLeaseEnabled());
+    RaftServerConfigKeys.Read.ReadIndex.setAppliedIndexEnabled(p, readIndexAppliedIndexEnabled());
   }
 
   @Test

ratis-test/src/test/java/org/apache/ratis/grpc/TestLinearizableLeaderLeaseReadWithGrpc.java

@@ -34,6 +34,11 @@ public boolean isLeaderLeaseEnabled() {
     return true;
   }
 
+  @Override
+  public boolean readIndexAppliedIndexEnabled() {
+    return false;
+  }
+
   @Override
   public void assertRaftProperties(RaftProperties p) {
     assertOption(LINEARIZABLE, p);

ratis-test/src/test/java/org/apache/ratis/grpc/TestLinearizableReadAppliedIndexLeaderLeaseReadWithGrpc.java

@@ -0,0 +1,27 @@
+/*
+ * 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.ratis.grpc;
+
+public class TestLinearizableReadAppliedIndexLeaderLeaseReadWithGrpc
+    extends TestLinearizableLeaderLeaseReadWithGrpc {
+
+  @Override
+  public boolean readIndexAppliedIndexEnabled() {
+    return true;
+  }
+}

ratis-test/src/test/java/org/apache/ratis/grpc/TestLinearizableReadAppliedIndexWithGrpc.java

@@ -0,0 +1,27 @@
+/*
+ * 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.ratis.grpc;
+
+public class TestLinearizableReadAppliedIndexWithGrpc
+  extends TestLinearizableReadWithGrpc {
+
+  @Override
+  public boolean readIndexAppliedIndexEnabled() {
+    return true;
+  }
+}

ratis-test/src/test/java/org/apache/ratis/grpc/TestLinearizableReadWithGrpc.java

@@ -34,6 +34,11 @@ public boolean isLeaderLeaseEnabled() {
     return false;
   }
 
+  @Override
+  public boolean readIndexAppliedIndexEnabled() {
+    return false;
+  }
+
   @Override
   public void assertRaftProperties(RaftProperties p) {
     assertOption(LINEARIZABLE, p);