RATIS-1482. fix shouldNotifyToInstallSnapshot to return the first avail term index

[Xushaohong]

What changes were proposed in this pull request?

1. shouldNotifyToInstallSnapshot should return the first available log's start term index, but right now returns LastEntryTermIndex. The possible corner case is when the log is purged and no newer log entries are appended.
2. Added of simulation of snapshot transmission delay in InstallSnapshotNotificationTests

What is the link to the Apache JIRA

https://issues.apache.org/jira/browse/RATIS-1482

How was this patch tested?

CI

[szetszwo]

@Xushaohong , thanks fa lot for working on this. We don't need to get the latest snapshot as shown below.

+++ b/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcLogAppender.java
@@ -590,24 +590,16 @@ public class GrpcLogAppender extends LogAppenderBase {
   private TermIndex shouldNotifyToInstallSnapshot() {
     final FollowerInfo follower = getFollower();
     final long leaderNextIndex = getRaftLog().getNextIndex();
+    final long leaderStartIndex = getRaftLog().getStartIndex();
+    final TermIndex firstAvailable = Optional.ofNullable(getRaftLog().getTermIndex(leaderStartIndex))
+        .orElseGet(() -> TermIndex.valueOf(getServer().getInfo().getCurrentTerm(), leaderNextIndex));
     final boolean isFollowerBootstrapping = getLeaderState().isFollowerBootstrapping(follower);
-
     if (isFollowerBootstrapping && !follower.hasAttemptedToInstallSnapshot()) {
       // If the follower is bootstrapping and has not yet installed any snapshot from leader, then the follower should
       // be notified to install a snapshot. Every follower should try to install at least one snapshot during
       // bootstrapping, if available.
-      TermIndex lastEntry = getRaftLog().getLastEntryTermIndex();
-      if (lastEntry == null) {
-        // lastEntry may need to be derived from snapshot
-        SnapshotInfo snapshot = getServer().getStateMachine().getLatestSnapshot();
-        if (snapshot != null) {
-          lastEntry = snapshot.getTermIndex();
-        }
-      }
-      if (LOG.isDebugEnabled()) {
-        LOG.debug("{}: Notify follower to install snapshot as it is bootstrapping to TermIndex {}.", this, lastEntry);
-      }
-      return lastEntry;
+      LOG.debug("{}: follower is bootstrapping, notify to install snapshot to {}.", this, firstAvailable);
+      return firstAvailable;
     }
 
     final long followerNextIndex = follower.getNextIndex();
@@ -615,18 +607,15 @@ public class GrpcLogAppender extends LogAppenderBase {
       return null;
     }
 
-    final long leaderStartIndex = getRaftLog().getStartIndex();
-
     if (followerNextIndex < leaderStartIndex) {
       // The Leader does not have the logs from the Follower's last log
       // index onwards. And install snapshot is disabled. So the Follower
       // should be notified to install the latest snapshot through its
       // State Machine.
-      return getRaftLog().getTermIndex(leaderStartIndex);
+      return firstAvailable;
     } else if (leaderStartIndex == RaftLog.INVALID_LOG_INDEX) {
       // Leader has no logs to check from, hence return next index.
-      return TermIndex.valueOf(getServer().getInfo().getCurrentTerm(),
-          leaderNextIndex);
+      return firstAvailable;
     }

[Xushaohong]

Yours is much more concise. I have submitted a new commit. PTAL @szetszwo

[szetszwo]

+1 the change looks good.