RATIS-2294. Fix NettyClientRpc exception and timeout handling (apache#1264)

(cherry picked from commit 8634542)

Author: slfan1989

ratis-netty/src/main/java/org/apache/ratis/netty/client/NettyClientRpc.java

@@ -29,8 +29,6 @@
 import org.apache.ratis.proto.RaftProtos.GroupManagementRequestProto;
 import org.apache.ratis.proto.RaftProtos.SetConfigurationRequestProto;
 import org.apache.ratis.proto.netty.NettyProtos.RaftNettyServerRequestProto;
-import org.apache.ratis.protocol.exceptions.LeaderNotReadyException;
-import org.apache.ratis.protocol.exceptions.NotLeaderException;
 import org.apache.ratis.protocol.exceptions.TimeoutIOException;
 import org.apache.ratis.util.JavaUtils;
 import org.apache.ratis.util.TimeDuration;
@@ -40,29 +38,26 @@
 
 import java.io.IOException;
 import java.util.concurrent.CompletableFuture;
-import java.util.function.Supplier;
 
 public class NettyClientRpc extends RaftClientRpcWithProxy<NettyRpcProxy> {
 
   public static final Logger LOG = LoggerFactory.getLogger(NettyClientRpc.class);
 
-  private Supplier<String> name;
-  private ClientId cId;
-  private final TimeDuration requestTimeoutDuration;
+  private ClientId clientId;
+  private final TimeDuration requestTimeout;
   private final TimeoutExecutor scheduler = TimeoutExecutor.getInstance();
 
   public NettyClientRpc(ClientId clientId, RaftProperties properties) {
     super(new NettyRpcProxy.PeerMap(clientId.toString(), properties));
-    this.cId = clientId;
-    this.requestTimeoutDuration = RaftClientConfigKeys.Rpc.requestTimeout(properties);
+    this.clientId = clientId;
+    this.requestTimeout = RaftClientConfigKeys.Rpc.requestTimeout(properties);
   }
 
   @Override
   public CompletableFuture<RaftClientReply> sendRequestAsync(RaftClientRequest request) {
     final RaftPeerId serverId = request.getServerId();
     long callId = request.getCallId();
     try {
-      name = JavaUtils.memoize(() -> cId + "->" + serverId);
       final NettyRpcProxy proxy = getProxies().getProxy(serverId);
       final RaftNettyServerRequestProto serverRequestProto = buildRequestProto(request);
       final CompletableFuture<RaftClientReply> replyFuture = new CompletableFuture<>();
@@ -75,51 +70,40 @@ public CompletableFuture<RaftClientReply> sendRequestAsync(RaftClientRequest req
         } else {
           return ClientProtoUtils.toRaftClientReply(replyProto.getRaftClientReply());
         }
-      }).thenCompose(raftReply -> {
-        final NotLeaderException nle = raftReply.getNotLeaderException();
-        if (nle != null) {
-          return failedFuture(nle);
-        }
-        final LeaderNotReadyException lnre = raftReply.getLeaderNotReadyException();
-        if (lnre != null) {
-          return failedFuture(lnre);
+      }).whenComplete((reply, e) -> {
+        if (e == null) {
+          if (reply == null) {
+            e = new NullPointerException("Both reply==null && e==null");
+          }
+          if (e == null) {
+            e = reply.getNotLeaderException();
+          }
+          if (e == null) {
+            e = reply.getLeaderNotReadyException();
+          }
         }
-        return CompletableFuture.completedFuture(raftReply);
-      }).whenComplete((raftReply, ex) -> {
-        if (ex != null) {
-          replyFuture.completeExceptionally(ex);
+
+        if (e != null) {
+          replyFuture.completeExceptionally(e);
         } else {
-          replyFuture.complete(raftReply);
+          replyFuture.complete(reply);
         }
       });
 
-      scheduler.onTimeout(requestTimeoutDuration,
-        () -> {
+      scheduler.onTimeout(requestTimeout, () -> {
           if (!replyFuture.isDone()) {
-            TimeoutIOException timeout = new TimeoutIOException(
-                getName()+ " request #" + callId + " timeout " +
-                requestTimeoutDuration.getDuration());
-            replyFuture.completeExceptionally(timeout);
+            final String s = clientId + "->" + serverId + " request #" +
+                callId + " timeout " + requestTimeout.getDuration();
+            replyFuture.completeExceptionally(new TimeoutIOException(s));
           }
-        }, LOG, () -> "Timeout check for client request #" + callId
-      );
+        }, LOG, () -> "Timeout check for client request #" + callId);
 
       return replyFuture;
     } catch (Throwable e) {
       return JavaUtils.completeExceptionally(e);
     }
   }
 
-  private String getName() {
-    return name.get();
-  }
-
-  private <T> CompletableFuture<T> failedFuture(Throwable ex) {
-    CompletableFuture<T> future = new CompletableFuture<>();
-    future.completeExceptionally(ex);
-    return future;
-  }
-
   @Override
   public RaftClientReply sendRequest(RaftClientRequest request) throws IOException {
     final RaftPeerId serverId = request.getServerId();

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

@@ -47,6 +47,7 @@
 import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
 import org.apache.ratis.thirdparty.com.google.protobuf.InvalidProtocolBufferException;
 import org.apache.ratis.util.JavaUtils;
+import org.apache.ratis.util.PlatformUtils;
 import org.apache.ratis.util.Slf4jUtils;
 import org.apache.ratis.util.TimeDuration;
 import org.apache.ratis.util.function.CheckedRunnable;
@@ -85,6 +86,10 @@ public abstract class RaftAsyncTests<CLUSTER extends MiniRaftCluster> extends Ba
   {
     getProperties().setClass(MiniRaftCluster.STATEMACHINE_CLASS_KEY,
         SimpleStateMachine4Testing.class, StateMachine.class);
+    if (!PlatformUtils.LINUX) {
+      getProperties().setBoolean("raft.netty.server.use-epoll", false);
+      getProperties().setBoolean("raft.netty.client.use-epoll", false);
+    }
   }
 
   @Test
@@ -110,8 +115,8 @@ public void testAsyncConfiguration() throws IOException {
 
   static void assertRaftRetryFailureException(RaftRetryFailureException rfe, RetryPolicy retryPolicy, String name) {
     Assertions.assertNotNull(rfe, name + " does not have RaftRetryFailureException");
-    Assertions.assertTrue(
-       rfe.getMessage().contains(retryPolicy.toString()), name + ": unexpected error message, rfe=" + rfe + ", retryPolicy=" + retryPolicy);
+    Assertions.assertTrue(rfe.getMessage().contains(retryPolicy.toString()),
+        name + ": unexpected error message, rfe=" + rfe + ", retryPolicy=" + retryPolicy);
   }
 
   @Test
@@ -284,8 +289,8 @@ public void testStaleReadAsync() throws Exception {
 
   void runTestStaleReadAsync(CLUSTER cluster) throws Exception {
     final int numMessages = 10;
-    try (RaftClient client = cluster.createClient()) {
-      RaftTestUtil.waitForLeader(cluster);
+    RaftServer.Division division = waitForLeader(cluster);
+    try (RaftClient client = cluster.createClient(division.getId())) {
 
       // submit some messages
       final List<CompletableFuture<RaftClientReply>> futures = new ArrayList<>();
@@ -306,6 +311,7 @@ void runTestStaleReadAsync(CLUSTER cluster) throws Exception {
       // Use a follower with the max commit index
       final RaftClientReply lastWriteReply = replies.get(replies.size() - 1);
       final RaftPeerId leader = lastWriteReply.getServerId();
+      Assertions.assertEquals(leader, lastWriteReply.getServerId());
       LOG.info("leader = " + leader);
       final Collection<CommitInfoProto> commitInfos = lastWriteReply.getCommitInfos();
       LOG.info("commitInfos = " + commitInfos);
@@ -368,8 +374,8 @@ public void testWriteAsyncCustomReplicationLevel() throws Exception {
 
   void runTestWriteAsyncCustomReplicationLevel(CLUSTER cluster) throws Exception {
     final int numMessages = 20;
-    try (RaftClient client = cluster.createClient()) {
-      RaftTestUtil.waitForLeader(cluster);
+    final RaftPeerId leader = waitForLeader(cluster).getId();
+    try (RaftClient client = cluster.createClient(leader)) {
 
       // submit some messages
       for (int i = 0; i < numMessages; i++) {
@@ -419,13 +425,13 @@ void runTestAppendEntriesTimeout(CLUSTER cluster) throws Exception {
     LOG.info("Running testAppendEntriesTimeout");
     final TimeDuration oldExpiryTime = RaftServerConfigKeys.RetryCache.expiryTime(getProperties());
     RaftServerConfigKeys.RetryCache.setExpiryTime(getProperties(), TimeDuration.valueOf(20, TimeUnit.SECONDS));
-    RaftPeerId id = waitForLeader(cluster).getId();
+    final RaftPeerId leader = waitForLeader(cluster).getId();
     long time = System.currentTimeMillis();
     long waitTime = 5000;
     try (final RaftClient client = cluster.createClient()) {
       // block append requests
       cluster.getServerAliveStream()
-          .filter(impl -> !impl.getInfo().isLeader() && impl.getPeer().getId() != id)
+          .filter(impl -> !impl.getInfo().isLeader() && !impl.getPeer().getId().equals(leader))
           .map(SimpleStateMachine4Testing::get)
           .forEach(SimpleStateMachine4Testing::blockWriteStateMachineData);
 
@@ -435,7 +441,7 @@ void runTestAppendEntriesTimeout(CLUSTER cluster) throws Exception {
       Assertions.assertFalse(replyFuture.isDone());
       // unblock append request.
       cluster.getServerAliveStream()
-          .filter(impl -> !impl.getInfo().isLeader() && impl.getPeer().getId() != id)
+          .filter(impl -> !impl.getInfo().isLeader() && !impl.getPeer().getId().equals(leader))
           .map(SimpleStateMachine4Testing::get)
           .forEach(SimpleStateMachine4Testing::unblockWriteStateMachineData);
 

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

@@ -457,7 +457,7 @@ public void testStateMachineMetrics() throws Exception {
 
   static void runTestStateMachineMetrics(boolean async, MiniRaftCluster cluster) throws Exception {
     RaftServer.Division leader = waitForLeader(cluster);
-    try (final RaftClient client = cluster.createClient()) {
+    try (final RaftClient client = cluster.createClient(leader.getId())) {
       Gauge appliedIndexGauge = getStatemachineGaugeWithName(leader,
           STATEMACHINE_APPLIED_INDEX_GAUGE);
       Gauge smAppliedIndexGauge = getStatemachineGaugeWithName(leader,