[incubator-ratis] branch master updated: RATIS-776. Handle ResourceUnavailabeException properly in Ratis Server. Contributed by Lokesh Jain.

[lj...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

ljain pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-ratis.git


The following commit(s) were added to refs/heads/master by this push:
     new 66774b5  RATIS-776. Handle ResourceUnavailabeException properly in Ratis Server. Contributed by Lokesh Jain.
66774b5 is described below

commit 66774b5e4a4f655bd81a55686db9578fe60773a2
Author: Lokesh Jain <lj...@apache.org>
AuthorDate: Mon Jan 6 21:25:18 2020 +0530

    RATIS-776. Handle ResourceUnavailabeException properly in Ratis Server. Contributed by Lokesh Jain.
---
 .../grpc/client/GrpcClientProtocolService.java     |  9 +++
 .../apache/ratis/server/impl/RaftServerImpl.java   |  6 +-
 .../apache/ratis/grpc/TestRetryCacheWithGrpc.java  | 67 ++++++++++++++++++++++
 3 files changed, 80 insertions(+), 2 deletions(-)

diff --git a/ratis-grpc/src/main/java/org/apache/ratis/grpc/client/GrpcClientProtocolService.java b/ratis-grpc/src/main/java/org/apache/ratis/grpc/client/GrpcClientProtocolService.java
index 68c1b3e..d878c8d 100644
--- a/ratis-grpc/src/main/java/org/apache/ratis/grpc/client/GrpcClientProtocolService.java
+++ b/ratis-grpc/src/main/java/org/apache/ratis/grpc/client/GrpcClientProtocolService.java
@@ -215,6 +215,10 @@ public class GrpcClientProtocolService extends RaftClientProtocolServiceImplBase
       return isClosed.compareAndSet(false, true);
     }
 
