RATIS-2329. NettyRpcProxy should support handling netty channel exception to prevent replication stuck (#1285)

Author: leixm

ratis-common/src/main/java/org/apache/ratis/util/IOUtils.java

@@ -91,9 +91,14 @@ static <T> T getFromFuture(CompletableFuture<T> future, Supplier<Object> name, T
   }
 
   static boolean shouldReconnect(Throwable e) {
-    return ReflectionUtils.isInstance(e,
-        SocketException.class, SocketTimeoutException.class, ClosedChannelException.class, EOFException.class,
-        AlreadyClosedException.class);
+    for (; e != null; e = e.getCause()) {
+      if (ReflectionUtils.isInstance(e,
+          SocketException.class, SocketTimeoutException.class, ClosedChannelException.class, EOFException.class,
+          AlreadyClosedException.class, TimeoutIOException.class)) {
+        return true;
+      }
+    }
+    return false;
   }
 
   static void readFully(InputStream in, int buffSize) throws IOException {

ratis-netty/src/main/java/org/apache/ratis/netty/NettyRpcProxy.java

@@ -30,10 +30,13 @@
 import org.apache.ratis.proto.RaftProtos.RaftRpcRequestProto;
 import org.apache.ratis.proto.netty.NettyProtos.RaftNettyServerReplyProto;
 import org.apache.ratis.proto.netty.NettyProtos.RaftNettyServerRequestProto;
+import org.apache.ratis.protocol.exceptions.AlreadyClosedException;
 import org.apache.ratis.util.IOUtils;
 import org.apache.ratis.util.PeerProxyMap;
 import org.apache.ratis.util.ProtoUtils;
 import org.apache.ratis.util.TimeDuration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.io.Closeable;
 import java.io.IOException;
@@ -47,6 +50,7 @@
 import static org.apache.ratis.proto.netty.NettyProtos.RaftNettyServerReplyProto.RaftNettyServerReplyCase.EXCEPTIONREPLY;
 
 public class NettyRpcProxy implements Closeable {
+  public static final Logger LOG = LoggerFactory.getLogger(NettyRpcProxy.class);
   public static class PeerMap extends PeerProxyMap<NettyRpcProxy> {
     private final EventLoopGroup group;
 
@@ -121,6 +125,18 @@ protected void channelRead0(ChannelHandlerContext ctx,
             future.complete(proto);
           }
         }
+
+        @Override
+        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
+          client.close();
+          failOutstandingRequests(new IOException("Caught an exception for the connection to " + peer, cause));
+        }
+
+        @Override
+        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+          failOutstandingRequests(new AlreadyClosedException("Channel to " + peer + " is inactive."));
+          super.channelInactive(ctx);
+        }
       };
       final ChannelInitializer<SocketChannel> initializer
           = new ChannelInitializer<SocketChannel>() {
@@ -153,9 +169,14 @@ synchronized CompletableFuture<RaftNettyServerReplyProto> pollReply() {
     @Override
     public synchronized void close() {
       client.close();
+      failOutstandingRequests(new AlreadyClosedException("Closing connection to " + peer));
+    }
+
+    private synchronized void failOutstandingRequests(Throwable cause) {
       if (!replies.isEmpty()) {
-        final IOException e = new IOException("Connection to " + peer + " is closed.");
-        replies.stream().forEach(f -> f.completeExceptionally(e));
+        LOG.warn("Still have {} requests outstanding from {} connection: {}",
+            replies.size(), peer, cause.toString());
+        replies.forEach(f -> f.completeExceptionally(cause));
         replies.clear();
       }
     }

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

@@ -181,12 +181,8 @@ void runTestRequestAsyncWithRetryFailure(boolean initialMessages, CLUSTER cluste
       // the second half still have retry time remaining.
       sleepTime.apply(t -> t*2).sleep();
 
-      if (leader != null) {
-        cluster.restartServer(leader, false);
-      } else {
-        cluster.start();
-      }
-
+      // The client will try to reconnect, but the server is
+      // not started at this time and the retry will fail anyway.
       // all the calls should fail for ordering guarantee
       for(int i = 0; i < replies.size(); i++) {
         final CheckedRunnable<Exception> getReply = replies.get(i)::get;
@@ -203,6 +199,12 @@ void runTestRequestAsyncWithRetryFailure(boolean initialMessages, CLUSTER cluste
 
       testFailureCaseAsync("last-request", () -> client.async().send(new SimpleMessage("last")),
           AlreadyClosedException.class, RaftRetryFailureException.class);
+
+      if (leader != null) {
+        cluster.restartServer(leader, false);
+      } else {
+        cluster.start();
+      }
     }
   }