RATIS-2233. make NOPROGRESS timeout configurable (#1204)

Author: SzyWilliam

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

@@ -104,6 +104,16 @@ When bootstrapping a new peer, If the gap between the match index of the
 peer and the leader's latest committed index is less than this gap, we
 treat the peer as caught-up. Increase this number when write throughput is high.
 
+---------------------------------------------------------------------------------
+| **Property**    | `raft.server.staging.timeout`            |
+|:----------------|:-----------------------------------------|
+| **Description** | timeout of bootstrapping a new peer      |
+| **Type**        | TimeDuration                             |
+| **Default**     | 3 times of `raft.server.rpc.timeout.max` |
+
+During the initialization of a new peer, the leader will classify the bootstrap process as "NO PROGRESS" 
+if it fails to receive any RPC responses from this peer within this specified timeout period.
+
 ---------------------------------------------------------------------------------
 ### ThreadPool - Configurations related to server thread pools.
 

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

@@ -116,6 +116,21 @@ static void setStagingCatchupGap(RaftProperties properties, int stagingCatchupGa
     setInt(properties::setInt, STAGING_CATCHUP_GAP_KEY, stagingCatchupGap);
   }
 
+  String STAGING_TIMEOUT_KEY = PREFIX + ".staging.timeout";
+
+  TimeDuration STAGING_TIMEOUT_DEFAULT = null;
+
+  static TimeDuration stagingTimeout(RaftProperties properties) {
+    final TimeDuration fallbackStagingTimeout = Rpc.timeoutMax(properties, null).multiply(3);
+    return getTimeDuration(properties.getTimeDuration(fallbackStagingTimeout.getUnit()),
+        STAGING_TIMEOUT_KEY, STAGING_TIMEOUT_DEFAULT,
+        Rpc.TIMEOUT_MAX_KEY, fallbackStagingTimeout, getDefaultLog());
+  }
+  static void setStagingTimeout(RaftProperties properties, TimeDuration stagingTimeout) {
+    setTimeDuration(properties::setTimeDuration, STAGING_TIMEOUT_KEY, stagingTimeout);
+  }
+
+
   interface ThreadPool {
     String PREFIX = RaftServerConfigKeys.PREFIX + ".threadpool";
 

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

@@ -350,6 +350,7 @@ boolean isApplied() {
 
   private final boolean logMetadataEnabled;
   private final int stagingCatchupGap;
+  private final TimeDuration stagingTimeout;
   private final RaftServerMetricsImpl raftServerMetrics;
   private final LogAppenderMetrics logAppenderMetrics;
   private final long followerMaxGapThreshold;
@@ -364,6 +365,7 @@ boolean isApplied() {
 
     final RaftProperties properties = server.getRaftServer().getProperties();
     stagingCatchupGap = RaftServerConfigKeys.stagingCatchupGap(properties);
+    stagingTimeout = RaftServerConfigKeys.stagingTimeout(properties);
 
     final ServerState state = server.getState();
     this.raftLog = state.getLog();
@@ -789,7 +791,7 @@ public void run() {
   private BootStrapProgress checkProgress(FollowerInfo follower, long committed) {
     Preconditions.assertTrue(!isCaughtUp(follower));
     final Timestamp progressTime = Timestamp.currentTime().addTimeMs(-server.getMaxTimeoutMs());
-    final Timestamp timeoutTime = Timestamp.currentTime().addTimeMs(-3L * server.getMaxTimeoutMs());
+    final Timestamp timeoutTime = Timestamp.currentTime().addTimeMs(-stagingTimeout.toLong(TimeUnit.MILLISECONDS));
     if (follower.getLastRpcResponseTime().compareTo(timeoutTime) < 0) {
       LOG.debug("{} detects a follower {} timeout ({}ms) for bootstrapping", this, follower,
           follower.getLastRpcResponseTime().elapsedTimeMs());