RATIS-2703. Fix GrpcLogAppender SNAPSHOT_EXPIRED pending queue leak

[amaliujia]

What changes were proposed in this pull request?

Fix GrpcLogAppender.InstallSnapshotResponseHandler to properly handle SNAPSHOT_EXPIRED install-snapshot replies by calling removePending(reply) and setAttemptedToInstallSnapshot(), matching other terminal snapshot results such as SNAPSHOT_UNAVAILABLE and SUCCESS.

What is the link to the Apache JIRA

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

How was this patch tested?

Unit test

[amaliujia]

R: @szetszwo

[szetszwo]

@amaliujia , thanks for working on this! Please see the comment inlined.

[szetszwo]

Let's handle SNAPSHOT_EXPIRED the same way as SNAPSHOT_UNAVAILABLE:

diff --git a/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcLogAppender.java b/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcLogAppender.java
index e6870b9b8..60d7ee2b2 100644
--- a/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcLogAppender.java
+++ b/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcLogAppender.java
@@ -775,24 +775,26 @@ public class GrpcLogAppender extends LogAppenderBase {
           removePending(reply);
           break;
         case SNAPSHOT_UNAVAILABLE:
-          BatchLogger.print(BatchLogKey.SNAPSHOT_UNAVAILABLE,  name,
-              suffix -> LOG.info("{}: Follower failed since the snapshot is unavailable {}", this, suffix));
-          getFollower().setAttemptedToInstallSnapshot();
-          notifyInstallSnapshotFinished(InstallSnapshotResult.SNAPSHOT_UNAVAILABLE, RaftLog.INVALID_LOG_INDEX);
-          removePending(reply);
-          break;
-        case UNRECOGNIZED:
-          LOG.error("{}: Reply result {}, {}",
-              name, reply.getResult(), ServerStringUtils.toInstallSnapshotReplyString(reply));
+          handleFailureReply(reply, BatchLogKey.SNAPSHOT_UNAVAILABLE);
           break;
         case SNAPSHOT_EXPIRED:
-          LOG.warn("{}: Follower failed since the request expired, {}",
-              name, ServerStringUtils.toInstallSnapshotReplyString(reply));
+          handleFailureReply(reply, BatchLogKey.SNAPSHOT_UNAVAILABLE);
+          break;
         default:
+          LOG.error("{}: Unexpected InstallSnapshotReply result {} in {}",
+              name, reply.getResult(), ServerStringUtils.toInstallSnapshotReplyString(reply));
           break;
       }
     }
 
+    private void handleFailureReply(InstallSnapshotReplyProto reply, BatchLogKey logKey) {
+      BatchLogger.print(logKey,  name,
+          suffix -> LOG.info("{}: Follower failed due to {}: {}{}", this, reply.getResult(), reply, suffix));
+      getFollower().setAttemptedToInstallSnapshot();
+      notifyInstallSnapshotFinished(reply.getResult(), RaftLog.INVALID_LOG_INDEX);
+      removePending(reply);
+    }
+
     @Override
     public void onError(Throwable t) {
       if (!isRunning()) {