RATIS-2403. Support leader batch write to improve linearizable follower read throughput

ratis-client/src/main/java/org/apache/ratis/client/impl/OrderedAsync.java

@@ -198,7 +198,7 @@ private void sendRequestWithRetry(PendingOrderedRequest pending) {
       return;
     }
 
-    if (getSlidingWindow((RaftPeerId) null).isFirst(pending.getSeqNum())) {
+    if (getSlidingWindow(request).isFirst(pending.getSeqNum())) {
       pending.setFirstRequest();
     }
     LOG.debug("{}: send* {}", client.getId(), request);

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

@@ -220,11 +220,38 @@ if it fails to receive any RPC responses from this peer within this specified ti
 
 ### 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                                                                 |
+| **Property**    | `raft.server.read.read-index.type`                                           |
+|:----------------|:-----------------------------------------------------------------------------|
+| **Description** | type of read index returned                                                  |
+| **Type**        | enum `Read.ReadIndex.Type` [`COMMIT_INDEX`, `APPLIED_INDEX`, `REPLIED_INDEX` |
+| **Default**     | `Read.ReadIndex.Type.COMMIT_INDEX`                                           |
+
+* `Read.ReadIndex.Type.COMMIT_INDEX` - Use leader's CommitIndex (see Raft Paper section 6.4)
+    * The safest type as it is specified in the Raft dissertation
+    * This ReadIndex type can be chosen if the base linearizable read from followers performance already meets expectations.
+
+* `Read.ReadIndex.Type.APPLIED_INDEX` - Use leader's AppliedIndex
+    * Allow leader to return AppliedIndex (instead of CommitIndex) as the ReadIndex
+    * This reduces the time follower applying logs up to ReadIndex since AppliedIndex ≤ CommitIndex
+    * This ReadIndex type can be chosen `Read.ReadIndex.Type.COMMIT_INDEX` read latency is too high.
+
+* `Read.ReadIndex.Type.REPLIED_INDEX` - Use leader's RepliedIndex
+    * RepliedIndex is defined as the last AppliedIndex of the leader when returning the last batch.
+    * Leader delays replying write requests and only reply them every write batch boundary configurable by `raft.server.read.read-index.replied-index.batch-interval`.
+    * This allows the ReadIndex to advance in a coarser, less frequent steps, so followers are more likely to have already applied past the ReadIndex when a read arrives.
+    * This is most effective on read-heavy, follower-read workloads which prioritizes overall read throughput without consistency sacrifice.
+    * There is a trade-off in increased write latency (up to one `raft.server.read.read-index.replied-index.batch-interval`) per write.
+    * RepliedIndex still guarantees linearizability (no stale read) since by definition each ReadIndex returns the index of the last replied request.
+    * If the RepliedIndex is set to 0, the behavior is identical to `Read.ReadIndex.Type.APPLIED_INDEX`
+
+Note that theoretically all the ReadIndex types still guarantee linearizability, 
+but there are tradeoffs (e.g. Write and Read performance) between different types.
+
+| **Property**    | `raft.server.read.read-index.replied-index.batch-interval`                                                                                  |
+|:----------------|:--------------------------------------------------------------------------------------------------------------------------------------------|
+| **Description** | if `Read.ReadIndex.Type` is `REPLIED_INDEX`, the interval at which held write replies are flushed to clients and `repliedIndex` is advanced |
+| **Type**        | TimeDuration                                                                                                                                |
+| **Default**     | 10ms                                                                                                                                        |
 
 | **Property**    | `raft.server.read.leader.heartbeat-check.enabled` |
 |:----------------|:--------------------------------------------------|

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

@@ -280,15 +280,34 @@ static void setWriteIndexCacheExpiryTime(RaftProperties properties, TimeDuration
     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());
+      enum Type {
+        /** ReadIndex returns leader's commitIndex (see Raft Paper section 6.4). */
+        COMMIT_INDEX,
+
+        /** ReadIndex returns leader's appliedIndex to reduce the ReadIndex latency. */
+        APPLIED_INDEX,
+
+        /** ReadIndex returns leader's repliedIndex, the index of the last replied request. */
+        REPLIED_INDEX
+      }
+
+      String TYPE_KEY = PREFIX + ".type";
+      Type TYPE_DEFAULT = Type.COMMIT_INDEX;
+      static Type type(RaftProperties properties) {
+        return get(properties::getEnum, TYPE_KEY, TYPE_DEFAULT, getDefaultLog());
+      }
+      static void setType(RaftProperties properties, Type type) {
+        set(properties::setEnum, TYPE_KEY, type);
       }
 
-      static void setAppliedIndexEnabled(RaftProperties properties, boolean enabled) {
-        setBoolean(properties::setBoolean, APPLIED_INDEX_ENABLED_KEY, enabled);
+      String REPLIED_INDEX_BATCH_INTERVAL_KEY = PREFIX + ".replied-index.batch-interval";
+      TimeDuration REPLIED_INDEX_BATCH_INTERVAL_DEFAULT = TimeDuration.valueOf(10, TimeUnit.MILLISECONDS);
+      static TimeDuration repliedIndexBatchInterval(RaftProperties properties) {
+        return getTimeDuration(properties.getTimeDuration(REPLIED_INDEX_BATCH_INTERVAL_DEFAULT.getUnit()),
+            REPLIED_INDEX_BATCH_INTERVAL_KEY, REPLIED_INDEX_BATCH_INTERVAL_DEFAULT, getDefaultLog());
+      }
+      static void setRepliedIndexBatchInterval(RaftProperties properties, TimeDuration interval) {
+        setTimeDuration(properties::setTimeDuration, REPLIED_INDEX_BATCH_INTERVAL_KEY, interval);
       }
     }
   }

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

@@ -39,6 +39,7 @@
 import org.apache.ratis.protocol.exceptions.ReadIndexException;
 import org.apache.ratis.protocol.exceptions.ReconfigurationTimeoutException;
 import org.apache.ratis.server.RaftServerConfigKeys;
+import org.apache.ratis.server.RaftServerConfigKeys.Read.ReadIndex.Type;
 import org.apache.ratis.server.impl.ReadIndexHeartbeats.AppendEntriesListener;
 import org.apache.ratis.server.leader.FollowerInfo;
 import org.apache.ratis.server.leader.LeaderState;
@@ -82,6 +83,7 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.LongSupplier;
 import java.util.function.Predicate;
+import java.util.function.Supplier;
 import java.util.function.ToLongFunction;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
@@ -353,10 +355,13 @@ boolean isApplied() {
   private final PendingStepDown pendingStepDown;
 
   private final ReadIndexHeartbeats readIndexHeartbeats;
-  private final boolean readIndexAppliedIndexEnabled;
+  private final RaftServerConfigKeys.Read.ReadIndex.Type readIndexType;
+  private final Supplier<Long> readIndexSupplier;
   private final boolean leaderHeartbeatCheckEnabled;
   private final LeaderLease lease;
 
+  private ReplyFlusher replyFlusher;
+
   LeaderStateImpl(RaftServerImpl server) {
     this.name = ServerStringUtils.generateUnifiedName(server.getMemberId(), getClass());
     this.server = server;
@@ -391,8 +396,21 @@ boolean isApplied() {
     } else {
       this.followerMaxGapThreshold = (long) (followerGapRatioMax * maxPendingRequests);
     }
-    this.readIndexAppliedIndexEnabled = RaftServerConfigKeys.Read.ReadIndex
-        .appliedIndexEnabled(properties);
+
+    this.readIndexType = RaftServerConfigKeys.Read.ReadIndex.type(properties);
+    switch (readIndexType) {
+    case REPLIED_INDEX:
+      this.replyFlusher = new ReplyFlusher(server.getId(), state.getLastAppliedIndex(),
+          RaftServerConfigKeys.Read.ReadIndex.repliedIndexBatchInterval(properties));
+      readIndexSupplier = replyFlusher::getRepliedIndex;
+      break;
+    case APPLIED_INDEX:
+      readIndexSupplier = () -> server.getState().getLastAppliedIndex();
+      break;
+    case COMMIT_INDEX:
+    default:
+      readIndexSupplier = () -> server.getRaftLog().getLastCommittedIndex();
+    }
     this.leaderHeartbeatCheckEnabled = RaftServerConfigKeys.Read
         .leaderHeartbeatCheckEnabled(properties);
 
@@ -418,6 +436,11 @@ void start() {
     // Initialize startup log entry and append it to the RaftLog
     startupLogEntry.get();
     processor.start();
+
+    if (replyFlusher != null) {
+      replyFlusher.start(startupLogEntry.get().startIndex);
+    }
+
     senders.forEach(LogAppender::start);
   }
 
@@ -453,6 +476,9 @@ CompletableFuture<Void> stop() {
     startupLogEntry.get().getAppliedIndexFuture().completeExceptionally(
         new ReadIndexException("failed to obtain read index since: ", nle));
     server.getServerRpc().notifyNotLeader(server.getMemberId().getGroupId());
+    if (replyFlusher != null) {
+      replyFlusher.stop();
+    }
     logAppenderMetrics.unregister();
     raftServerMetrics.unregister();
     pendingRequests.close();
@@ -620,7 +646,7 @@ public AppendEntriesRequestProto newAppendEntriesRequestProto(FollowerInfo follo
     final boolean initializing = !isCaughtUp(follower);
     final RaftPeerId targetId = follower.getId();
     return ServerProtoUtils.toAppendEntriesRequestProto(server.getMemberId(), targetId, getCurrentTerm(), entries,
-        ServerImplUtils.effectiveCommitIndex(raftLog.getLastCommittedIndex(), previous, entries.size()),
+        ServerImplUtils.effectiveCommitIndex(readIndexSupplier.get(), previous, entries.size()),
         initializing, previous, server.getCommitInfos(), callId);
   }
 
@@ -1140,23 +1166,21 @@ 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 or applied index (depending on configuration) as readIndex.
+   * 2. Leader record last committed index or applied index or replied 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 index = readIndexAppliedIndexEnabled ?
-        server.getState().getLastAppliedIndex() : server.getRaftLog().getLastCommittedIndex();
+    final long index = readIndexSupplier.get();
     final long readIndex;
     if (readAfterWriteConsistentIndex != null && readAfterWriteConsistentIndex > index) {
       readIndex = readAfterWriteConsistentIndex;
     } else {
       readIndex = index;
     }
-    LOG.debug("readIndex={} ({}Index={}, readAfterWriteConsistentIndex={})",
-        readIndex, readIndexAppliedIndexEnabled ? "applied" : "commit",
-        index, readAfterWriteConsistentIndex);
+    LOG.debug("readIndex={} ({}={}, readAfterWriteConsistentIndex={})",
+        readIndex, readIndexType, index, readAfterWriteConsistentIndex);
 
     // if group contains only one member, fast path
     if (server.getRaftConf().isSingleton()) {
@@ -1217,8 +1241,23 @@ private boolean checkLeaderLease() {
         && (server.getRaftConf().isSingleton() || lease.isValid());
   }
 
-  void replyPendingRequest(TermIndex termIndex, RaftClientReply reply) {
-    pendingRequests.replyPendingRequest(termIndex, reply);
+  void replyPendingRequest(TermIndex termIndex, RaftClientReply reply, RetryCacheImpl.CacheEntry cacheEntry) {
+    final PendingRequest pending = pendingRequests.remove(termIndex);
+    if (pending == null) {
+      return;
+    }
+
+    final LongSupplier replyMethod = () -> {
+      cacheEntry.updateResult(reply);
+      pending.setReply(reply);
+      return termIndex.getIndex();
+    };
+
+    if (readIndexType == Type.REPLIED_INDEX) {
+      replyFlusher.hold(replyMethod);
+    } else {
+      replyMethod.getAsLong();
+    }
   }
 
   TransactionContext getTransactionContext(TermIndex termIndex) {

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

@@ -264,12 +264,13 @@ TransactionContext getTransactionContext(TermIndex termIndex) {
     return pendingRequest != null ? pendingRequest.getEntry() : null;
   }
 
-  void replyPendingRequest(TermIndex termIndex, RaftClientReply reply) {
+  /** @return the removed the {@link PendingRequest} for the given {@link TermIndex}. */
+  PendingRequest remove(TermIndex termIndex) {
     final PendingRequest pending = pendingRequests.remove(termIndex);
     if (pending != null) {
       Preconditions.assertEquals(termIndex, pending.getTermIndex(), "termIndex");
-      pending.setReply(reply);
     }
+    return pending;
   }
 
   /**

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

@@ -1837,8 +1837,12 @@ private CompletableFuture<Message> replyPendingRequest(
       }
 
       // update pending request
-      role.getLeaderState().ifPresent(leader -> leader.replyPendingRequest(termIndex, r));
-      cacheEntry.updateResult(r);
+      final LeaderStateImpl leader = role.getLeaderState().orElse(null);
+      if (leader != null) {
+        leader.replyPendingRequest(termIndex, r, cacheEntry);
+      } else {
+        cacheEntry.updateResult(r);
+      }
     });
   }