You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ratis.apache.org by sh...@apache.org on 2020/01/23 08:56:08 UTC
[incubator-ratis] branch master updated: RATIS-796. Add watch time
out for Ratis Client. Contributed by Bharat Viswanadham.
This is an automated email from the ASF dual-hosted git repository.
shashikant 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 e51d531 RATIS-796. Add watch time out for Ratis Client. Contributed by Bharat Viswanadham.
e51d531 is described below
commit e51d531081c60a755bfaadaa2ada6f77b24bfaf8
Author: Shashikant Banerjee <sh...@apache.org>
AuthorDate: Thu Jan 23 14:23:50 2020 +0530
RATIS-796. Add watch time out for Ratis Client. Contributed by Bharat Viswanadham.
---
.../apache/ratis/client/RaftClientConfigKeys.java | 12 ++++
.../grpc/client/GrpcClientProtocolClient.java | 18 +++--
.../java/org/apache/ratis/WatchRequestTests.java | 82 +++++++++++++++++++++-
3 files changed, 105 insertions(+), 7 deletions(-)
diff --git a/ratis-client/src/main/java/org/apache/ratis/client/RaftClientConfigKeys.java b/ratis-client/src/main/java/org/apache/ratis/client/RaftClientConfigKeys.java
index d2c3679..48042a4 100644
--- a/ratis-client/src/main/java/org/apache/ratis/client/RaftClientConfigKeys.java
+++ b/ratis-client/src/main/java/org/apache/ratis/client/RaftClientConfigKeys.java
@@ -48,6 +48,18 @@ public interface RaftClientConfigKeys {
static void setRequestTimeout(RaftProperties properties, TimeDuration timeoutDuration) {
setTimeDuration(properties::setTimeDuration, REQUEST_TIMEOUT_KEY, timeoutDuration);
}
+
+ String WATCH_REQUEST_TIMEOUT_KEY = PREFIX + ".watch.request.timeout";
+ TimeDuration WATCH_REQUEST_TIMEOUT_DEFAULT =
+ TimeDuration.valueOf(10000, TimeUnit.MILLISECONDS);
+ static TimeDuration watchRequestTimeout(RaftProperties properties) {
+ return getTimeDuration(properties.getTimeDuration(WATCH_REQUEST_TIMEOUT_DEFAULT.getUnit()),
+ WATCH_REQUEST_TIMEOUT_KEY, WATCH_REQUEST_TIMEOUT_DEFAULT, getDefaultLog());
+ }
+ static void setWatchRequestTimeout(RaftProperties properties,
+ TimeDuration timeoutDuration) {
+ setTimeDuration(properties::setTimeDuration, WATCH_REQUEST_TIMEOUT_KEY, timeoutDuration);
+ }
}
interface Async {
diff --git a/ratis-grpc/src/main/java/org/apache/ratis/grpc/client/GrpcClientProtocolClient.java b/ratis-grpc/src/main/java/org/apache/ratis/grpc/client/GrpcClientProtocolClient.java
index be99bb5..5aec0b5 100644
--- a/ratis-grpc/src/main/java/org/apache/ratis/grpc/client/GrpcClientProtocolClient.java
+++ b/ratis-grpc/src/main/java/org/apache/ratis/grpc/client/GrpcClientProtocolClient.java
@@ -80,6 +80,7 @@ public class GrpcClientProtocolClient implements Closeable {
private final ManagedChannel channel;
private final TimeDuration requestTimeoutDuration;
+ private final TimeDuration watchRequestTimeoutDuration;
private final TimeoutScheduler scheduler = TimeoutScheduler.getInstance();
private final RaftClientProtocolServiceBlockingStub blockingStub;
@@ -130,6 +131,8 @@ public class GrpcClientProtocolClient implements Closeable {
asyncStub = RaftClientProtocolServiceGrpc.newStub(channel);
adminBlockingStub = AdminProtocolServiceGrpc.newBlockingStub(channel);
this.requestTimeoutDuration = RaftClientConfigKeys.Rpc.requestTimeout(properties);
+ this.watchRequestTimeoutDuration =
+ RaftClientConfigKeys.Rpc.watchRequestTimeout(properties);
}
String getName() {
@@ -316,14 +319,21 @@ public class GrpcClientProtocolClient implements Closeable {
return f;
}
- scheduler.onTimeout(requestTimeoutDuration, () -> timeoutCheck(callId), LOG,
- () -> "Timeout check failed for client request #" + callId);
+ if (RaftClientRequestProto.TypeCase.WATCH.equals(request.getType().getTypeCase())) {
+ scheduler.onTimeout(watchRequestTimeoutDuration, () ->
+ timeoutCheck(callId, watchRequestTimeoutDuration), LOG,
+ () -> "Timeout check failed for client request #" + callId);
+ } else {
+ scheduler.onTimeout(requestTimeoutDuration,
+ () -> timeoutCheck(callId, requestTimeoutDuration), LOG,
+ () -> "Timeout check failed for client request #" + callId);
+ }
return f;
}
- private void timeoutCheck(long callId) {
+ private void timeoutCheck(long callId, TimeDuration timeOutDuration) {
handleReplyFuture(callId, f -> f.completeExceptionally(
- new TimeoutIOException("Request #" + callId + " timeout " + requestTimeoutDuration)));
+ new TimeoutIOException("Request #" + callId + " timeout " + timeOutDuration)));
}
private void handleReplyFuture(long callId, Consumer<CompletableFuture<RaftClientReply>> handler) {
diff --git a/ratis-server/src/test/java/org/apache/ratis/WatchRequestTests.java b/ratis-server/src/test/java/org/apache/ratis/WatchRequestTests.java
index 3a80c80..0f5d902 100644
--- a/ratis-server/src/test/java/org/apache/ratis/WatchRequestTests.java
+++ b/ratis-server/src/test/java/org/apache/ratis/WatchRequestTests.java
@@ -19,11 +19,15 @@ package org.apache.ratis;
import org.apache.log4j.Level;
import org.apache.ratis.client.RaftClient;
+import org.apache.ratis.client.RaftClientConfigKeys;
import org.apache.ratis.conf.RaftProperties;
import org.apache.ratis.proto.RaftProtos.CommitInfoProto;
import org.apache.ratis.proto.RaftProtos.ReplicationLevel;
import org.apache.ratis.protocol.NotReplicatedException;
import org.apache.ratis.protocol.RaftClientReply;
+import org.apache.ratis.protocol.RaftRetryFailureException;
+import org.apache.ratis.protocol.TimeoutIOException;
+import org.apache.ratis.retry.RetryPolicies;
import org.apache.ratis.server.RaftServerConfigKeys;
import org.apache.ratis.server.impl.RaftServerImpl;
import org.apache.ratis.server.impl.RaftServerTestUtil;
@@ -48,6 +52,8 @@ import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;
+import static org.junit.Assert.fail;
+
public abstract class WatchRequestTests<CLUSTER extends MiniRaftCluster>
extends BaseTest
implements MiniRaftCluster.Factory.Get<CLUSTER> {
@@ -107,6 +113,24 @@ public abstract class WatchRequestTests<CLUSTER extends MiniRaftCluster>
}
}
+ CompletableFuture<RaftClientReply> sendWatchRequest(long logIndex)
+ throws Exception {
+
+ try (final RaftClient watchClient =
+ cluster.createClient(RaftTestUtil.waitForLeader(cluster).getId(),
+ RetryPolicies.retryUpToMaximumCountWithFixedSleep(5,
+ TimeDuration.valueOf(5, TimeUnit.SECONDS)))) {
+
+ CompletableFuture<RaftClientReply> reply =
+ watchClient.sendAsync(new RaftTestUtil.SimpleMessage("message"));
+ long writeIndex = reply.get().getLogIndex();
+ Assert.assertTrue(writeIndex > 0);
+ watchClient.sendWatchAsync(writeIndex, ReplicationLevel.MAJORITY_COMMITTED);
+ return watchClient.sendWatchAsync(logIndex, ReplicationLevel.MAJORITY);
+ }
+
+ }
+
@Override
public String toString() {
return "numMessages=" + numMessages;
@@ -125,6 +149,19 @@ public abstract class WatchRequestTests<CLUSTER extends MiniRaftCluster>
}
}
+ static void runSingleTest(CheckedConsumer<TestParameters, Exception> testCase,
+ MiniRaftCluster cluster, Logger LOG)
+ throws Exception {
+ try(final RaftClient client = cluster.createClient(RaftTestUtil.waitForLeader(cluster).getId())) {
+ final int[] numMessages = {1};
+ for(int n : numMessages) {
+ final TestParameters p = new TestParameters(n, client, cluster, LOG);
+ LOG.info("{}) {}, {}", n, p, cluster.printServers());
+ testCase.accept(p);
+ }
+ }
+ }
+
static class WatchReplies {
private final long logIndex;
private final CompletableFuture<RaftClientReply> majority;
@@ -281,9 +318,9 @@ public abstract class WatchRequestTests<CLUSTER extends MiniRaftCluster>
futures.forEach(f -> {
if (f.isDone()) {
try {
- Assert.fail("Done unexpectedly: " + f.get());
+ fail("Done unexpectedly: " + f.get());
} catch(Exception e) {
- Assert.fail("Done unexpectedly and failed to get: " + e);
+ fail("Done unexpectedly and failed to get: " + e);
}
}
});
@@ -388,6 +425,45 @@ public abstract class WatchRequestTests<CLUSTER extends MiniRaftCluster>
LOG.info("unblock follower {}", blockedFollower.getId());
}
+ @Test
+ public void testWatchRequestClientTimeout() throws Exception {
+ final RaftProperties p = getProperties();
+ RaftServerConfigKeys.Watch.setTimeout(p, TimeDuration.valueOf(100,
+ TimeUnit.SECONDS));
+ RaftClientConfigKeys.Rpc.setWatchRequestTimeout(p,
+ TimeDuration.valueOf(15, TimeUnit.SECONDS));
+ try {
+ runWithNewCluster(NUM_SERVERS,
+ cluster -> runSingleTest(WatchRequestTests::runTestWatchRequestClientTimeout, cluster, LOG));
+ } finally {
+ RaftServerConfigKeys.Watch.setTimeout(p, RaftServerConfigKeys.Watch.TIMEOUT_DEFAULT);
+ RaftClientConfigKeys.Rpc.setWatchRequestTimeout(p,
+ RaftClientConfigKeys.Rpc.WATCH_REQUEST_TIMEOUT_DEFAULT);
+ }
+ }
+
+ static void runTestWatchRequestClientTimeout(TestParameters p) throws Exception {
+ final Logger LOG = p.log;
+
+ CompletableFuture<RaftClientReply> watchReply;
+ watchReply = p.sendWatchRequest(1000);
+
+ try {
+ watchReply.get();
+ fail("runTestWatchRequestClientTimeout failed");
+ } catch (Exception ex) {
+ LOG.error("error occurred", ex);
+ Assert.assertEquals(RaftRetryFailureException.class,
+ ex.getCause().getClass());
+ if (ex.getCause() != null) {
+ if (ex.getCause().getCause() != null) {
+ Assert.assertEquals(TimeoutIOException.class,
+ ex.getCause().getCause().getClass());
+ }
+ }
+ }
+ }
+
static void checkTimeout(List<CompletableFuture<RaftClientReply>> replies,
List<CompletableFuture<WatchReplies>> watches, Logger LOG) throws Exception {
for(int i = 0; i < replies.size(); i++) {
@@ -409,7 +485,7 @@ public abstract class WatchRequestTests<CLUSTER extends MiniRaftCluster>
CheckedSupplier<RaftClientReply, Exception> replySupplier) throws Exception {
try {
replySupplier.get();
- Assert.fail();
+ fail();
} catch (ExecutionException e) {
final Throwable cause = e.getCause();
assertNotReplicatedException(logIndex, replication, cause);