Skip to content

RATIS-2559. Add linearizable check for streaming read request #1490

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
szetszwo merged 6 commits into from
Jul 6, 2026
+256 −44
Merged

RATIS-2559. Add linearizable check for streaming read request #1490

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
eed01a4
RATIS-2559
peterxcli Jun 24, 2026
2704558
apply review suggestion
peterxcli Jul 1, 2026
4509555
fix build
peterxcli Jul 4, 2026
ed2f358
Merge branch 'master' into RATIS-2559
peterxcli Jul 4, 2026
8bb145c
Return Success Reply Per Query Request
peterxcli Jul 4, 2026
dbf7594
suggestion
peterxcli Jul 6, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions dev-support/checkstyle.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 -->
Expand Down
68 changes: 63 additions & 5 deletions ratis-common/src/main/java/org/apache/ratis/protocol/RaftClientRequest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -71,6 +71,44 @@ public static Type writeRequestType(ReplicationLevel replication) {
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);
}
Expand All @@ -91,6 +129,10 @@ public static Type messageStreamRequestType(long streamId, long messageId, boole
.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;
}
Expand Down Expand Up @@ -312,6 +354,22 @@ public RaftClientRequest build() {
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;
Expand Down
54 changes: 43 additions & 11 deletions ratis-netty/src/main/java/org/apache/ratis/netty/server/ReadStreamManagement.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -47,6 +49,7 @@

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 {
Expand All @@ -60,23 +63,24 @@ static class ReadStream implements WritableByteChannel {
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();
}

Expand Down Expand Up @@ -186,17 +190,45 @@ private boolean processImpl(DataStreamRequestByteBuf requestBuf, ChannelHandlerC
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;
Expand Down
1 change: 1 addition & 0 deletions ratis-proto/src/main/proto/Raft.proto
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -302,6 +302,7 @@ message ForwardRequestTypeProto {
message ReadRequestTypeProto {
bool preferNonLinearizable = 1;
bool readAfterWriteConsistent = 2;
bool dummy = 3;
}

message StaleReadRequestTypeProto {
Expand Down
31 changes: 25 additions & 6 deletions ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,7 @@
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;
Expand Down Expand Up @@ -260,6 +261,8 @@ public long[] getFollowerMatchIndices() {
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();
Expand Down Expand Up @@ -303,6 +306,13 @@ public long[] getFollowerMatchIndices() {
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) {
Expand Down Expand Up @@ -1113,11 +1123,12 @@ private CompletableFuture<RaftClientReply> readAsync(RaftClientRequest request)
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))

@amaliujia amaliujia Jun 26, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Just to learn the context, why do we need this dummy read in Ratis streaming read while we do not have it for Ratis streaming write?

@szetszwo szetszwo Jun 26, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It is to run the linearizable check; see #1469 (comment)

: queryStateMachine(request);
Comment on lines +1129 to +1130

@szetszwo szetszwo Jun 26, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Let's add a DUMMY_SUCCESS_REPLY and move it to queryStateMachine:

  static final CompletableFuture<RaftClientReply> DUMMY_SUCCESS_REPLY
      = CompletableFuture.completedFuture(RaftClientReply.newBuilder().setSuccess().build());
  CompletableFuture<RaftClientReply> queryStateMachine(RaftClientRequest request) {
    if (request.getType().getRead().getDummy()) {
      return DUMMY_SUCCESS_REPLY;
    }
    return processQueryFuture(stateMachine.query(request.getMessage()), request);
  }

} else if (readOption == RaftServerConfigKeys.Read.Option.LINEARIZABLE) {
final LeaderStateImpl leader = role.getLeaderState().orElse(null);
final CompletableFuture<Long> replyFuture;
if (leader != null) {
Expand All @@ -1136,12 +1147,17 @@ private CompletableFuture<RaftClientReply> readAsync(RaftClientRequest request)
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());
}
Comment on lines +1157 to +1159

@szetszwo szetszwo Jun 26, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We cannot use DUMMY since user applications can set any message. They may already have used it in their messages. We probably need to add a flag to the 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;
 }


private RaftClientReply readException2Reply(RaftClientRequest request, Throwable e) {
e = JavaUtils.unwrapCompletionException(e);
if (e instanceof StateMachineException ) {
Expand Down Expand Up @@ -1183,6 +1199,9 @@ private CompletableFuture<RaftClientRequest> streamEndOfRequestAsync(RaftClientR
}

CompletableFuture<RaftClientReply> queryStateMachine(RaftClientRequest request) {
if (request.getType().getRead().getDummy()) {
return dummySuccessReply;
}
return processQueryFuture(stateMachine.query(request.getMessage()), request);
}

Expand Down
17 changes: 0 additions & 17 deletions ratis-test/src/test/java/org/apache/ratis/client/impl/TestDataStreamClientImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,17 +22,14 @@
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;
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;
Expand Down Expand Up @@ -69,20 +66,6 @@ public CompletableFuture<DataStreamReply> streamAsync(
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);
Expand Down
3 changes: 0 additions & 3 deletions ratis-test/src/test/java/org/apache/ratis/datastream/DataStreamClusterTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,9 +32,6 @@
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;
Expand Down
Loading
Loading