RATIS-1411. Alleviate slow follower issue

ratis-common/src/main/java/org/apache/ratis/conf/ConfUtils.java

@@ -68,6 +68,15 @@ static BiConsumer<String, Integer> requireMax(int max) {
     };
   }
 
+  static BiConsumer<String, Float> requireMax(float max) {
+    return (key, value) -> {
+      if (value > max) {
+        throw new IllegalArgumentException(
+            key + " = " + value + " > max = " + max);
+      }
+    };
+  }
+
   static BiConsumer<String, Long> requireMin(SizeInBytes min) {
     return requireMin(min.getSize());
   }
@@ -149,6 +158,13 @@ static long getLong(
     return get(longGetter, key, defaultValue, logger, assertions);
   }
 
+  @SafeVarargs
+  static float getFloat(
+      BiFunction<String, Float, Float> floatGetter,
+      String key, float defaultValue, Consumer<String> logger, BiConsumer<String, Float>... assertions) {
+    return get(floatGetter, key, defaultValue, logger, assertions);
+  }
+
   @SafeVarargs
   static File getFile(
       BiFunction<String, File, File> fileGetter,
@@ -218,6 +234,13 @@ static void setLong(
     set(longSetter, key, value, assertions);
   }
 
+  @SafeVarargs
+  static void setFloat(
+      BiConsumer<String, Float> floatSetter, String key, float value,
+      BiConsumer<String, Float>... assertions) {
+    set(floatSetter, key, value, assertions);
+  }
+
   @SafeVarargs
   static void setFile(
       BiConsumer<String, File> fileSetter, String key, File value,

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

@@ -119,6 +119,17 @@ static SizeInBytes byteLimit(RaftProperties properties) {
     static void setByteLimit(RaftProperties properties, SizeInBytes byteLimit) {
       setSizeInBytes(properties::set, BYTE_LIMIT_KEY, byteLimit, requireMin(1L));
     }
+
+    String FOLLOWER_MAX_GAP_RATIO_KEY = PREFIX + ".follower-max-gap-ratio";
+    float FOLLOWER_MAX_GAP_RATIO_DEFAULT = 1;
+
+    static float followerMaxGapRatio(RaftProperties properties) {
+      return getFloat(properties::getFloat, FOLLOWER_MAX_GAP_RATIO_KEY,
+          FOLLOWER_MAX_GAP_RATIO_DEFAULT, getDefaultLog(), requireMax(1f));
+    }
+    static void setFollowerMaxGapRatio(RaftProperties properties, float ratio) {
+      setFloat(properties::setFloat, FOLLOWER_MAX_GAP_RATIO_KEY, ratio, requireMax(1f));
+    }
   }
 
   interface Watch {

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

@@ -79,6 +79,7 @@
 import java.util.stream.Stream;
 
 import static org.apache.ratis.server.RaftServer.Division.LOG;
+import static org.apache.ratis.server.RaftServerConfigKeys.Write.FOLLOWER_MAX_GAP_RATIO_KEY;
 
 /**
  * States for leader only. It contains three different types of processors:
@@ -250,6 +251,8 @@ boolean removeAll(Collection<LogAppender> c) {
   private final long placeHolderIndex;
   private final RaftServerMetricsImpl raftServerMetrics;
   private final LogAppenderMetrics logAppenderMetrics;
+  private final long maxPendingRequests;
+  private final long followerMaxGapThreshold;
 
   LeaderStateImpl(RaftServerImpl server) {
     this.name = server.getMemberId() + "-" + JavaUtils.getClassSimpleName(getClass());
@@ -269,6 +272,13 @@ boolean removeAll(Collection<LogAppender> c) {
     this.pendingRequests = new PendingRequests(server.getMemberId(), properties, raftServerMetrics);
     this.watchRequests = new WatchRequests(server.getMemberId(), properties);
     this.messageStreamRequests = new MessageStreamRequests(server.getMemberId());
+    this.maxPendingRequests = RaftServerConfigKeys.Write.elementLimit(properties);
+    float followerMaxGapRatio = RaftServerConfigKeys.Write.followerMaxGapRatio(properties);
+
+    if (followerMaxGapRatio > 1f || followerMaxGapRatio <= 0f) {
+      throw new IllegalArgumentException(FOLLOWER_MAX_GAP_RATIO_KEY + "s value must between [1, 0)");
+    }
+    this.followerMaxGapThreshold = (long)(followerMaxGapRatio * maxPendingRequests);
 
     final RaftConfigurationImpl conf = state.getRaftConf();
     Collection<RaftPeer> others = conf.getOtherPeers(server.getId());
@@ -700,6 +710,17 @@ static MinMajorityMax valueOf(long[] sorted) {
       return new MinMajorityMax(sorted[0], getMajority(sorted), getMax(sorted));
     }
 
+    static MinMajorityMax valueOf(long[] sorted, long gapThreshold) {
+      long majority = getMajority(sorted);
+      long min = sorted[0];
+      if ((majority - min) > gapThreshold) {
+        // The the gap between majority and min(the slow follower) is greater than gapThreshold,
+        // set the majority to min, which will skip one round of lastCommittedIndex update in updateCommit().
+        majority = min;
+      }
+      return new MinMajorityMax(min, majority, getMax(sorted));
+    }
+
     static long getMajority(long[] sorted) {
       return sorted[(sorted.length - 1) / 2];
     }
@@ -725,7 +746,7 @@ private Optional<MinMajorityMax> getMajorityMin(ToLongFunction<FollowerInfo> fol
     }
 
     final long[] indicesInNewConf = getSorted(followers, includeSelf, followerIndex, logIndex);
-    final MinMajorityMax newConf = MinMajorityMax.valueOf(indicesInNewConf);
+    final MinMajorityMax newConf = MinMajorityMax.valueOf(indicesInNewConf, followerMaxGapThreshold);
 
     if (!conf.isTransitional()) {
       return Optional.of(newConf);
@@ -737,7 +758,7 @@ private Optional<MinMajorityMax> getMajorityMin(ToLongFunction<FollowerInfo> fol
       }
 
       final long[] indicesInOldConf = getSorted(oldFollowers, includeSelfInOldConf, followerIndex, logIndex);
-      final MinMajorityMax oldConf = MinMajorityMax.valueOf(indicesInOldConf);
+      final MinMajorityMax oldConf = MinMajorityMax.valueOf(indicesInOldConf, followerMaxGapThreshold);
       return Optional.of(newConf.combine(oldConf));
     }
   }