RATIS-2225. RaftClientRequest leak in RaftServerImpl. (#1198)

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

@@ -631,6 +631,15 @@ synchronized void changeToLeader() {
 
   @Override
   public Collection<CommitInfoProto> getCommitInfos() {
+    try {
+      return getCommitInfosImpl();
+    } catch (Throwable t) {
+      LOG.warn("{} Failed to getCommitInfos", getMemberId(), t);
+      return Collections.emptyList();
+    }
+  }
+
+  private Collection<CommitInfoProto> getCommitInfosImpl() {
     final List<CommitInfoProto> infos = new ArrayList<>();
     // add the commit info of this server
     final long commitIndex = updateCommitInfoCache();
@@ -922,17 +931,10 @@ CompletableFuture<RaftClientReply> executeSubmitClientRequestAsync(
   public CompletableFuture<RaftClientReply> submitClientRequestAsync(
       ReferenceCountedObject<RaftClientRequest> requestRef) {
     final RaftClientRequest request = requestRef.retain();
-    LOG.debug("{}: receive client request({})", getMemberId(), request);
-
     try {
+      LOG.debug("{}: receive client request({})", getMemberId(), request);
       assertLifeCycleState(LifeCycle.States.RUNNING);
-    } catch (ServerNotReadyException e) {
-      final RaftClientReply reply = newExceptionReply(request, e);
-      requestRef.release();
-      return CompletableFuture.completedFuture(reply);
-    }
 
-    try {
       RaftClientRequest.Type type = request.getType();
       final Timekeeper timer = raftServerMetrics.getClientRequestTimer(type);
       final Optional<Timekeeper.Context> timerContext = Optional.ofNullable(timer).map(Timekeeper::time);
@@ -942,6 +944,11 @@ public CompletableFuture<RaftClientReply> submitClientRequestAsync(
           raftServerMetrics.incFailedRequestCount(type);
         }
       });
+    } catch (RaftException e) {
+      return CompletableFuture.completedFuture(newExceptionReply(request, e));
+    } catch (Throwable t) {
+      LOG.error("{} Failed to submitClientRequestAsync for {}", getMemberId(), request, t);
+      return CompletableFuture.completedFuture(newExceptionReply(request, new RaftException(t)));
     } finally {
       requestRef.release();
     }