RATIS-2559. Add linearizable check for streaming read request

[peterxcli]

What changes were proposed in this pull request?

• Refactored ReadStreamManagement to perform an asynchronous linearizability check using a "dummy read" request before invoking the state machine for read-only queries. The state machine query is only performed if the check succeeds, and an appropriate reply is sent if it fails.
• Updated RaftServerImpl to recognize and handle dummy read requests by returning a success reply immediately, avoiding unnecessary state machine queries.

What is the link to the Apache JIRA

https://issues.apache.org/jira/browse/RATIS-2559

How was this patch tested?

(Please explain how this patch was tested. Ex: unit tests, manual tests)
(If this patch involves UI changes, please attach a screen-shot; otherwise, remove this)

[peterxcli]

cc @szetszwo @amaliujia

[amaliujia]

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]

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

[szetszwo]

@peterxcli , thanks for working on this!

Please see the comments inlined and also https://issues.apache.org/jira/secure/attachment/13082966/1490_review.patch

[szetszwo]

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;
 }

[szetszwo]

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);
  }

[szetszwo]

Since server returns a DUMMY_SUCCESS_REPLY, we could build the failed reply directly:

      if (!readCheckReply.isSuccess()) {
        final RaftClientReply reply = RaftClientReply.newBuilder()
            .setRequest(request)
            .setSuccess(false)
            .build();
        ctx.writeAndFlush(newDataStreamReplyByteBuffer(requestBuf, reply));
        return;
      }

      final ReadStream stream = new ReadStream(request, requestBuf.getStreamId(), ctx);

[szetszwo]

Let's add a set(RaftClientRequest request) to Builder to copy everything.

[szetszwo]

@peterxcli , thanks for the update. There is a build failures. Could you take a look?

[peterxcli]

This somehow leads the test failure.

[peterxcli]

8bb145c (this PR)

[szetszwo]

I tried to build the dummySuccessReply by setting the required fields. It seems working.

@@ -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());
   }

@@ -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);
   }

[szetszwo]

+1 the change looks good.

[peterxcli]

@szetszwo thanks for the review and the review patch! let me start to work on the integration with ozone