RATIS-2294. Improve Some Code.

Author: slfan1989

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

@@ -46,23 +46,21 @@ 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 +73,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

@@ -309,6 +309,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();
+      Assert.assertEquals(leader, lastWriteReply.getServerId());
       LOG.info("leader = " + leader);
       final Collection<CommitInfoProto> commitInfos = lastWriteReply.getCommitInfos();
       LOG.info("commitInfos = " + commitInfos);
@@ -371,8 +372,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++) {
@@ -422,13 +423,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);
 
@@ -438,7 +439,7 @@ void runTestAppendEntriesTimeout(CLUSTER cluster) throws Exception {
       Assert.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);