This is an automated email from the ASF dual-hosted git repository.
szetszwo 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 0355d33e0 RATIS-2511. Follower should throw ReadException if it is
installing snapshot (#1444)
0355d33e0 is described below
commit 0355d33e06c98d8410260bb2c36feba0661af16f
Author: Ivan Andika <[email protected]>
AuthorDate: Wed May 20 00:55:56 2026 +0800
RATIS-2511. Follower should throw ReadException if it is installing
snapshot (#1444)
---
.../apache/ratis/server/impl/RaftServerImpl.java | 24 +++----
.../org/apache/ratis/server/impl/ReadRequests.java | 26 ++++++--
.../server/impl/SnapshotInstallationHandler.java | 3 +-
.../org/apache/ratis/LinearizableReadTests.java | 7 +-
.../org/apache/ratis/ReadOnlyRequestTests.java | 74 ++++++++++++++++++++++
5 files changed, 113 insertions(+), 21 deletions(-)
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 a35d7e0f2..f758fd0ed 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
@@ -1086,12 +1086,15 @@ class RaftServerImpl implements RaftServer.Division,
}
return processQueryFuture(stateMachine.queryStale(request.getMessage(),
minIndex), request);
}
-
- ReadRequests getReadRequests() {
- return getState().getReadRequests();
+ ReadException getReadException(String op, long installSnapshot, boolean
started) {
+ return installSnapshot == RaftLog.INVALID_LOG_INDEX ? null : new
ReadException(getMemberId() + ": Failed to " + op
+ + " readIndex as snapshot (" + installSnapshot + ") installation is "
+ (started ? "started" : "in progress"));
}
-
private CompletableFuture<ReadIndexReplyProto>
sendReadIndexAsync(RaftClientRequest clientRequest) {
+ final long installSnapshot =
snapshotInstallationHandler.getInProgressInstallSnapshotIndex();
+ if (installSnapshot != RaftLog.INVALID_LOG_INDEX) {
+ return JavaUtils.completeExceptionally(getReadException("get",
installSnapshot, false));
+ }
final RaftPeerId leaderId = getInfo().getLeaderId();
if (leaderId == null) {
return JavaUtils.completeExceptionally(new
ReadIndexException(getMemberId() + ": Leader is unknown."));
@@ -1103,11 +1106,9 @@ class RaftServerImpl implements RaftServer.Division,
return JavaUtils.completeExceptionally(e);
}
}
-
private CompletableFuture<Long> getReadIndex(RaftClientRequest request,
LeaderStateImpl leader) {
return
writeIndexCache.getWriteIndexFuture(request).thenCompose(leader::getReadIndex);
}
-
private CompletableFuture<RaftClientReply> readAsync(RaftClientRequest
request) {
if (request.getType().getRead().getPreferNonLinearizable()
|| readOption == RaftServerConfigKeys.Read.Option.DEFAULT) {
@@ -1117,14 +1118,7 @@ class RaftServerImpl implements RaftServer.Division,
}
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.
- 2. Then waits for statemachine to advance at least as far as readIndex.
- 3. Finally, query the statemachine and return the result.
- */
final LeaderStateImpl leader = role.getLeaderState().orElse(null);
-
final CompletableFuture<Long> replyFuture;
if (leader != null) {
replyFuture = getReadIndex(request, leader);
@@ -1140,14 +1134,14 @@ class RaftServerImpl implements RaftServer.Division,
}
return replyFuture
- .thenCompose(readIndex -> getReadRequests().waitToAdvance(readIndex))
+ .thenCompose(readIndex ->
getState().getReadRequests().waitToAdvance(readIndex,
+ () -> getReadException("add",
snapshotInstallationHandler.getInProgressInstallSnapshotIndex(), false)))
.thenCompose(readIndex -> queryStateMachine(request))
.exceptionally(e -> readException2Reply(request, e));
} else {
throw new IllegalStateException("Unexpected read option: " + readOption);
}
}
-
private RaftClientReply readException2Reply(RaftClientRequest request,
Throwable e) {
e = JavaUtils.unwrapCompletionException(e);
if (e instanceof StateMachineException ) {
diff --git
a/ratis-server/src/main/java/org/apache/ratis/server/impl/ReadRequests.java
b/ratis-server/src/main/java/org/apache/ratis/server/impl/ReadRequests.java
index 6112a4600..df6f3d93f 100644
--- a/ratis-server/src/main/java/org/apache/ratis/server/impl/ReadRequests.java
+++ b/ratis-server/src/main/java/org/apache/ratis/server/impl/ReadRequests.java
@@ -20,16 +20,19 @@ package org.apache.ratis.server.impl;
import org.apache.ratis.conf.RaftProperties;
import org.apache.ratis.protocol.exceptions.ReadException;
import org.apache.ratis.server.RaftServerConfigKeys;
+import org.apache.ratis.util.JavaUtils;
import org.apache.ratis.util.Preconditions;
import org.apache.ratis.util.TimeDuration;
import org.apache.ratis.util.TimeoutExecutor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.Collection;
import java.util.NavigableMap;
import java.util.TreeMap;
import java.util.concurrent.CompletableFuture;
import java.util.function.LongConsumer;
+import java.util.function.Supplier;
/** For supporting linearizable read. */
class ReadRequests {
@@ -43,7 +46,7 @@ class ReadRequests {
* Map : readIndex -> appliedIndexFuture (when completes, readIndex
<= appliedIndex).
* Invariant: all keys > lastAppliedIndex.
*/
- private final NavigableMap<Long, CompletableFuture<Long>> sorted = new
TreeMap<>();
+ private NavigableMap<Long, CompletableFuture<Long>> sorted = new
TreeMap<>();
private final TimeDuration readTimeout;
@@ -52,10 +55,14 @@ class ReadRequests {
this.readTimeout = readTimeout;
}
- CompletableFuture<Long> add(long readIndex) {
+ CompletableFuture<Long> add(long readIndex, Supplier<ReadException>
checkReadException) {
final CompletableFuture<Long> returned;
final boolean create;
synchronized (this) {
+ final ReadException exception = checkReadException.get();
+ if (exception != null) {
+ return JavaUtils.completeExceptionally(exception);
+ }
if (readIndex <= lastAppliedIndex) {
return CompletableFuture.completedFuture(lastAppliedIndex);
}
@@ -88,6 +95,11 @@ class ReadRequests {
removed.completeExceptionally(new ReadException("Read timeout " +
readTimeout + " for index " + readIndex));
}
+ synchronized Collection<CompletableFuture<Long>> clear() {
+ final Collection<CompletableFuture<Long>> futures = sorted.values();
+ sorted = new TreeMap<>();
+ return futures;
+ }
/** Complete all the entries less than or equal to the given applied
index. */
synchronized void complete(long appliedIndex) {
@@ -119,7 +131,13 @@ class ReadRequests {
return readIndexQueue::complete;
}
- CompletableFuture<Long> waitToAdvance(long readIndex) {
- return readIndexQueue.add(readIndex);
+ CompletableFuture<Long> waitToAdvance(long readIndex,
Supplier<ReadException> checkReadException) {
+ return readIndexQueue.add(readIndex, checkReadException);
+ }
+
+ void fail(Throwable cause) {
+ for (CompletableFuture<Long> f : readIndexQueue.clear()) {
+ f.completeExceptionally(cause);
+ }
}
}
diff --git
a/ratis-server/src/main/java/org/apache/ratis/server/impl/SnapshotInstallationHandler.java
b/ratis-server/src/main/java/org/apache/ratis/server/impl/SnapshotInstallationHandler.java
index 46b6aaf87..870625662 100644
---
a/ratis-server/src/main/java/org/apache/ratis/server/impl/SnapshotInstallationHandler.java
+++
b/ratis-server/src/main/java/org/apache/ratis/server/impl/SnapshotInstallationHandler.java
@@ -276,6 +276,7 @@ class SnapshotInstallationHandler {
InstallSnapshotResult.ALREADY_INSTALLED, snapshotIndex);
return future.thenApply(dummy -> reply);
}
+ server.getState().getReadRequests().fail(server.getReadException("wait
for", firstAvailableLogIndex, true));
final RaftPeerProto leaderProto;
if (!request.hasLastRaftConfigurationLogEntryProto()) {
@@ -401,4 +402,4 @@ class SnapshotInstallationHandler {
.setFollowerInfo(followerInfo)
.build();
}
-}
\ No newline at end of file
+}
diff --git
a/ratis-server/src/test/java/org/apache/ratis/LinearizableReadTests.java
b/ratis-server/src/test/java/org/apache/ratis/LinearizableReadTests.java
index 09781b546..832457e9d 100644
--- a/ratis-server/src/test/java/org/apache/ratis/LinearizableReadTests.java
+++ b/ratis-server/src/test/java/org/apache/ratis/LinearizableReadTests.java
@@ -169,6 +169,11 @@ public abstract class LinearizableReadTests<CLUSTER
extends MiniRaftCluster>
}
}
+ @Test
+ public void testFollowerLinearizableReadFailsWhenInstallingSnapshot() throws
Exception {
+
runWithNewCluster(ReadOnlyRequestTests::runTestFollowerLinearizableReadFailsWhenInstallingSnapshot);
+ }
+
@Test
public void testFollowerLinearizableReadParallel() throws Exception {
runWithNewCluster(LinearizableReadTests::runTestFollowerReadOnlyParallel);
@@ -285,4 +290,4 @@ public abstract class LinearizableReadTests<CLUSTER extends
MiniRaftCluster>
assertReplyAtLeast(2, asyncReply.join());
}
}
-}
\ No newline at end of file
+}
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 94e9433b1..d88780764 100644
--- a/ratis-server/src/test/java/org/apache/ratis/ReadOnlyRequestTests.java
+++ b/ratis-server/src/test/java/org/apache/ratis/ReadOnlyRequestTests.java
@@ -40,8 +40,12 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.slf4j.event.Level;
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
import java.nio.charset.StandardCharsets;
+import java.util.List;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
import java.util.concurrent.atomic.AtomicLong;
public abstract class ReadOnlyRequestTests<CLUSTER extends MiniRaftCluster>
@@ -139,6 +143,76 @@ public abstract class ReadOnlyRequestTests<CLUSTER extends
MiniRaftCluster>
}
}
+ private static void setInProgressInstallSnapshotIndex(RaftServer.Division
server, long index) throws Exception {
+ final Field snapshotInstallationHandler =
server.getClass().getDeclaredField("snapshotInstallationHandler");
+ snapshotInstallationHandler.setAccessible(true);
+ final Object handler = snapshotInstallationHandler.get(server);
+ final Field inProgressInstallSnapshotIndex = handler.getClass()
+ .getDeclaredField("inProgressInstallSnapshotIndex");
+ inProgressInstallSnapshotIndex.setAccessible(true);
+ ((AtomicLong) inProgressInstallSnapshotIndex.get(handler)).set(index);
+ }
+
+ private static void startSnapshotInstallation(RaftServer.Division server,
long index) throws Exception {
+ setInProgressInstallSnapshotIndex(server, index);
+ final Method getState = server.getClass().getDeclaredMethod("getState");
+ getState.setAccessible(true);
+ final Object state = getState.invoke(server);
+ final Method getReadRequests =
state.getClass().getDeclaredMethod("getReadRequests");
+ getReadRequests.setAccessible(true);
+ final Object readRequests = getReadRequests.invoke(state);
+ final Method fail = readRequests.getClass().getDeclaredMethod("fail",
Throwable.class);
+ fail.setAccessible(true);
+ fail.invoke(readRequests, new ReadException(server.getMemberId()
+ + ": Failed to wait for readIndex as snapshot (" + index + ")
installation is started"));
+ }
+
+ static void assertSnapshotInstallationReadException(Throwable exception) {
+ final Throwable cause = exception instanceof CompletionException &&
exception.getCause() != null
+ ? exception.getCause() : exception;
+ Assertions.assertInstanceOf(ReadException.class, cause);
+ Assertions.assertTrue(cause.getMessage().contains("snapshot (1)
installation is"),
+ () -> "Unexpected exception: " + exception);
+ }
+
+ static <C extends MiniRaftCluster> void
runTestFollowerLinearizableReadFailsWhenInstallingSnapshot(C cluster)
+ throws Exception {
+ final RaftPeerId leaderId = RaftTestUtil.waitForLeader(cluster).getId();
+
+ final List<RaftServer.Division> followers = cluster.getFollowers();
+ Assertions.assertEquals(2, followers.size());
+
+ final RaftServer.Division follower = followers.get(0);
+ final RaftPeerId followerId = follower.getId();
+
+ try (RaftClient leaderClient = cluster.createClient(leaderId);
+ RaftClient followerClient = cluster.createClient(followerId,
RetryPolicies.noRetry())) {
+ assertReplyExact(1, leaderClient.io().send(INCREMENT));
+ assertReplyExact(1, followerClient.io().sendReadOnly(QUERY, followerId));
+
+ final CompletableFuture<RaftClientReply> writeReply =
leaderClient.async().send(WAIT_AND_INCREMENT);
+ Thread.sleep(100);
+ final CompletableFuture<RaftClientReply> pendingRead =
followerClient.async().sendReadOnly(QUERY, followerId);
+ Assertions.assertFalse(pendingRead.isDone(), () -> "pendingRead=" +
pendingRead);
+
+ startSnapshotInstallation(follower, 1);
+ try {
+ final CompletionException pendingException =
Assertions.assertThrows(CompletionException.class,
+ pendingRead::join);
+ assertSnapshotInstallationReadException(pendingException);
+
+ final ReadException readException =
Assertions.assertThrows(ReadException.class,
+ () -> followerClient.io().sendReadOnly(QUERY, followerId));
+ assertSnapshotInstallationReadException(readException);
+ } finally {
+ setInProgressInstallSnapshotIndex(follower, -1);
+ }
+
+ assertReplyExact(2, writeReply.join());
+ assertReplyExact(2, followerClient.io().sendReadOnly(QUERY, followerId));
+ }
+ }
+
static int retrieve(RaftClientReply reply) {
Assertions.assertTrue(reply.isSuccess());
return
Integer.parseInt(reply.getMessage().getContent().toString(StandardCharsets.UTF_8));