+    boolean isClosed() {
+      return isClosed.get();
+    }
+
     CompletableFuture<Void> processClientRequest(RaftClientRequest request, Consumer<RaftClientReply> replyHandler) {
       try {
         String errMsg = LOG.isDebugEnabled() ? "processClientRequest for " + request.toString() : "";
@@ -330,6 +334,11 @@ public class GrpcClientProtocolService extends RaftClientProtocolServiceImplBase
 
     @Override
     void processClientRequest(RaftClientRequest r) {
+      if (isClosed()) {
+        final AlreadyClosedException exception = new AlreadyClosedException(getName() + ": the stream is closed");
+        responseError(exception, () -> "processClientRequest (stream already closed) for " + r);
+      }
+
       final RaftGroupId requestGroupId = r.getRaftGroupId();
       // use the group id in the first request as the group id of this observer
       final RaftGroupId updated = groupId.updateAndGet(g -> g != null ? g: requestGroupId);
diff --git a/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java b/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java
index d3c4e4f..364cb9e 100644
--- a/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java
+++ b/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java
@@ -509,8 +509,9 @@ public class RaftServerImpl implements RaftServerProtocol, RaftServerAsynchronou
       final LeaderState leaderState = role.getLeaderStateNonNull();
       final PendingRequests.Permit permit = leaderState.tryAcquirePendingRequest(request.getMessage());
       if (permit == null) {
-        return JavaUtils.completeExceptionally(new ResourceUnavailableException(
+        cacheEntry.failWithException(new ResourceUnavailableException(
             getMemberId() + ": Failed to acquire a pending write request for " + request));
+        return cacheEntry.getReplyFuture();
       }
       try {
         state.appendLog(context);
@@ -529,8 +530,9 @@ public class RaftServerImpl implements RaftServerProtocol, RaftServerAsynchronou
       // put the request into the pending queue
       pending = leaderState.addPendingRequest(permit, request, context);
       if (pending == null) {
-        return JavaUtils.completeExceptionally(new ResourceUnavailableException(
+        cacheEntry.failWithException(new ResourceUnavailableException(
             getMemberId() + ": Failed to add a pending write request for " + request));
+        return cacheEntry.getReplyFuture();
       }
       leaderState.notifySenders();
     }
diff --git a/ratis-test/src/test/java/org/apache/ratis/grpc/TestRetryCacheWithGrpc.java b/ratis-test/src/test/java/org/apache/ratis/grpc/TestRetryCacheWithGrpc.java
index 579cd24..192b25c 100644
--- a/ratis-test/src/test/java/org/apache/ratis/grpc/TestRetryCacheWithGrpc.java
+++ b/ratis-test/src/test/java/org/apache/ratis/grpc/TestRetryCacheWithGrpc.java
@@ -17,9 +17,76 @@
  */
 package org.apache.ratis.grpc;
 
+import org.apache.ratis.MiniRaftCluster;
+import org.apache.ratis.RaftTestUtil;
 import org.apache.ratis.RetryCacheTests;
+import org.apache.ratis.conf.RaftProperties;
+import org.apache.ratis.protocol.ClientId;
+import org.apache.ratis.protocol.RaftClientReply;
+import org.apache.ratis.protocol.RaftClientRequest;
+import org.apache.ratis.protocol.exceptions.ResourceUnavailableException;
+import org.apache.ratis.server.RaftServerConfigKeys;
+import org.apache.ratis.server.impl.RaftServerImpl;
+import org.apache.ratis.server.impl.RaftServerProxy;
+import org.apache.ratis.server.impl.RaftServerTestUtil;
+import org.apache.ratis.statemachine.SimpleStateMachine4Testing;
+import org.apache.ratis.statemachine.StateMachine;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 public class TestRetryCacheWithGrpc
     extends RetryCacheTests<MiniRaftClusterWithGrpc>
     implements MiniRaftClusterWithGrpc.FactoryGet {
+
+  @Test(timeout = 10000)
+  public void testRetryOnResourceUnavailableException()
+      throws InterruptedException, IOException {
+    RaftProperties properties = new RaftProperties();
+    properties.setClass(MiniRaftCluster.STATEMACHINE_CLASS_KEY,
+        SimpleStateMachine4Testing.class, StateMachine.class);
+    RaftServerConfigKeys.Write.setElementLimit(properties, 1);
+    MiniRaftClusterWithGrpc cluster = getFactory().newCluster(NUM_SERVERS, properties);
+    cluster.start();
+
+    RaftServerImpl leader = RaftTestUtil.waitForLeader(cluster);
+    RaftServerProxy leaderProxy = leader.getProxy();
+    for (RaftServerImpl follower : cluster.getFollowers()) {
+      // block followers to trigger ResourceUnavailableException
+      ((SimpleStateMachine4Testing) follower.getStateMachine()).blockWriteStateMachineData();
+    }
+    AtomicBoolean failure = new AtomicBoolean(false);
+    long callId = 1;
+    ClientId clientId = ClientId.randomId();
+    RaftClientRequest r = null;
+    while (!failure.get()) {
+      long cid = callId;
+      r = cluster.newRaftClientRequest(clientId, leaderProxy.getId(), callId++,
+          new RaftTestUtil.SimpleMessage("message"));
+      CompletableFuture<RaftClientReply> f = leaderProxy.submitClientRequestAsync(r);
+      f.exceptionally(e -> {
+        if (e.getCause() instanceof ResourceUnavailableException) {
+          RaftServerTestUtil.isRetryCacheEntryFailed(RaftServerTestUtil.getRetryEntry(leader, clientId, cid));
+          failure.set(true);
+        }
+        return null;
+      });
+    }
+    for (RaftServerImpl follower : cluster.getFollowers()) {
+      // unblock followers
+      ((SimpleStateMachine4Testing)follower.getStateMachine()).unblockWriteStateMachineData();
+    }
+
+    while (failure.get()) {
+      try {
+        // retry until the request failed with ResourceUnavailableException succeeds.
+        leaderProxy.submitClientRequestAsync(r).get();
+      } catch (Exception e) {
+        failure.set(false);
+      }
+    }
+    cluster.shutdown();
+  }
 }