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 11e9e1f9d RATIS-2559. Add linearizable check for streaming read
request (#1490)
11e9e1f9d is described below
commit 11e9e1f9d576f5abcc4d7972b860209a6bf9662a
Author: Peter Lee <[email protected]>
AuthorDate: Mon Jul 6 17:23:20 2026 +0800
RATIS-2559. Add linearizable check for streaming read request (#1490)
---
dev-support/checkstyle.xml | 4 +
.../apache/ratis/protocol/RaftClientRequest.java | 68 +++++++++++-
.../ratis/netty/server/ReadStreamManagement.java | 54 +++++++--
ratis-proto/src/main/proto/Raft.proto | 1 +
.../apache/ratis/server/impl/RaftServerImpl.java | 31 +++++-
.../client/impl/TestDataStreamClientImpl.java | 17 ---
.../ratis/datastream/DataStreamClusterTests.java | 3 -
.../netty/server/TestDataStreamManagement.java | 122 ++++++++++++++++++++-
8 files changed, 256 insertions(+), 44 deletions(-)
diff --git a/dev-support/checkstyle.xml b/dev-support/checkstyle.xml
index db4954fb4..16dd31ad7 100644
--- a/dev-support/checkstyle.xml
+++ b/dev-support/checkstyle.xml
@@ -60,6 +60,10 @@
</module>
<module name="SuppressWarningsFilter"/>
+ <module name="SuppressionSingleFilter">
+ <property name="checks" value="FileLength"/>
+ <property name="files"
value="[/\\]ratis-server[/\\]src[/\\]main[/\\]java[/\\]org[/\\]apache[/\\]ratis[/\\]server[/\\]impl[/\\]RaftServerImpl\.java"/>
+ </module>
<!-- Checks that a package-info.java file exists for each package. -->
<!-- See http://checkstyle.sf.net/config_javadoc.html#JavadocPackage -->
diff --git
a/ratis-common/src/main/java/org/apache/ratis/protocol/RaftClientRequest.java
b/ratis-common/src/main/java/org/apache/ratis/protocol/RaftClientRequest.java
index b04402fe1..ca7b6cc88 100644
---
a/ratis-common/src/main/java/org/apache/ratis/protocol/RaftClientRequest.java
+++
b/ratis-common/src/main/java/org/apache/ratis/protocol/RaftClientRequest.java
@@ -46,11 +46,11 @@ public class RaftClientRequest extends RaftClientMessage {
private static final Type WATCH_DEFAULT = new Type(
WatchRequestTypeProto.newBuilder().setIndex(0L).setReplication(ReplicationLevel.MAJORITY).build());
- private static final Type READ_AFTER_WRITE_CONSISTENT_DEFAULT
- = new
Type(ReadRequestTypeProto.newBuilder().setReadAfterWriteConsistent(true).build());
- private static final Type READ_DEFAULT = new
Type(ReadRequestTypeProto.getDefaultInstance());
- private static final Type READ_NONLINEARIZABLE_DEFAULT
- = new
Type(ReadRequestTypeProto.newBuilder().setPreferNonLinearizable(true).build());
+ private static final ReadTypes READ_TYPES = new ReadTypes();
+ private static final Type READ_DEFAULT = readRequestType(false, false,
false);
+ private static final Type READ_NONLINEARIZABLE_DEFAULT =
readRequestType(true, false, false);
+ private static final Type READ_AFTER_WRITE_CONSISTENT_DEFAULT =
readRequestType(false, true, false);
+
private static final Type STALE_READ_DEFAULT = new
Type(StaleReadRequestTypeProto.getDefaultInstance());
private static final Map<ReplicationLevel, Type> WRITE_REQUEST_TYPES;
@@ -71,6 +71,44 @@ public class RaftClientRequest extends RaftClientMessage {
return WRITE_REQUEST_TYPES.get(replication);
}
+ private static final class ReadTypes {
+ private final Type[] array = new Type[8];
+
+ private ReadTypes() {
+ for (int i = 0; i < array.length; i++) {
+ array[i] = new Type(ReadRequestTypeProto.newBuilder()
+ .setPreferNonLinearizable((i & 1) != 0)
+ .setReadAfterWriteConsistent((i & 2) != 0)
+ .setDummy((i & 4) != 0)
+ .build());
+ }
+
+ assertArray();
+ }
+
+ private Type getImpl(boolean nonLinearizable, boolean
readAfterWriteConsistent, boolean dummy) {
+ final int i = (nonLinearizable ? 1 : 0)
+ | (readAfterWriteConsistent ? 2 : 0)
+ | (dummy ? 4 : 0);
+ return array[i];
+ }
+
+ Type get(boolean nonLinearizable, boolean readAfterWriteConsistent,
boolean dummy) {
+ if (nonLinearizable && readAfterWriteConsistent) {
+ throw new IllegalArgumentException("Cannot be both nonLinearizable and
readAfterWriteConsistent");
+ }
+ return getImpl(nonLinearizable, readAfterWriteConsistent, dummy);
+ }
+
+ private void assertArray() {
+ for (final Type type : array) {
+ final ReadRequestTypeProto read = type.getRead();
+ final Type got = getImpl(read.getPreferNonLinearizable(),
read.getReadAfterWriteConsistent(), read.getDummy());
+ Preconditions.assertSame(type, got, "type");
+ }
+ }
+ }
+
public static Type writeRequestType() {
return writeRequestType(ReplicationLevel.MAJORITY);
}
@@ -91,6 +129,10 @@ public class RaftClientRequest extends RaftClientMessage {
.build());
}
+ public static Type readRequestType(boolean nonLinearizable, boolean
readAfterWriteConsistent, boolean dummy) {
+ return READ_TYPES.get(nonLinearizable, readAfterWriteConsistent, dummy);
+ }
+
public static Type readAfterWriteConsistentRequestType() {
return READ_AFTER_WRITE_CONSISTENT_DEFAULT;
}
@@ -312,6 +354,22 @@ public class RaftClientRequest extends RaftClientMessage {
return new RaftClientRequest(this);
}
+ public Builder set(RaftClientRequest request) {
+ this.clientId = request.getClientId();
+ this.serverId = request.getServerId();
+ this.groupId = request.getRaftGroupId();
+ this.callId = request.getCallId();
+ this.toLeader = request.isToLeader();
+ this.repliedCallIds = request.getRepliedCallIds();
+ this.message = request.getMessage();
+ this.type = request.getType();
+ this.slidingWindowEntry = request.getSlidingWindowEntry();
+ this.routingTable = request.getRoutingTable();
+ this.timeoutMs = request.getTimeoutMs();
+ this.spanContext = request.getSpanContext();
+ return this;
+ }
+
public Builder setClientId(ClientId clientId) {
this.clientId = clientId;
return this;
diff --git
a/ratis-netty/src/main/java/org/apache/ratis/netty/server/ReadStreamManagement.java
b/ratis-netty/src/main/java/org/apache/ratis/netty/server/ReadStreamManagement.java
index 5336760a0..46d169bbf 100644
---
a/ratis-netty/src/main/java/org/apache/ratis/netty/server/ReadStreamManagement.java
+++
b/ratis-netty/src/main/java/org/apache/ratis/netty/server/ReadStreamManagement.java
@@ -17,9 +17,11 @@
*/
package org.apache.ratis.netty.server;
+import org.apache.ratis.client.impl.OrderedAsync;
import org.apache.ratis.conf.RaftProperties;
import org.apache.ratis.datastream.impl.DataStreamReplyByteBuffer;
import org.apache.ratis.datastream.impl.DataStreamRequestByteBuf;
+import org.apache.ratis.proto.RaftProtos;
import org.apache.ratis.proto.RaftProtos.DataStreamPacketHeaderProto.Type;
import org.apache.ratis.proto.RaftProtos.RaftClientRequestProto;
import org.apache.ratis.proto.RaftProtos.RaftClientRequestProto.TypeCase;
@@ -47,6 +49,7 @@ import java.util.concurrent.ExecutorService;
import static
org.apache.ratis.client.impl.ClientProtoUtils.toRaftClientRequest;
import static
org.apache.ratis.client.impl.ClientProtoUtils.toRaftClientReplyProto;
+import static
org.apache.ratis.netty.server.DataStreamManagement.newDataStreamReplyByteBuffer;
import static
org.apache.ratis.netty.server.DataStreamManagement.replyDataStreamException;
public class ReadStreamManagement {
@@ -60,23 +63,24 @@ public class ReadStreamManagement {
private final DataStreamReplyByteBuffer terminalReply;
private long streamOffset;
- ReadStream(RaftClientRequest request, long streamId, ChannelHandlerContext
ctx) {
+ ReadStream(RaftClientRequest request, long streamId, ChannelHandlerContext
ctx, RaftClientReply terminalReply) {
this.clientId = request.getClientId();
this.streamId = streamId;
this.ctx = ctx;
- final RaftClientReply reply = RaftClientReply.newBuilder()
- .setRequest(request)
- .setSuccess()
- .build();
- this.terminalReply = DataStreamReplyByteBuffer.newBuilder()
+ this.terminalReply = newReadStreamTerminalReply(clientId, streamId,
terminalReply);
+ }
+
+ private static DataStreamReplyByteBuffer newReadStreamTerminalReply(
+ ClientId clientId, long streamId, RaftClientReply reply) {
+ return DataStreamReplyByteBuffer.newBuilder()
.setClientId(clientId)
.setType(Type.STREAM_HEADER)
.setStreamId(streamId)
.setStreamOffset(0)
.setBuffer(toRaftClientReplyProto(reply).toByteString().asReadOnlyByteBuffer())
- .setSuccess(true)
- .setBytesWritten(0)
+ .setSuccess(reply.isSuccess())
+ .setCommitInfos(reply.getCommitInfos())
.build();
}
@@ -186,17 +190,45 @@ public class ReadStreamManagement {
return true;
}
- final ReadStream stream = new ReadStream(request,
requestBuf.getStreamId(), ctx);
- requestExecutor.execute(() -> {
+ final CompletableFuture<RaftClientReply> readCheck;
+ try {
+ readCheck =
server.submitClientRequestAsync(newDummyReadRequest(request));
+ } catch (IOException e) {
+ replyDataStreamException(server, e, request, requestBuf, ctx);
+ return true;
+ }
+
+ readCheck.whenCompleteAsync((readCheckReply, exception) -> {
+ if (exception != null) {
+ replyDataStreamException(server, exception, request, requestBuf, ctx);
+ return;
+ }
+
+ if (!readCheckReply.isSuccess()) {
+ ctx.writeAndFlush(newDataStreamReplyByteBuffer(requestBuf,
readCheckReply));
+ return;
+ }
+
+ final ReadStream stream = new ReadStream(request,
requestBuf.getStreamId(), ctx, readCheckReply);
try {
division.getStateMachine().data().query(request.getMessage(), stream);
} catch (Throwable t) {
LOG.error("{}: Failed read-only data stream query for {}", this,
request, t);
}
- });
+ }, requestExecutor);
return true;
}
+ private static RaftClientRequest newDummyReadRequest(RaftClientRequest
request) {
+ final RaftProtos.ReadRequestTypeProto original =
request.getType().getRead();
+ return RaftClientRequest.newBuilder()
+ .set(request)
+ .setMessage(OrderedAsync.DUMMY)
+ .setType(RaftClientRequest.readRequestType(
+ original.getPreferNonLinearizable(),
original.getReadAfterWriteConsistent(), true))
+ .build();
+ }
+
@Override
public String toString() {
return name;
diff --git a/ratis-proto/src/main/proto/Raft.proto
b/ratis-proto/src/main/proto/Raft.proto
index eba5de3b7..c1bb9d64e 100644
--- a/ratis-proto/src/main/proto/Raft.proto
+++ b/ratis-proto/src/main/proto/Raft.proto
@@ -302,6 +302,7 @@ message ForwardRequestTypeProto {
message ReadRequestTypeProto {
bool preferNonLinearizable = 1;
bool readAfterWriteConsistent = 2;
+ bool dummy = 3;
}
message StaleReadRequestTypeProto {
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 f758fd0ed..f4e740407 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
@@ -43,6 +43,7 @@ import
org.apache.ratis.proto.RaftProtos.RequestVoteRequestProto;
import org.apache.ratis.proto.RaftProtos.RoleInfoProto;
import org.apache.ratis.proto.RaftProtos.StartLeaderElectionReplyProto;
import org.apache.ratis.proto.RaftProtos.StartLeaderElectionRequestProto;
+import org.apache.ratis.protocol.ClientId;
import org.apache.ratis.protocol.ClientInvocationId;
import org.apache.ratis.protocol.GroupInfoReply;
import org.apache.ratis.protocol.GroupInfoRequest;
@@ -260,6 +261,8 @@ class RaftServerImpl implements RaftServer.Division,
private final ExecutorService clientExecutor;
private final ThreadGroup threadGroup;
+ private final CompletableFuture<RaftClientReply> dummySuccessReply;
+
RaftServerImpl(RaftGroup group, StateMachine stateMachine, RaftServerProxy
proxy, RaftStorage.StartupOption option)
throws IOException {
final RaftPeerId id = proxy.getId();
@@ -303,6 +306,13 @@ class RaftServerImpl implements RaftServer.Division,
RaftServerConfigKeys.ThreadPool.clientSize(properties),
id + "-client");
this.threadGroup = new ThreadGroup(proxy.getThreadGroup(),
getMemberId().toString());
+
+ this.dummySuccessReply =
CompletableFuture.completedFuture(RaftClientReply.newBuilder()
+ .setClientId(ClientId.emptyClientId())
+ .setServerId(id)
+ .setGroupId(group.getGroupId())
+ .setSuccess()
+ .build());
}
private long getCommitIndex(RaftPeerId id) {
@@ -1113,11 +1123,12 @@ class RaftServerImpl implements RaftServer.Division,
if (request.getType().getRead().getPreferNonLinearizable()
|| readOption == RaftServerConfigKeys.Read.Option.DEFAULT) {
final CompletableFuture<RaftClientReply> reply =
checkLeaderState(request);
- if (reply != null) {
- return reply;
- }
- return queryStateMachine(request);
- } else if (readOption == RaftServerConfigKeys.Read.Option.LINEARIZABLE){
+ if (reply != null) {
+ return reply;
+ }
+ return isDummyRead(request) ?
CompletableFuture.completedFuture(newSuccessReply(request))
+ : queryStateMachine(request);
+ } else if (readOption == RaftServerConfigKeys.Read.Option.LINEARIZABLE) {
final LeaderStateImpl leader = role.getLeaderState().orElse(null);
final CompletableFuture<Long> replyFuture;
if (leader != null) {
@@ -1136,12 +1147,17 @@ class RaftServerImpl implements RaftServer.Division,
return replyFuture
.thenCompose(readIndex ->
getState().getReadRequests().waitToAdvance(readIndex,
() -> getReadException("add",
snapshotInstallationHandler.getInProgressInstallSnapshotIndex(), false)))
- .thenCompose(readIndex -> queryStateMachine(request))
+ .thenCompose(readIndex -> isDummyRead(request)
+ ? CompletableFuture.completedFuture(newSuccessReply(request)) :
queryStateMachine(request))
.exceptionally(e -> readException2Reply(request, e));
} else {
throw new IllegalStateException("Unexpected read option: " + readOption);
}
}
+ private static boolean isDummyRead(RaftClientRequest request) {
+ return request.getMessage() != null &&
OrderedAsync.DUMMY.getContent().equals(request.getMessage().getContent());
+ }
+
private RaftClientReply readException2Reply(RaftClientRequest request,
Throwable e) {
e = JavaUtils.unwrapCompletionException(e);
if (e instanceof StateMachineException ) {
@@ -1183,6 +1199,9 @@ class RaftServerImpl implements RaftServer.Division,
}
CompletableFuture<RaftClientReply> queryStateMachine(RaftClientRequest
request) {
+ if (request.getType().getRead().getDummy()) {
+ return dummySuccessReply;
+ }
return processQueryFuture(stateMachine.query(request.getMessage()),
request);
}
diff --git
a/ratis-test/src/test/java/org/apache/ratis/client/impl/TestDataStreamClientImpl.java
b/ratis-test/src/test/java/org/apache/ratis/client/impl/TestDataStreamClientImpl.java
index 523c49f05..a85a9f483 100644
---
a/ratis-test/src/test/java/org/apache/ratis/client/impl/TestDataStreamClientImpl.java
+++
b/ratis-test/src/test/java/org/apache/ratis/client/impl/TestDataStreamClientImpl.java
@@ -22,9 +22,7 @@ import org.apache.ratis.client.DataStreamClientRpc;
import org.apache.ratis.client.api.DataStreamInput;
import org.apache.ratis.conf.RaftProperties;
import org.apache.ratis.datastream.DataStreamObserver;
-import org.apache.ratis.datastream.impl.DataStreamReplyByteBuf;
import org.apache.ratis.datastream.impl.DataStreamRequestByteBuffer;
-import org.apache.ratis.proto.RaftProtos.DataStreamPacketHeaderProto.Type;
import org.apache.ratis.proto.RaftProtos.RaftClientRequestProto;
import org.apache.ratis.protocol.ClientId;
import org.apache.ratis.protocol.DataStreamReply;
@@ -32,7 +30,6 @@ import org.apache.ratis.protocol.DataStreamRequest;
import org.apache.ratis.protocol.RaftClientRequest;
import org.apache.ratis.protocol.RaftGroupId;
import org.apache.ratis.protocol.RaftPeer;
-import org.apache.ratis.thirdparty.io.netty.buffer.Unpooled;
import org.apache.ratis.util.ReferenceCountedObject;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -69,20 +66,6 @@ public class TestDataStreamClientImpl {
return future;
}
- RaftClientRequest getRequest() {
- return request.get();
- }
-
- void receive(DataStreamReplyByteBuf reply) {
- final ReferenceCountedObject<DataStreamReply> ref =
DataStreamReplyByteBuf.asReferenceCounted(reply);
- ref.retain();
- try {
- replyHandler.get().onNext(ref);
- } finally {
- ref.release();
- }
- }
-
void complete() {
replyHandler.get().onCompleted();
replyFuture.get().complete(null);
diff --git
a/ratis-test/src/test/java/org/apache/ratis/datastream/DataStreamClusterTests.java
b/ratis-test/src/test/java/org/apache/ratis/datastream/DataStreamClusterTests.java
index 95e9bef49..7478c3628 100644
---
a/ratis-test/src/test/java/org/apache/ratis/datastream/DataStreamClusterTests.java
+++
b/ratis-test/src/test/java/org/apache/ratis/datastream/DataStreamClusterTests.java
@@ -32,9 +32,6 @@ import
org.apache.ratis.datastream.impl.DataStreamReplyByteBuf;
import org.apache.ratis.proto.RaftProtos.DataStreamPacketHeaderProto.Type;
import org.apache.ratis.proto.RaftProtos.ReplicationLevel;
import org.apache.ratis.protocol.DataStreamReply;
-import org.apache.ratis.protocol.DataStreamRequest;
-import org.apache.ratis.protocol.DataStreamRequestHeader;
-import org.apache.ratis.protocol.Message;
import org.apache.ratis.protocol.RaftClientReply;
import org.apache.ratis.protocol.RaftClientRequest;
import org.apache.ratis.retry.RetryPolicies;
diff --git
a/ratis-test/src/test/java/org/apache/ratis/netty/server/TestDataStreamManagement.java
b/ratis-test/src/test/java/org/apache/ratis/netty/server/TestDataStreamManagement.java
index 188f119fc..56ecd166b 100644
---
a/ratis-test/src/test/java/org/apache/ratis/netty/server/TestDataStreamManagement.java
+++
b/ratis-test/src/test/java/org/apache/ratis/netty/server/TestDataStreamManagement.java
@@ -19,6 +19,7 @@ package org.apache.ratis.netty.server;
import org.apache.ratis.client.impl.ClientProtoUtils;
import org.apache.ratis.client.impl.DataStreamClientImpl.DataStreamOutputImpl;
+import org.apache.ratis.client.impl.OrderedAsync;
import org.apache.ratis.conf.RaftProperties;
import org.apache.ratis.datastream.impl.DataStreamReplyByteBuffer;
import org.apache.ratis.datastream.impl.DataStreamRequestByteBuf;
@@ -28,10 +29,12 @@ import
org.apache.ratis.proto.RaftProtos.DataStreamPacketHeaderProto.Type;
import org.apache.ratis.protocol.ClientId;
import org.apache.ratis.protocol.DataStreamReply;
import org.apache.ratis.protocol.Message;
+import org.apache.ratis.protocol.RaftClientReply;
import org.apache.ratis.protocol.RaftClientRequest;
import org.apache.ratis.protocol.RaftGroupId;
import org.apache.ratis.protocol.RaftPeer;
import org.apache.ratis.protocol.RaftPeerId;
+import org.apache.ratis.protocol.exceptions.ReadIndexException;
import org.apache.ratis.server.RaftServer;
import org.apache.ratis.statemachine.StateMachine;
import org.apache.ratis.statemachine.StateMachine.DataApi;
@@ -59,14 +62,17 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Set;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Function;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
class TestDataStreamManagement {
@@ -96,8 +102,9 @@ class TestDataStreamManagement {
assertTrue(management.process(readOnlyRequest.request,
embeddedChannel.pipeline().firstContext()));
assertEquals(0, readOnlyRequest.headerBuf.refCnt());
+ JavaUtils.attempt(() -> assertNotNull(streamRef.get()), 10,
+ TimeDuration.valueOf(100, TimeUnit.MILLISECONDS), "read-only
stream", null);
final WritableByteChannel stream = streamRef.get();
- assertNotNull(stream);
stream.write(response.asReadOnlyByteBuffer());
stream.close();
@@ -116,6 +123,99 @@ class TestDataStreamManagement {
assertTrue(ClientProtoUtils.getRaftClientReply(replies.get(1)).isSuccess());
} finally {
embeddedChannel.finishAndReleaseAll();
+ management.shutdown();
+ }
+ }
+
+ @Test
+ void readOnlyRequestWaitsForLinearizableCheck() throws Exception {
+ final RaftPeerId serverId = RaftPeerId.valueOf("s1");
+ final ClientId clientId = ClientId.randomId();
+ final RaftGroupId groupId = RaftGroupId.randomId();
+ final ByteString query = ByteString.copyFromUtf8("query");
+ final CompletableFuture<RaftClientReply> readOnlyCheck = new
CompletableFuture<>();
+ final AtomicReference<RaftClientRequest> submittedReadOnlyCheck = new
AtomicReference<>();
+ final AtomicReference<Message> messageRef = new AtomicReference<>();
+ final AtomicReference<WritableByteChannel> streamRef = new
AtomicReference<>();
+
+ final DataApi dataApi = new DataApi() {
+ @Override
+ public void query(Message request, WritableByteChannel stream) {
+ messageRef.set(request);
+ streamRef.set(stream);
+ }
+ };
+ final ReadStreamManagement management = newReadStreamManagement(serverId,
groupId, dataApi, request -> {
+ submittedReadOnlyCheck.set(request);
+ return readOnlyCheck;
+ });
+ final EmbeddedChannel embeddedChannel = new EmbeddedChannel(new
ChannelInboundHandlerAdapter());
+ final ReadOnlyRequest readOnlyRequest = newReadOnlyRequest(clientId,
serverId, groupId, 1L, query);
+
+ try {
+ assertTrue(management.process(readOnlyRequest.request,
embeddedChannel.pipeline().firstContext()));
+ assertEquals(0, readOnlyRequest.headerBuf.refCnt());
+
+ final RaftClientRequest checkRequest = submittedReadOnlyCheck.get();
+ assertNotNull(checkRequest);
+ assertEquals(OrderedAsync.DUMMY.getContent(),
checkRequest.getMessage().getContent());
+ assertNull(streamRef.get(), "state machine query should wait for the
read-only check");
+
+
readOnlyCheck.complete(RaftClientReply.newBuilder().setRequest(checkRequest).setSuccess().build());
+ JavaUtils.attempt(() -> assertNotNull(streamRef.get()), 10,
+ TimeDuration.valueOf(100, TimeUnit.MILLISECONDS), "linearizable
read-only stream", null);
+ assertEquals(query, messageRef.get().getContent());
+ } finally {
+ embeddedChannel.finishAndReleaseAll();
+ management.shutdown();
+ }
+ }
+
+ @Test
+ void readOnlyCheckFailureSkipsStateMachineQuery() throws Exception {
+ final RaftPeerId serverId = RaftPeerId.valueOf("s1");
+ final ClientId clientId = ClientId.randomId();
+ final RaftGroupId groupId = RaftGroupId.randomId();
+ final ByteString query = ByteString.copyFromUtf8("query");
+ final AtomicBoolean queryCalled = new AtomicBoolean();
+
+ final DataApi dataApi = new DataApi() {
+ @Override
+ public void query(Message request, WritableByteChannel stream) {
+ queryCalled.set(true);
+ }
+ };
+ final ReadStreamManagement management = newReadStreamManagement(serverId,
groupId, dataApi, request ->
+ CompletableFuture.completedFuture(RaftClientReply.newBuilder()
+ .setRequest(request)
+ .setException(new ReadIndexException("read index failed"))
+ .build()));
+ final EmbeddedChannel embeddedChannel = new EmbeddedChannel(new
ChannelInboundHandlerAdapter());
+ final ReadOnlyRequest readOnlyRequest = newReadOnlyRequest(clientId,
serverId, groupId, 1L, query);
+
+ try {
+ assertTrue(management.process(readOnlyRequest.request,
embeddedChannel.pipeline().firstContext()));
+ assertEquals(0, readOnlyRequest.headerBuf.refCnt());
+
+ final List<DataStreamReply> replies = new ArrayList<>();
+ JavaUtils.attempt(() -> {
+ for (Object outbound; (outbound = embeddedChannel.readOutbound()) !=
null;) {
+ replies.add((DataStreamReply) outbound);
+ }
+ assertEquals(1, replies.size());
+ }, 10, TimeDuration.valueOf(100, TimeUnit.MILLISECONDS), "read-only
check failure reply", null);
+
+ assertFalse(queryCalled.get(), "state machine query should not run when
the read-only check fails");
+ final DataStreamReply reply = replies.get(0);
+ assertEquals(Type.STREAM_HEADER, reply.getType());
+ assertFalse(reply.isSuccess());
+ final RaftClientReply clientReply =
ClientProtoUtils.getRaftClientReply(reply);
+ assertFalse(clientReply.isSuccess());
+ assertNotNull(clientReply.getReadIndexException());
+ assertEquals(serverId, clientReply.getServerId());
+ } finally {
+ embeddedChannel.finishAndReleaseAll();
+ management.shutdown();
}
}
@@ -231,13 +331,19 @@ class TestDataStreamManagement {
private static ReadStreamManagement newReadStreamManagement(
RaftPeerId serverId, RaftGroupId groupId, DataApi dataApi) {
+ return newReadStreamManagement(serverId, groupId, dataApi,
TestDataStreamManagement::successReply);
+ }
+
+ private static ReadStreamManagement newReadStreamManagement(RaftPeerId
serverId, RaftGroupId groupId,
+ DataApi dataApi, Function<RaftClientRequest,
CompletableFuture<RaftClientReply>> submitClientRequestAsync) {
final StateMachine stateMachine = new BaseStateMachine() {
@Override
public DataApi data() {
return dataApi;
}
};
- final RaftServer server = newRaftServer(serverId, new RaftProperties(),
groupId, newDivision(stateMachine));
+ final RaftServer server = newRaftServer(serverId, new RaftProperties(),
groupId, newDivision(stateMachine),
+ submitClientRequestAsync);
return new ReadStreamManagement(server);
}
@@ -275,6 +381,16 @@ class TestDataStreamManagement {
private static RaftServer newRaftServer(RaftPeerId serverId, RaftProperties
properties,
RaftGroupId groupId, RaftServer.Division division) {
+ return newRaftServer(serverId, properties, groupId, division,
TestDataStreamManagement::successReply);
+ }
+
+ private static CompletableFuture<RaftClientReply>
successReply(RaftClientRequest request) {
+ return
CompletableFuture.completedFuture(RaftClientReply.newBuilder().setRequest(request).setSuccess().build());
+ }
+
+ private static RaftServer newRaftServer(RaftPeerId serverId, RaftProperties
properties,
+ RaftGroupId groupId, RaftServer.Division division,
+ Function<RaftClientRequest, CompletableFuture<RaftClientReply>>
submitClientRequestAsync) {
return (RaftServer)
Proxy.newProxyInstance(RaftServer.class.getClassLoader(), new
Class<?>[]{RaftServer.class},
(proxy, method, args) -> {
switch (method.getName()) {
@@ -287,6 +403,8 @@ class TestDataStreamManagement {
return division;
}
throw new IOException("Division not found: " + args[0]);
+ case "submitClientRequestAsync":
+ return submitClientRequestAsync.apply((RaftClientRequest) args[0]);
case "close":
return null;
case "toString":
