RATIS-1622. Fix high CPU load when some followers are down

ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcLogAppender.java

@@ -158,12 +158,20 @@ public void run() throws IOException {
   public long getWaitTimeMs() {
     if (haveTooManyPendingRequests()) {
       return getHeartbeatWaitTimeMs(); // Should wait for a short time
-    } else if (shouldSendAppendEntries()) {
+    } else if (shouldSendAppendEntries() && !isSlowFollower()) {
+      // For normal nodes, new entries should be sent ASAP
+      // however for slow followers (especially when the follower is down),
+      // keep sending without any wait time only ends up in high CPU load
       return 0L;
     }
     return Math.min(10L, getHeartbeatWaitTimeMs());
   }
 
+  private boolean isSlowFollower() {
+    final TimeDuration elapsedTime = getFollower().getLastRpcResponseTime().elapsedTime();
+    return elapsedTime.compareTo(getServer().properties().rpcSlownessTimeout()) > 0;
+  }
+
   private void mayWait() {
     // use lastSend time instead of lastResponse time
     try {

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

@@ -69,9 +69,8 @@ default RaftPeerId getId() {
 
     /** @return the {@link RaftPeer} for this division. */
     default RaftPeer getPeer() {
-      return Optional.ofNullable(getGroup())
-          .map(g -> g.getPeer(getId()))
-          .orElseGet(() -> getRaftServer().getPeer());
+      return Optional.ofNullable(getRaftConf().getPeer(getId()))
+        .orElseGet(() -> getRaftServer().getPeer());
     }
 
     /** @return the information about this division. */

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

@@ -379,7 +379,7 @@ private void runTestRestartWithCorruptedLogEntry(CLUSTER cluster) throws Excepti
       final long mid = size / 2;
       raf.seek(mid);
       for (long i = mid; i < size; i++) {
-        raf.write(0);
+        raf.write(-1);
       }
     }