This is an automated email from the ASF dual-hosted git repository.
williamsong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ratis.git
The following commit(s) were added to refs/heads/master by this push:
new bd98fa3ac RATIS-1864. Support lease based read-only requests (#928)
bd98fa3ac is described below
commit bd98fa3ac097b32f78f060ee1d2c6fe33c7f4c8b
Author: William Song <[email protected]>
AuthorDate: Sun Oct 1 09:23:35 2023 +0800
RATIS-1864. Support lease based read-only requests (#928)
---
.../main/java/org/apache/ratis/conf/ConfUtils.java | 9 ++
.../main/java/org/apache/ratis/util/Timestamp.java | 5 +
ratis-docs/src/site/markdown/configurations.md | 15 +++
.../apache/ratis/grpc/server/GrpcLogAppender.java | 11 ++-
.../apache/ratis/server/RaftServerConfigKeys.java | 22 +++++
.../apache/ratis/server/leader/FollowerInfo.java | 6 ++
.../apache/ratis/server/impl/FollowerInfoImpl.java | 12 +++
.../org/apache/ratis/server/impl/LeaderLease.java | 104 +++++++++++++++++++++
.../apache/ratis/server/impl/LeaderStateImpl.java | 37 ++++++++
.../apache/ratis/server/impl/RaftServerImpl.java | 17 ++--
.../ratis/server/impl/TransferLeadership.java | 9 ++
.../ratis/server/leader/LogAppenderDefault.java | 3 +
.../org/apache/ratis/ReadOnlyRequestTests.java | 71 ++++++++++----
.../ratis/server/impl/LeaderElectionTests.java | 84 +++++++++++++++++
.../ratis/server/impl/RaftServerTestUtil.java | 6 ++
15 files changed, 384 insertions(+), 27 deletions(-)
diff --git a/ratis-common/src/main/java/org/apache/ratis/conf/ConfUtils.java
b/ratis-common/src/main/java/org/apache/ratis/conf/ConfUtils.java
index 629d5a389..a642ebb9d 100644
--- a/ratis-common/src/main/java/org/apache/ratis/conf/ConfUtils.java
+++ b/ratis-common/src/main/java/org/apache/ratis/conf/ConfUtils.java
@@ -76,6 +76,15 @@ public interface ConfUtils {
};
}
+ static BiConsumer<String, Double> requireMin(double min) {
+ return (key, value) -> {
+ if (value < min) {
+ throw new IllegalArgumentException(
+ key + " = " + value + " < min = " + min);
+ }
+ };
+ }
+
static BiConsumer<String, Double> requireMax(double max) {
return (key, value) -> {
if (value > max) {
diff --git a/ratis-common/src/main/java/org/apache/ratis/util/Timestamp.java
b/ratis-common/src/main/java/org/apache/ratis/util/Timestamp.java
index ba5fb8c5b..cff143eef 100644
--- a/ratis-common/src/main/java/org/apache/ratis/util/Timestamp.java
+++ b/ratis-common/src/main/java/org/apache/ratis/util/Timestamp.java
@@ -51,6 +51,11 @@ public final class Timestamp implements
Comparable<Timestamp> {
return a.compareTo(b) > 0? a: b;
}
+ /** @return the earliest timestamp. */
+ public static Timestamp earliest(Timestamp a, Timestamp b) {
+ return a.compareTo(b) > 0? b: a;
+ }
+
private final long nanos;
private Timestamp(long nanos) {
diff --git a/ratis-docs/src/site/markdown/configurations.md
b/ratis-docs/src/site/markdown/configurations.md
index 4e37cfd76..0500a053a 100644
--- a/ratis-docs/src/site/markdown/configurations.md
+++ b/ratis-docs/src/site/markdown/configurations.md
@@ -185,6 +185,21 @@ treat the peer as caught-up. Increase this number when
write throughput is high.
--------------------------------------------------------------------------------
+| **Property** | `raft.server.read.leader.lease.enabled`
|
+|:----------------|:-----------------------------------------------------------|
+| **Description** | whether to enable lease in linearizable read-only requests
|
+| **Type** | boolean
|
+| **Default** | true
|
+
+--------------------------------------------------------------------------------
+
+| **Property** | `raft.server.read.leader.lease.timeout.ratio` |
+|:----------------|:----------------------------------------------|
+| **Description** | maximum timeout ratio of leader lease |
+| **Type** | double, ranging from (0.0,1.0) |
+| **Default** | 0.9 |
+
+
### Write - Configurations related to write requests.
* Limits on pending write requests
diff --git
a/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcLogAppender.java
b/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcLogAppender.java
index 8bc616ad9..07aecf6fe 100644
--- a/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcLogAppender.java
+++ b/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcLogAppender.java
@@ -390,7 +390,9 @@ public class GrpcLogAppender extends LogAppenderBase {
AppendEntriesRequest request = pendingRequests.remove(reply);
if (request != null) {
request.stopRequestTimer(); // Update completion time
+
getFollower().updateLastRespondedAppendEntriesSendTime(request.getSendTime());
}
+ getFollower().updateLastRpcResponseTime();
if (LOG.isDebugEnabled()) {
LOG.debug("{}: received {} reply {}, request={}",
@@ -407,8 +409,6 @@ public class GrpcLogAppender extends LogAppenderBase {
}
private void onNextImpl(AppendEntriesReplyProto reply) {
- // update the last rpc time
- getFollower().updateLastRpcResponseTime();
errCount.set(0);
if (!firstResponseReceived) {
@@ -770,6 +770,8 @@ public class GrpcLogAppender extends LogAppenderBase {
private final TermIndex lastEntry;
+ private volatile Timestamp sendTime;
+
AppendEntriesRequest(AppendEntriesRequestProto proto, RaftPeerId
followerId, GrpcServerMetrics grpcServerMetrics) {
this.callId = proto.getServerRequest().getCallId();
this.previousLog = proto.hasPreviousLog()?
TermIndex.valueOf(proto.getPreviousLog()): null;
@@ -788,8 +790,13 @@ public class GrpcLogAppender extends LogAppenderBase {
return previousLog;
}
+ public Timestamp getSendTime() {
+ return sendTime;
+ }
+
void startRequestTimer() {
timerContext = timer.time();
+ sendTime = Timestamp.currentTime();
}
void stopRequestTimer() {
diff --git
a/ratis-server-api/src/main/java/org/apache/ratis/server/RaftServerConfigKeys.java
b/ratis-server-api/src/main/java/org/apache/ratis/server/RaftServerConfigKeys.java
index a8a7892dc..cd38e5667 100644
---
a/ratis-server-api/src/main/java/org/apache/ratis/server/RaftServerConfigKeys.java
+++
b/ratis-server-api/src/main/java/org/apache/ratis/server/RaftServerConfigKeys.java
@@ -192,6 +192,28 @@ public interface RaftServerConfigKeys {
set(properties::setEnum, OPTION_KEY, option);
}
+ String LEADER_LEASE_ENABLED_KEY = PREFIX + ".leader.lease.enabled";
+ boolean LEADER_LEASE_ENABLED_DEFAULT = false;
+ static boolean leaderLeaseEnabled(RaftProperties properties) {
+ return getBoolean(properties::getBoolean, LEADER_LEASE_ENABLED_KEY,
+ LEADER_LEASE_ENABLED_DEFAULT, getDefaultLog());
+ }
+ static void setLeaderLeaseEnabled(RaftProperties properties, boolean
enabled) {
+ setBoolean(properties::setBoolean, LEADER_LEASE_ENABLED_KEY, enabled);
+ }
+
+ String LEADER_LEASE_TIMEOUT_RATIO_KEY = PREFIX +
".leader.lease.timeout.ratio";
+ double LEADER_LEASE_TIMEOUT_RATIO_DEFAULT = 0.9;
+ static double leaderLeaseTimeoutRatio(RaftProperties properties) {
+ return getDouble(properties::getDouble, LEADER_LEASE_TIMEOUT_RATIO_KEY,
+ LEADER_LEASE_TIMEOUT_RATIO_DEFAULT, getDefaultLog(),
+ requireMin(0.0), requireMax(1.0));
+ }
+
+ static void setLeaderLeaseTimeoutRatio(RaftProperties properties, double
ratio) {
+ setDouble(properties::setDouble, LEADER_LEASE_TIMEOUT_RATIO_KEY, ratio);
+ }
+
interface ReadAfterWriteConsistent {
String PREFIX = RaftServerConfigKeys.PREFIX +
".read-after-write-consistent";
diff --git
a/ratis-server-api/src/main/java/org/apache/ratis/server/leader/FollowerInfo.java
b/ratis-server-api/src/main/java/org/apache/ratis/server/leader/FollowerInfo.java
index 1dd4066e8..9d5c891d9 100644
---
a/ratis-server-api/src/main/java/org/apache/ratis/server/leader/FollowerInfo.java
+++
b/ratis-server-api/src/main/java/org/apache/ratis/server/leader/FollowerInfo.java
@@ -101,4 +101,10 @@ public interface FollowerInfo {
/** @return the latest heartbeat send time. */
Timestamp getLastHeartbeatSendTime();
+
+ /** @return the send time of last responded rpc */
+ Timestamp getLastRespondedAppendEntriesSendTime();
+
+ /** Update lastRpcResponseTime and LastRespondedAppendEntriesSendTime */
+ void updateLastRespondedAppendEntriesSendTime(Timestamp sendTime);
}
diff --git
a/ratis-server/src/main/java/org/apache/ratis/server/impl/FollowerInfoImpl.java
b/ratis-server/src/main/java/org/apache/ratis/server/impl/FollowerInfoImpl.java
index 245cbc888..91ab90a20 100644
---
a/ratis-server/src/main/java/org/apache/ratis/server/impl/FollowerInfoImpl.java
+++
b/ratis-server/src/main/java/org/apache/ratis/server/impl/FollowerInfoImpl.java
@@ -39,6 +39,7 @@ class FollowerInfoImpl implements FollowerInfo {
private final AtomicReference<Timestamp> lastRpcResponseTime;
private final AtomicReference<Timestamp> lastRpcSendTime;
private final AtomicReference<Timestamp> lastHeartbeatSendTime;
+ private final AtomicReference<Timestamp> lastRespondedAppendEntriesSendTime;
private final RaftLogIndex nextIndex;
private final RaftLogIndex matchIndex = new RaftLogIndex("matchIndex",
RaftLog.INVALID_LOG_INDEX);
private final RaftLogIndex commitIndex = new RaftLogIndex("commitIndex",
RaftLog.INVALID_LOG_INDEX);
@@ -57,6 +58,7 @@ class FollowerInfoImpl implements FollowerInfo {
this.lastRpcResponseTime = new AtomicReference<>(lastRpcTime);
this.lastRpcSendTime = new AtomicReference<>(lastRpcTime);
this.lastHeartbeatSendTime = new AtomicReference<>(lastRpcTime);
+ this.lastRespondedAppendEntriesSendTime = new
AtomicReference<>(lastRpcTime);
this.nextIndex = new RaftLogIndex("nextIndex", nextIndex);
this.caughtUp = caughtUp;
}
@@ -202,4 +204,14 @@ class FollowerInfoImpl implements FollowerInfo {
public Timestamp getLastHeartbeatSendTime() {
return lastHeartbeatSendTime.get();
}
+
+ @Override
+ public Timestamp getLastRespondedAppendEntriesSendTime() {
+ return lastRespondedAppendEntriesSendTime.get();
+ }
+
+ @Override
+ public void updateLastRespondedAppendEntriesSendTime(Timestamp sendTime) {
+ lastRespondedAppendEntriesSendTime.set(sendTime);
+ }
}
diff --git
a/ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderLease.java
b/ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderLease.java
new file mode 100644
index 000000000..315cc9f14
--- /dev/null
+++ b/ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderLease.java
@@ -0,0 +1,104 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.ratis.server.impl;
+
+import org.apache.ratis.conf.RaftProperties;
+import org.apache.ratis.protocol.RaftPeerId;
+import org.apache.ratis.server.RaftServerConfigKeys;
+import org.apache.ratis.server.leader.FollowerInfo;
+import org.apache.ratis.util.Preconditions;
+import org.apache.ratis.util.Timestamp;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+class LeaderLease {
+
+ private final AtomicBoolean enabled;
+ private final long leaseTimeoutMs;
+ private final AtomicReference<Timestamp> lease = new
AtomicReference<>(Timestamp.currentTime());
+
+ LeaderLease(RaftProperties properties) {
+ this.enabled = new
AtomicBoolean(RaftServerConfigKeys.Read.leaderLeaseEnabled(properties));
+ final double leaseRatio =
RaftServerConfigKeys.Read.leaderLeaseTimeoutRatio(properties);
+ Preconditions.assertTrue(leaseRatio > 0.0 && leaseRatio <= 1.0,
+ "leader ratio should sit in (0,1], now is " + leaseRatio);
+ this.leaseTimeoutMs = RaftServerConfigKeys.Rpc.timeoutMin(properties)
+ .multiply(leaseRatio)
+ .toIntExact(TimeUnit.MILLISECONDS);
+ }
+
+ boolean getAndSetEnabled(boolean newValue) {
+ return enabled.getAndSet(newValue);
+ }
+
+ boolean isEnabled() {
+ return enabled.get();
+ }
+
+ boolean isValid() {
+ return isEnabled() && lease.get().elapsedTimeMs() < leaseTimeoutMs;
+ }
+
+ /**
+ * try extending the lease based on group heartbeats
+ * @param old nullable
+ */
+ void extend(List<FollowerInfo> current, List<FollowerInfo> old,
Predicate<List<RaftPeerId>> hasMajority) {
+ final List<RaftPeerId> activePeers =
+ // check the latest heartbeats of all peers (including those in
transitional)
+ Stream.concat(current.stream(),
Optional.ofNullable(old).map(List::stream).orElse(Stream.empty()))
+ .filter(f ->
f.getLastRespondedAppendEntriesSendTime().elapsedTimeMs() < leaseTimeoutMs)
+ .map(FollowerInfo::getId)
+ .collect(Collectors.toList());
+
+ if (!hasMajority.test(activePeers)) {
+ return;
+ }
+
+ // update the new lease
+ final Timestamp newLease =
+ Timestamp.earliest(getMaxTimestampWithMajorityAck(current),
getMaxTimestampWithMajorityAck(old));
+ lease.set(newLease);
+ }
+
+ /**
+ * return maximum timestamp at when the majority of followers are known to
be active
+ * return {@link Timestamp#currentTime()} if peers are empty
+ */
+ private Timestamp getMaxTimestampWithMajorityAck(List<FollowerInfo>
followers) {
+ if (followers == null || followers.isEmpty()) {
+ return Timestamp.currentTime();
+ }
+
+ final int mid = followers.size() / 2;
+ return followers.stream()
+ .map(FollowerInfo::getLastRespondedAppendEntriesSendTime)
+ .sorted()
+ .limit(mid+1)
+ .skip(mid)
+ .iterator()
+ .next();
+ }
+}
diff --git
a/ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderStateImpl.java
b/ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderStateImpl.java
index 5156585f8..4ebfc3d56 100644
---
a/ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderStateImpl.java
+++
b/ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderStateImpl.java
@@ -348,6 +348,7 @@ class LeaderStateImpl implements LeaderState {
private final PendingStepDown pendingStepDown;
private final ReadIndexHeartbeats readIndexHeartbeats;
+ private final LeaderLease lease;
LeaderStateImpl(RaftServerImpl server) {
this.name = server.getMemberId() + "-" +
JavaUtils.getClassSimpleName(getClass());
@@ -369,6 +370,7 @@ class LeaderStateImpl implements LeaderState {
this.messageStreamRequests = new
MessageStreamRequests(server.getMemberId());
this.pendingStepDown = new PendingStepDown(this);
this.readIndexHeartbeats = new ReadIndexHeartbeats();
+ this.lease = new LeaderLease(properties);
long maxPendingRequests =
RaftServerConfigKeys.Write.elementLimit(properties);
double followerGapRatioMax =
RaftServerConfigKeys.Write.followerGapRatioMax(properties);
@@ -436,6 +438,7 @@ class LeaderStateImpl implements LeaderState {
messageStreamRequests.clear();
// TODO client should retry on NotLeaderException
readIndexHeartbeats.failListeners(nle);
+ lease.getAndSetEnabled(false);
server.getServerRpc().notifyNotLeader(server.getMemberId().getGroupId());
logAppenderMetrics.unregister();
raftServerMetrics.unregister();
@@ -673,6 +676,7 @@ class LeaderStateImpl implements LeaderState {
private void stepDown(long term, StepDownReason reason) {
try {
+ lease.getAndSetEnabled(false);
server.changeToFollowerAndPersistMetadata(term, false, reason);
pendingStepDown.complete(server::newSuccessReply);
} catch(IOException e) {
@@ -951,6 +955,7 @@ class LeaderStateImpl implements LeaderState {
pendingRequests.replySetConfiguration(server::newSuccessReply);
// if the leader is not included in the current configuration, step down
if (!conf.containsInConf(server.getId(), RaftPeerRole.FOLLOWER,
RaftPeerRole.LISTENER)) {
+ lease.getAndSetEnabled(false);
LOG.info("{} is not included in the new configuration {}. Will
shutdown server...", this, conf);
try {
// leave some time for all RPC senders to send out new conf entry
@@ -1111,6 +1116,12 @@ class LeaderStateImpl implements LeaderState {
new LeaderNotReadyException(server.getMemberId())));
}
+ // if lease is enabled, check lease first
+ if (hasLease()) {
+ return CompletableFuture.completedFuture(readIndex);
+ }
+
+ // send heartbeats and wait for majority acknowledgments
final AppendEntriesListener listener =
readIndexHeartbeats.addAppendEntriesListener(
readIndex, i -> new AppendEntriesListener(i, senders));
@@ -1127,6 +1138,32 @@ class LeaderStateImpl implements LeaderState {
readIndexHeartbeats.onAppendEntriesReply(appender, reply,
this::hasMajority);
}
+ boolean getAndSetLeaseEnabled(boolean newValue) {
+ return lease.getAndSetEnabled(newValue);
+ }
+
+ boolean hasLease() {
+ if (!lease.isEnabled()) {
+ return false;
+ }
+
+ if (checkLeaderLease()) {
+ return true;
+ }
+
+ // try extending the leader lease
+ final RaftConfigurationImpl conf = server.getRaftConf();
+ final CurrentOldFollowerInfos info =
followerInfoMap.getFollowerInfos(conf);
+ lease.extend(info.getCurrent(), info.getOld(), peers ->
conf.hasMajority(peers, server.getId()));
+
+ return checkLeaderLease();
+ }
+
+ private boolean checkLeaderLease() {
+ return isRunning() && isReady()
+ && (server.getRaftConf().isSingleton() || lease.isValid());
+ }
+
void replyPendingRequest(long logIndex, RaftClientReply reply) {
pendingRequests.replyPendingRequest(logIndex, reply);
}
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 40a17c4e9..3fb0cb2fa 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
@@ -999,8 +999,14 @@ class RaftServerImpl implements RaftServer.Division,
}
private CompletableFuture<RaftClientReply> readAsync(RaftClientRequest
request) {
- if (readOption == RaftServerConfigKeys.Read.Option.LINEARIZABLE
- && !request.getType().getRead().getPreferNonLinearizable()) {
+ if (request.getType().getRead().getPreferNonLinearizable()
+ || readOption == RaftServerConfigKeys.Read.Option.DEFAULT) {
+ final CompletableFuture<RaftClientReply> reply =
checkLeaderState(request, null, false);
+ if (reply != null) {
+ return reply;
+ }
+ return queryStateMachine(request);
+ } else if (readOption == RaftServerConfigKeys.Read.Option.LINEARIZABLE){
/*
Linearizable read using ReadIndex. See Raft paper section 6.4.
1. First obtain readIndex from Leader.
@@ -1027,13 +1033,6 @@ class RaftServerImpl implements RaftServer.Division,
.thenCompose(readIndex -> getReadRequests().waitToAdvance(readIndex))
.thenCompose(readIndex -> queryStateMachine(request))
.exceptionally(e -> readException2Reply(request, e));
- } else if (readOption == RaftServerConfigKeys.Read.Option.DEFAULT
- || request.getType().getRead().getPreferNonLinearizable()) {
- CompletableFuture<RaftClientReply> reply = checkLeaderState(request,
null, false);
- if (reply != null) {
- return reply;
- }
- return queryStateMachine(request);
} else {
throw new IllegalStateException("Unexpected read option: " + readOption);
}
diff --git
a/ratis-server/src/main/java/org/apache/ratis/server/impl/TransferLeadership.java
b/ratis-server/src/main/java/org/apache/ratis/server/impl/TransferLeadership.java
index 74ada6541..e54bee748 100644
---
a/ratis-server/src/main/java/org/apache/ratis/server/impl/TransferLeadership.java
+++
b/ratis-server/src/main/java/org/apache/ratis/server/impl/TransferLeadership.java
@@ -295,6 +295,9 @@ public class TransferLeadership {
if (previous != null) {
return createReplyFutureFromPreviousRequest(request, previous);
}
+ // disable the lease before transferring leader
+ final boolean previousLeaseEnabled = server.getRole().getLeaderState()
+ .map(l -> l.getAndSetLeaseEnabled(false)).orElse(false);
final PendingRequest pendingRequest = supplier.get();
final Result result = tryTransferLeadership(context);
final Result.Type type = result.getType();
@@ -308,6 +311,12 @@ public class TransferLeadership {
timeout.toString(TimeUnit.SECONDS, 3))),
LOG, () -> "Failed to handle timeout");
}
+ // reset back lease if the current transfer fails
+ pendingRequest.getReplyFuture().whenCompleteAsync((reply, ex) -> {
+ if (ex != null || !reply.isSuccess()) {
+ server.getRole().getLeaderState().ifPresent(l ->
l.getAndSetLeaseEnabled(previousLeaseEnabled));
+ }
+ });
return pendingRequest.getReplyFuture();
}
diff --git
a/ratis-server/src/main/java/org/apache/ratis/server/leader/LogAppenderDefault.java
b/ratis-server/src/main/java/org/apache/ratis/server/leader/LogAppenderDefault.java
index 8f71f91fc..6f38f5009 100644
---
a/ratis-server/src/main/java/org/apache/ratis/server/leader/LogAppenderDefault.java
+++
b/ratis-server/src/main/java/org/apache/ratis/server/leader/LogAppenderDefault.java
@@ -26,6 +26,7 @@ import org.apache.ratis.server.RaftServer;
import org.apache.ratis.server.raftlog.RaftLogIOException;
import org.apache.ratis.server.util.ServerStringUtils;
import org.apache.ratis.statemachine.SnapshotInfo;
+import org.apache.ratis.util.Timestamp;
import java.io.IOException;
import java.io.InterruptedIOException;
@@ -73,9 +74,11 @@ class LogAppenderDefault extends LogAppenderBase {
}
resetHeartbeatTrigger();
+ final Timestamp sendTime = Timestamp.currentTime();
getFollower().updateLastRpcSendTime(request.getEntriesCount() == 0);
final AppendEntriesReplyProto r =
getServerRpc().appendEntries(request);
getFollower().updateLastRpcResponseTime();
+ getFollower().updateLastRespondedAppendEntriesSendTime(sendTime);
getLeaderState().onFollowerCommitIndex(getFollower(),
r.getFollowerCommit());
return r;
diff --git
a/ratis-server/src/test/java/org/apache/ratis/ReadOnlyRequestTests.java
b/ratis-server/src/test/java/org/apache/ratis/ReadOnlyRequestTests.java
index a919a9292..eea75592e 100644
--- a/ratis-server/src/test/java/org/apache/ratis/ReadOnlyRequestTests.java
+++ b/ratis-server/src/test/java/org/apache/ratis/ReadOnlyRequestTests.java
@@ -43,7 +43,6 @@ import org.slf4j.event.Level;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
@@ -69,16 +68,21 @@ public abstract class ReadOnlyRequestTests<CLUSTER extends
MiniRaftCluster>
final RaftProperties p = getProperties();
p.setClass(MiniRaftCluster.STATEMACHINE_CLASS_KEY,
CounterStateMachine.class, StateMachine.class);
-
- p.setEnum(RaftServerConfigKeys.Read.OPTION_KEY,
RaftServerConfigKeys.Read.Option.LINEARIZABLE);
}
@Test
public void testLinearizableRead() throws Exception {
- runWithNewCluster(NUM_SERVERS, this::testLinearizableReadImpl);
+ getProperties().setEnum(RaftServerConfigKeys.Read.OPTION_KEY,
RaftServerConfigKeys.Read.Option.LINEARIZABLE);
+ runWithNewCluster(NUM_SERVERS, this::testReadOnlyImpl);
}
- private void testLinearizableReadImpl(CLUSTER cluster) throws Exception {
+ @Test
+ public void testLeaseRead() throws Exception {
+
getProperties().setBoolean(RaftServerConfigKeys.Read.LEADER_LEASE_ENABLED_KEY,
true);
+ runWithNewCluster(NUM_SERVERS, this::testReadOnlyImpl);
+ }
+
+ private void testReadOnlyImpl(CLUSTER cluster) throws Exception {
try {
RaftTestUtil.waitForLeader(cluster);
final RaftPeerId leaderId = cluster.getLeader().getId();
@@ -98,10 +102,17 @@ public abstract class ReadOnlyRequestTests<CLUSTER extends
MiniRaftCluster>
@Test
public void testLinearizableReadTimeout() throws Exception {
- runWithNewCluster(NUM_SERVERS, this::testLinearizableReadTimeoutImpl);
+ getProperties().setEnum(RaftServerConfigKeys.Read.OPTION_KEY,
RaftServerConfigKeys.Read.Option.LINEARIZABLE);
+ runWithNewCluster(NUM_SERVERS, this::testReadOnlyTimeoutImpl);
+ }
+
+ @Test
+ public void testLeaseReadTimeout() throws Exception {
+
getProperties().setBoolean(RaftServerConfigKeys.Read.LEADER_LEASE_ENABLED_KEY,
true);
+ runWithNewCluster(NUM_SERVERS, this::testReadOnlyTimeoutImpl);
}
- private void testLinearizableReadTimeoutImpl(CLUSTER cluster) throws
Exception {
+ private void testReadOnlyTimeoutImpl(CLUSTER cluster) throws Exception {
try {
RaftTestUtil.waitForLeader(cluster);
final RaftPeerId leaderId = cluster.getLeader().getId();
@@ -126,10 +137,17 @@ public abstract class ReadOnlyRequestTests<CLUSTER
extends MiniRaftCluster>
@Test
public void testFollowerLinearizableRead() throws Exception {
- runWithNewCluster(NUM_SERVERS, this::testFollowerLinearizableReadImpl);
+ getProperties().setEnum(RaftServerConfigKeys.Read.OPTION_KEY,
RaftServerConfigKeys.Read.Option.LINEARIZABLE);
+ runWithNewCluster(NUM_SERVERS, this::testFollowerReadOnlyImpl);
+ }
+
+ @Test
+ public void testFollowerLeaseRead() throws Exception {
+
getProperties().setBoolean(RaftServerConfigKeys.Read.LEADER_LEASE_ENABLED_KEY,
true);
+ runWithNewCluster(NUM_SERVERS, this::testFollowerReadOnlyImpl);
}
- private void testFollowerLinearizableReadImpl(CLUSTER cluster) throws
Exception {
+ private void testFollowerReadOnlyImpl(CLUSTER cluster) throws Exception {
try {
RaftTestUtil.waitForLeader(cluster);
@@ -155,10 +173,17 @@ public abstract class ReadOnlyRequestTests<CLUSTER
extends MiniRaftCluster>
@Test
public void testFollowerLinearizableReadParallel() throws Exception {
- runWithNewCluster(NUM_SERVERS,
this::testFollowerLinearizableReadParallelImpl);
+ getProperties().setEnum(RaftServerConfigKeys.Read.OPTION_KEY,
RaftServerConfigKeys.Read.Option.LINEARIZABLE);
+ runWithNewCluster(NUM_SERVERS, this::testFollowerReadOnlyParallelImpl);
}
- private void testFollowerLinearizableReadParallelImpl(CLUSTER cluster)
throws Exception {
+ @Test
+ public void testFollowerLeaseReadParallel() throws Exception {
+
getProperties().setBoolean(RaftServerConfigKeys.Read.LEADER_LEASE_ENABLED_KEY,
true);
+ runWithNewCluster(NUM_SERVERS, this::testFollowerReadOnlyParallelImpl);
+ }
+
+ private void testFollowerReadOnlyParallelImpl(CLUSTER cluster) throws
Exception {
try {
RaftTestUtil.waitForLeader(cluster);
@@ -183,10 +208,17 @@ public abstract class ReadOnlyRequestTests<CLUSTER
extends MiniRaftCluster>
@Test
public void testFollowerLinearizableReadFailWhenLeaderDown() throws
Exception {
- runWithNewCluster(NUM_SERVERS,
this::testFollowerLinearizableReadFailWhenLeaderDownImpl);
+ getProperties().setEnum(RaftServerConfigKeys.Read.OPTION_KEY,
RaftServerConfigKeys.Read.Option.LINEARIZABLE);
+ runWithNewCluster(NUM_SERVERS,
this::testFollowerReadOnlyFailWhenLeaderDownImpl);
}
- private void testFollowerLinearizableReadFailWhenLeaderDownImpl(CLUSTER
cluster) throws Exception {
+ @Test
+ public void testFollowerLeaseReadWhenLeaderDown() throws Exception {
+
getProperties().setBoolean(RaftServerConfigKeys.Read.LEADER_LEASE_ENABLED_KEY,
true);
+ runWithNewCluster(NUM_SERVERS,
this::testFollowerReadOnlyFailWhenLeaderDownImpl);
+ }
+
+ private void testFollowerReadOnlyFailWhenLeaderDownImpl(CLUSTER cluster)
throws Exception {
try {
RaftTestUtil.waitForLeader(cluster);
@@ -215,11 +247,18 @@ public abstract class ReadOnlyRequestTests<CLUSTER
extends MiniRaftCluster>
}
@Test
- public void testFollowerLinearizableReadRetryWhenLeaderDown() throws
Exception {
- runWithNewCluster(NUM_SERVERS,
this::testFollowerLinearizableReadRetryWhenLeaderDown);
+ public void testFollowerReadOnlyRetryWhenLeaderDown() throws Exception {
+ getProperties().setEnum(RaftServerConfigKeys.Read.OPTION_KEY,
RaftServerConfigKeys.Read.Option.LINEARIZABLE);
+ runWithNewCluster(NUM_SERVERS,
this::testFollowerReadOnlyRetryWhenLeaderDown);
+ }
+
+ @Test
+ public void testFollowerLeaseReadRetryWhenLeaderDown() throws Exception {
+
getProperties().setBoolean(RaftServerConfigKeys.Read.LEADER_LEASE_ENABLED_KEY,
true);
+ runWithNewCluster(NUM_SERVERS,
this::testFollowerReadOnlyRetryWhenLeaderDown);
}
- private void testFollowerLinearizableReadRetryWhenLeaderDown(CLUSTER
cluster) throws Exception {
+ private void testFollowerReadOnlyRetryWhenLeaderDown(CLUSTER cluster) throws
Exception {
// only retry on readIndexException
final RetryPolicy retryPolicy = ExceptionDependentRetry
.newBuilder()
diff --git
a/ratis-server/src/test/java/org/apache/ratis/server/impl/LeaderElectionTests.java
b/ratis-server/src/test/java/org/apache/ratis/server/impl/LeaderElectionTests.java
index 9e2b7bd2d..6453e8e94 100644
---
a/ratis-server/src/test/java/org/apache/ratis/server/impl/LeaderElectionTests.java
+++
b/ratis-server/src/test/java/org/apache/ratis/server/impl/LeaderElectionTests.java
@@ -43,6 +43,7 @@ import org.apache.ratis.util.LifeCycle;
import org.apache.ratis.util.Slf4jUtils;
import org.apache.ratis.util.TimeDuration;
import org.apache.ratis.util.Timestamp;
+import org.apache.ratis.util.function.CheckedBiConsumer;
import org.junit.Assert;
import org.junit.Test;
import org.slf4j.Logger;
@@ -640,6 +641,89 @@ public abstract class LeaderElectionTests<CLUSTER extends
MiniRaftCluster>
}
}
+ private void runLeaseTest(CLUSTER cluster, CheckedBiConsumer<CLUSTER, Long,
Exception> testCase) throws Exception {
+ final double leaseRatio =
RaftServerConfigKeys.Read.leaderLeaseTimeoutRatio(getProperties());
+ final long leaseTimeoutMs =
RaftServerConfigKeys.Rpc.timeoutMin(getProperties())
+ .multiply(leaseRatio)
+ .toIntExact(TimeUnit.MILLISECONDS);
+ testCase.accept(cluster, leaseTimeoutMs);
+ }
+
+ @Test
+ public void testLeaderLease() throws Exception {
+ // use a strict lease
+ RaftServerConfigKeys.Read.setLeaderLeaseEnabled(getProperties(), true);
+ RaftServerConfigKeys.Read.setLeaderLeaseTimeoutRatio(getProperties(), 0.5);
+ runWithNewCluster(3, c -> runLeaseTest(c, this::runTestLeaderLease));
+ }
+
+ void runTestLeaderLease(CLUSTER cluster, long leaseTimeoutMs) throws
Exception {
+ final RaftServer.Division leader = RaftTestUtil.waitForLeader(cluster);
+ try (final RaftClient client = cluster.createClient(leader.getId())) {
+ client.io().send(new RaftTestUtil.SimpleMessage("message"));
+
+ Assert.assertTrue(leader.getInfo().isLeader());
+ Assert.assertTrue(leader.getInfo().isLeaderReady());
+ RaftServerTestUtil.assertLeaderLease(leader, true);
+
+ isolate(cluster, leader.getId());
+ Thread.sleep(leaseTimeoutMs);
+
+ Assert.assertTrue(leader.getInfo().isLeader());
+ Assert.assertTrue(leader.getInfo().isLeaderReady());
+ RaftServerTestUtil.assertLeaderLease(leader, false);
+ } finally {
+ deIsolate(cluster, leader.getId());
+ }
+ }
+
+ @Test
+ public void testLeaderLeaseDuringReconfiguration() throws Exception {
+ // use a strict lease
+ RaftServerConfigKeys.Read.setLeaderLeaseEnabled(getProperties(), true);
+ RaftServerConfigKeys.Read.setLeaderLeaseTimeoutRatio(getProperties(), 0.5);
+ runWithNewCluster(3, c -> runLeaseTest(c,
this::runTestLeaderLeaseDuringReconfiguration));
+ }
+
+ void runTestLeaderLeaseDuringReconfiguration(CLUSTER cluster, long
leaseTimeoutMs) throws Exception {
+ final RaftServer.Division leader = RaftTestUtil.waitForLeader(cluster);
+ try (final RaftClient client = cluster.createClient(leader.getId())) {
+ client.io().send(new RaftTestUtil.SimpleMessage("message"));
+
+ Assert.assertTrue(leader.getInfo().isLeader());
+ Assert.assertTrue(leader.getInfo().isLeaderReady());
+ RaftServerTestUtil.assertLeaderLease(leader, true);
+
+ final List<RaftServer.Division> followers = cluster.getFollowers();
+ final MiniRaftCluster.PeerChanges changes = cluster.addNewPeers(2, true);
+
+ // blocking the original 2 followers
+
BlockRequestHandlingInjection.getInstance().blockReplier(followers.get(0).getId().toString());
+
BlockRequestHandlingInjection.getInstance().blockReplier(followers.get(1).getId().toString());
+
+ // start reconfiguration in another thread, shall fail eventually
+ new Thread(() -> {
+ try {
+ client.admin().setConfiguration(changes.allPeersInNewConf);
+ } catch (IOException e) {
+ System.out.println("as expected: " + e.getMessage());
+ }
+ }).start();
+
+ Thread.sleep(leaseTimeoutMs);
+
+ Assert.assertTrue(leader.getInfo().isLeader());
+ Assert.assertTrue(leader.getInfo().isLeaderReady());
+ RaftServerTestUtil.assertLeaderLease(leader, false);
+
+ } finally {
+ BlockRequestHandlingInjection.getInstance().unblockAll();
+ }
+ }
+
+
+
+
private static RaftServerImpl createMockServer(boolean alive) {
final DivisionInfo info = mock(DivisionInfo.class);
when(info.isAlive()).thenReturn(alive);
diff --git
a/ratis-server/src/test/java/org/apache/ratis/server/impl/RaftServerTestUtil.java
b/ratis-server/src/test/java/org/apache/ratis/server/impl/RaftServerTestUtil.java
index 618e398b3..958c19442 100644
---
a/ratis-server/src/test/java/org/apache/ratis/server/impl/RaftServerTestUtil.java
+++
b/ratis-server/src/test/java/org/apache/ratis/server/impl/RaftServerTestUtil.java
@@ -147,6 +147,12 @@ public class RaftServerTestUtil {
return
getLeaderState(server).map(LeaderStateImpl::getLogAppenders).orElse(null);
}
+ public static void assertLeaderLease(RaftServer.Division leader, boolean
hasLease) {
+ final LeaderStateImpl l = getLeaderState(leader).orElse(null);
+ Assert.assertNotNull(l);
+ Assert.assertEquals(l.hasLease(), hasLease);
+ }
+
public static void restartLogAppenders(RaftServer.Division server) {
final LeaderStateImpl leaderState = getLeaderState(server).orElseThrow(
() -> new IllegalStateException(server + " is not the leader"));
