Repository: incubator-ratis Updated Branches: refs/heads/master cbfa28a02 -> 1185121ab
RATIS-214. Support timeout using withDeadlineAfter for the blockingStub(Grpc). Contributed by Lokesh Jain Project: http://git-wip-us.apache.org/repos/asf/incubator-ratis/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-ratis/commit/1185121a Tree: http://git-wip-us.apache.org/repos/asf/incubator-ratis/tree/1185121a Diff: http://git-wip-us.apache.org/repos/asf/incubator-ratis/diff/1185121a Branch: refs/heads/master Commit: 1185121ab57cb64f14947510f7e7ebafe8c6ef54 Parents: cbfa28a Author: Tsz-Wo Nicholas Sze <[email protected]> Authored: Wed Mar 21 11:38:19 2018 +0800 Committer: Tsz-Wo Nicholas Sze <[email protected]> Committed: Wed Mar 21 11:38:19 2018 +0800 ---------------------------------------------------------------------- .../ratis/client/impl/RaftClientImpl.java | 4 ++ .../ratis/client/impl/RaftClientTestUtil.java | 4 ++ .../apache/ratis/grpc/client/GrpcClientRpc.java | 2 +- .../grpc/client/RaftClientProtocolClient.java | 18 +++++++-- .../grpc/server/RaftServerProtocolClient.java | 11 +++++- .../org/apache/ratis/grpc/TestRaftWithGrpc.java | 7 ++++ .../java/org/apache/ratis/RaftBasicTests.java | 39 ++++++++++++++++++++ .../ratis/server/impl/RetryCacheTestUtil.java | 4 ++ 8 files changed, 83 insertions(+), 6 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/1185121a/ratis-client/src/main/java/org/apache/ratis/client/impl/RaftClientImpl.java ---------------------------------------------------------------------- diff --git a/ratis-client/src/main/java/org/apache/ratis/client/impl/RaftClientImpl.java b/ratis-client/src/main/java/org/apache/ratis/client/impl/RaftClientImpl.java index ca1f057..b7aae19 100644 --- a/ratis-client/src/main/java/org/apache/ratis/client/impl/RaftClientImpl.java +++ b/ratis-client/src/main/java/org/apache/ratis/client/impl/RaftClientImpl.java @@ -385,6 +385,10 @@ final class RaftClientImpl implements RaftClient { Preconditions.assertTrue(((ScheduledThreadPoolExecutor) scheduler).getCorePoolSize() == numThreads); } + long getCallId() { + return callIdCounter.get(); + } + @Override public RaftClientRpc getClientRpc() { return clientRpc; http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/1185121a/ratis-client/src/main/java/org/apache/ratis/client/impl/RaftClientTestUtil.java ---------------------------------------------------------------------- diff --git a/ratis-client/src/main/java/org/apache/ratis/client/impl/RaftClientTestUtil.java b/ratis-client/src/main/java/org/apache/ratis/client/impl/RaftClientTestUtil.java index cab2dd0..20647de 100644 --- a/ratis-client/src/main/java/org/apache/ratis/client/impl/RaftClientTestUtil.java +++ b/ratis-client/src/main/java/org/apache/ratis/client/impl/RaftClientTestUtil.java @@ -29,4 +29,8 @@ public interface RaftClientTestUtil { static void assertScheduler(RaftClient client, int numThreads){ ((RaftClientImpl) client).assertScheduler(numThreads); } + + static long getCallId(RaftClient client) { + return ((RaftClientImpl) client).getCallId(); + } } http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/1185121a/ratis-grpc/src/main/java/org/apache/ratis/grpc/client/GrpcClientRpc.java ---------------------------------------------------------------------- diff --git a/ratis-grpc/src/main/java/org/apache/ratis/grpc/client/GrpcClientRpc.java b/ratis-grpc/src/main/java/org/apache/ratis/grpc/client/GrpcClientRpc.java index c235997..513638a 100644 --- a/ratis-grpc/src/main/java/org/apache/ratis/grpc/client/GrpcClientRpc.java +++ b/ratis-grpc/src/main/java/org/apache/ratis/grpc/client/GrpcClientRpc.java @@ -107,7 +107,7 @@ public class GrpcClientRpc extends RaftClientRpcWithProxy<RaftClientProtocolClie new CompletableFuture<>(); // create a new grpc stream for each non-async call. final StreamObserver<RaftClientRequestProto> requestObserver = - proxy.append(new StreamObserver<RaftClientReplyProto>() { + proxy.appendWithTimeout(new StreamObserver<RaftClientReplyProto>() { @Override public void onNext(RaftClientReplyProto value) { replyFuture.complete(value); http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/1185121a/ratis-grpc/src/main/java/org/apache/ratis/grpc/client/RaftClientProtocolClient.java ---------------------------------------------------------------------- diff --git a/ratis-grpc/src/main/java/org/apache/ratis/grpc/client/RaftClientProtocolClient.java b/ratis-grpc/src/main/java/org/apache/ratis/grpc/client/RaftClientProtocolClient.java index 97fc9a3..feb5cc5 100644 --- a/ratis-grpc/src/main/java/org/apache/ratis/grpc/client/RaftClientProtocolClient.java +++ b/ratis-grpc/src/main/java/org/apache/ratis/grpc/client/RaftClientProtocolClient.java @@ -34,6 +34,7 @@ import org.apache.ratis.util.CheckedSupplier; import org.apache.ratis.util.CollectionUtils; import org.apache.ratis.util.JavaUtils; import org.apache.ratis.util.SizeInBytes; +import org.apache.ratis.util.TimeDuration; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -42,6 +43,7 @@ import java.io.IOException; import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; import java.util.function.Supplier; @@ -51,6 +53,7 @@ public class RaftClientProtocolClient implements Closeable { private final Supplier<String> name; private final RaftPeer target; private final ManagedChannel channel; + private TimeDuration timeout = TimeDuration.valueOf(3, TimeUnit.SECONDS); private final RaftClientProtocolServiceBlockingStub blockingStub; private final RaftClientProtocolServiceStub asyncStub; private final AdminProtocolServiceBlockingStub adminBlockingStub; @@ -85,17 +88,20 @@ public class RaftClientProtocolClient implements Closeable { RaftClientReplyProto reinitialize( ReinitializeRequestProto request) throws IOException { - return blockingCall(() -> adminBlockingStub.reinitialize(request)); + TimeUnit unit = timeout.getUnit(); + return blockingCall(() -> adminBlockingStub.withDeadlineAfter(timeout.toInt(unit), unit).reinitialize(request)); } ServerInformationReplyProto serverInformation( ServerInformationRequestProto request) throws IOException { - return adminBlockingStub.serverInformation(request); + TimeUnit unit = timeout.getUnit(); + return adminBlockingStub.withDeadlineAfter(timeout.toInt(unit), unit).serverInformation(request); } RaftClientReplyProto setConfiguration( SetConfigurationRequestProto request) throws IOException { - return blockingCall(() -> blockingStub.setConfiguration(request)); + TimeUnit unit = timeout.getUnit(); + return blockingCall(() -> blockingStub.withDeadlineAfter(timeout.toInt(unit), unit).setConfiguration(request)); } private static RaftClientReplyProto blockingCall( @@ -113,6 +119,12 @@ public class RaftClientProtocolClient implements Closeable { return asyncStub.append(responseHandler); } + StreamObserver<RaftClientRequestProto> appendWithTimeout( + StreamObserver<RaftClientReplyProto> responseHandler) { + TimeUnit unit = timeout.getUnit(); + return asyncStub.withDeadlineAfter(timeout.toInt(unit), unit).append(responseHandler); + } + AsyncStreamObservers getAppendStreamObservers() { return appendStreamObservers.updateAndGet(a -> a != null? a : new AsyncStreamObservers()); } http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/1185121a/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/RaftServerProtocolClient.java ---------------------------------------------------------------------- diff --git a/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/RaftServerProtocolClient.java b/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/RaftServerProtocolClient.java index 53f962a..42a2b85 100644 --- a/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/RaftServerProtocolClient.java +++ b/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/RaftServerProtocolClient.java @@ -25,6 +25,9 @@ import org.apache.ratis.shaded.proto.grpc.RaftServerProtocolServiceGrpc; import org.apache.ratis.shaded.proto.grpc.RaftServerProtocolServiceGrpc.RaftServerProtocolServiceBlockingStub; import org.apache.ratis.shaded.proto.grpc.RaftServerProtocolServiceGrpc.RaftServerProtocolServiceStub; import org.apache.ratis.protocol.RaftPeer; +import org.apache.ratis.util.TimeDuration; + +import java.util.concurrent.TimeUnit; /** * This is a RaftClient implementation that supports streaming data to the raft @@ -32,6 +35,7 @@ import org.apache.ratis.protocol.RaftPeer; */ public class RaftServerProtocolClient { private final ManagedChannel channel; + private TimeDuration timeout = TimeDuration.valueOf(3, TimeUnit.SECONDS); private final RaftServerProtocolServiceBlockingStub blockingStub; private final RaftServerProtocolServiceStub asyncStub; @@ -49,7 +53,9 @@ public class RaftServerProtocolClient { public RequestVoteReplyProto requestVote(RequestVoteRequestProto request) { // the StatusRuntimeException will be handled by the caller - return blockingStub.requestVote(request); + TimeUnit unit = timeout.getUnit(); + RequestVoteReplyProto r= blockingStub.withDeadlineAfter(timeout.toInt(unit), unit).requestVote(request); + return r; } StreamObserver<AppendEntriesRequestProto> appendEntries( @@ -59,6 +65,7 @@ public class RaftServerProtocolClient { StreamObserver<InstallSnapshotRequestProto> installSnapshot( StreamObserver<InstallSnapshotReplyProto> responseHandler) { - return asyncStub.installSnapshot(responseHandler); + TimeUnit unit = timeout.getUnit(); + return asyncStub.withDeadlineAfter(timeout.toInt(unit), unit).installSnapshot(responseHandler); } } http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/1185121a/ratis-grpc/src/test/java/org/apache/ratis/grpc/TestRaftWithGrpc.java ---------------------------------------------------------------------- diff --git a/ratis-grpc/src/test/java/org/apache/ratis/grpc/TestRaftWithGrpc.java b/ratis-grpc/src/test/java/org/apache/ratis/grpc/TestRaftWithGrpc.java index b79e41e..604e51a 100644 --- a/ratis-grpc/src/test/java/org/apache/ratis/grpc/TestRaftWithGrpc.java +++ b/ratis-grpc/src/test/java/org/apache/ratis/grpc/TestRaftWithGrpc.java @@ -23,6 +23,7 @@ import org.junit.Assert; import org.junit.Test; import java.io.IOException; +import java.util.concurrent.ExecutionException; public class TestRaftWithGrpc extends RaftBasicTests { private final MiniRaftClusterWithGRpc cluster; @@ -44,4 +45,10 @@ public class TestRaftWithGrpc extends RaftBasicTests { super.testWithLoad(); BlockRequestHandlingInjection.getInstance().unblockAll(); } + + @Test + public void testRequestTimeout() + throws IOException, InterruptedException, ExecutionException { + testRequestTimeout(false, getCluster(), LOG, getProperties()); + } } http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/1185121a/ratis-server/src/test/java/org/apache/ratis/RaftBasicTests.java ---------------------------------------------------------------------- diff --git a/ratis-server/src/test/java/org/apache/ratis/RaftBasicTests.java b/ratis-server/src/test/java/org/apache/ratis/RaftBasicTests.java index f79c857..8aa05cd 100644 --- a/ratis-server/src/test/java/org/apache/ratis/RaftBasicTests.java +++ b/ratis-server/src/test/java/org/apache/ratis/RaftBasicTests.java @@ -20,15 +20,19 @@ package org.apache.ratis; import org.apache.log4j.Level; import org.apache.ratis.RaftTestUtil.*; import org.apache.ratis.client.RaftClient; +import org.apache.ratis.client.impl.RaftClientTestUtil; import org.apache.ratis.conf.RaftProperties; import org.apache.ratis.protocol.RaftClientReply; import org.apache.ratis.protocol.RaftPeerId; +import org.apache.ratis.server.RaftServerConfigKeys; import org.apache.ratis.server.impl.BlockRequestHandlingInjection; import org.apache.ratis.server.impl.RaftServerImpl; +import org.apache.ratis.server.impl.RetryCacheTestUtil; import org.apache.ratis.server.storage.RaftLog; import org.apache.ratis.shaded.proto.RaftProtos.LogEntryProto; import org.apache.ratis.util.JavaUtils; import org.apache.ratis.util.LogUtils; +import org.apache.ratis.util.TimeDuration; import org.junit.After; import org.junit.Assert; import org.junit.Before; @@ -40,6 +44,8 @@ import java.util.List; import java.util.Timer; import java.util.TimerTask; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; @@ -54,6 +60,8 @@ public abstract class RaftBasicTests extends BaseTest { { LogUtils.setLogLevel(RaftServerImpl.LOG, Level.DEBUG); LogUtils.setLogLevel(RaftClient.LOG, Level.DEBUG); + RaftServerConfigKeys.RetryCache.setExpiryTime(properties, TimeDuration + .valueOf(10, TimeUnit.SECONDS)); } public static final int NUM_SERVERS = 5; @@ -365,4 +373,35 @@ public abstract class RaftBasicTests extends BaseTest { RaftTestUtil.assertLogEntries(cluster.getServers(), c.messages); } } + + public static void testRequestTimeout(boolean async, MiniRaftCluster cluster, Logger LOG, + RaftProperties properties) throws InterruptedException, IOException, ExecutionException { + LOG.info("Running testRequestTimeout"); + waitForLeader(cluster); + long time = System.currentTimeMillis(); + try (final RaftClient client = cluster.createClient()) { + // Get the next callId to be used by the client + long callId = RaftClientTestUtil.getCallId(client); + // Create an entry corresponding to the callId and clientId + // in each server's retry cache. + cluster.getServerAliveStream().forEach( + raftServer -> RetryCacheTestUtil.getOrCreateEntry(raftServer.getRetryCache(), client.getId(), callId)); + // Client request for the callId now waits + // as there is already a cache entry in the server for the request. + // Ideally the client request should timeout and the client should retry. + // The retry is successful when the retry cache entry for the corresponding callId and clientId expires. + if (async) { + CompletableFuture<RaftClientReply> replyFuture = client.sendAsync(new SimpleMessage("abc")); + replyFuture.get(); + } else { + client.send(new SimpleMessage("abc")); + } + // Eventually the request would be accepted by the server + // when the retry cache entry is invalidated. + // The duration for which the client waits should be more than the retryCacheExpiryDuration. + TimeDuration duration = TimeDuration.valueOf(System.currentTimeMillis() - time, TimeUnit.MILLISECONDS); + TimeDuration retryCacheExpiryDuration = RaftServerConfigKeys.RetryCache.expiryTime(properties); + Assert.assertTrue(duration.compareTo(retryCacheExpiryDuration) >= 0); + } + } } http://git-wip-us.apache.org/repos/asf/incubator-ratis/blob/1185121a/ratis-server/src/test/java/org/apache/ratis/server/impl/RetryCacheTestUtil.java ---------------------------------------------------------------------- diff --git a/ratis-server/src/test/java/org/apache/ratis/server/impl/RetryCacheTestUtil.java b/ratis-server/src/test/java/org/apache/ratis/server/impl/RetryCacheTestUtil.java index 0d28e2c..9e63063 100644 --- a/ratis-server/src/test/java/org/apache/ratis/server/impl/RetryCacheTestUtil.java +++ b/ratis-server/src/test/java/org/apache/ratis/server/impl/RetryCacheTestUtil.java @@ -45,4 +45,8 @@ public class RetryCacheTestUtil { Assert.assertEquals(isFailed, cache.get(clientId, callId).isFailed()); } } + + public static void getOrCreateEntry(RetryCache cache, ClientId clientId, long callId){ + cache.getOrCreateEntry(clientId, callId); + } }
