Add a DUMMY message to the dummy watch request.

Author: szetszwo

ratis-client/src/main/java/org/apache/ratis/client/impl/OrderedAsync.java

@@ -58,6 +58,8 @@
 public final class OrderedAsync {
   public static final Logger LOG = LoggerFactory.getLogger(OrderedAsync.class);
 
+  public static final Message DUMMY = Message.valueOf("DUMMY");
+
   private enum BatchLogKey implements BatchLogger.Key {
     SEND_REQUEST_EXCEPTION
   }
@@ -121,7 +123,7 @@ static OrderedAsync newInstance(RaftClientImpl client, RaftPeerId server, RaftPr
     // send a dummy watch request to establish the connection
     // TODO: this is a work around, it is better to fix the underlying RPC implementation
     if (RaftClientConfigKeys.Async.Experimental.sendDummyRequest(properties)) {
-      ordered.send(RaftClientRequest.watchRequestType(), null, server);
+      ordered.send(RaftClientRequest.watchRequestType(), DUMMY, server);
     }
     return ordered;
   }

ratis-common/src/main/java/org/apache/ratis/protocol/Message.java

@@ -19,6 +19,7 @@
 
 import org.apache.ratis.thirdparty.com.google.protobuf.AbstractMessage;
 import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
+import org.apache.ratis.thirdparty.com.google.protobuf.TextFormat;
 import org.apache.ratis.util.MemoizedSupplier;
 import org.apache.ratis.util.StringUtils;
 
@@ -47,11 +48,11 @@ public String toString() {
   }
 
   static Message valueOf(AbstractMessage abstractMessage) {
-    return valueOf(abstractMessage.toByteString(), abstractMessage::toString);
+    return valueOf(abstractMessage.toByteString(), () -> TextFormat.shortDebugString(abstractMessage));
   }
 
   static Message valueOf(ByteString bytes) {
-    return valueOf(bytes, () -> "Message:" + StringUtils.bytes2HexShortString(bytes));
+    return valueOf(bytes, () -> "Message:" + StringUtils.bytes2ShortString(bytes));
   }
 
   static Message valueOf(String string) {

ratis-common/src/main/java/org/apache/ratis/util/MemoizedFunction.java

@@ -58,7 +58,7 @@ private MemoizedFunction(Function<PARAMETER, RETURN> initializer) {
   /**
    * @param parameter for passing to the initializer.
    *                  Since the returned function is memoized, the parameter is only used at the first call.
-   *                  The parameter in the subsequent calls is ignored.
+   *                  The parameter is ignored in the subsequent calls.
    *
    * @return the lazily initialized object.
    */

ratis-common/src/main/java/org/apache/ratis/util/StringUtils.java

@@ -81,6 +81,23 @@ public static String format(final String format, final Object... objects) {
     return String.format(Locale.ENGLISH, format, objects);
   }
 
+  public static String bytes2ShortString(ByteString bytes) {
+    return bytes.isValidUtf8() ? bytes2ShortUtf8String(bytes) : StringUtils.bytes2HexShortString(bytes);
+  }
+
+  public static String bytes2ShortUtf8String(ByteString bytes) {
+    final String utf8 = bytes.toStringUtf8();
+    if (utf8.isEmpty()) {
+      return "<EMPTY_UTF8>";
+    }
+    final int length = utf8.length();
+    if (length <= 10) {
+      return utf8;
+    }
+    // return only the first 10 characters
+    return utf8.substring(0, 10) + "...(length=" + length + ")";
+  }
+
   public static String bytes2HexShortString(ByteString bytes) {
     final int size = bytes.size();
     if (size == 0) {

ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java

@@ -18,6 +18,7 @@
 package org.apache.ratis.server.impl;
 
 import org.apache.ratis.client.impl.ClientProtoUtils;
+import org.apache.ratis.client.impl.OrderedAsync;
 import org.apache.ratis.conf.RaftProperties;
 import org.apache.ratis.metrics.Timekeeper;
 import org.apache.ratis.proto.RaftProtos.AppendEntriesReplyProto;
@@ -1064,6 +1065,10 @@ private CompletableFuture<RaftClientReply> writeAsyncImpl(RaftClientRequest requ
   }
 
   private CompletableFuture<RaftClientReply> watchAsync(RaftClientRequest request) {
+    if (OrderedAsync.DUMMY.getContent().equals(request.getMessage().getContent())) {
+      return CompletableFuture.completedFuture(RaftClientReply.newBuilder().setRequest(request).build());
+    }
+
     final CompletableFuture<RaftClientReply> reply = checkLeaderState(request);
     if (reply != null) {
       return reply;
@@ -1078,7 +1083,7 @@ private CompletableFuture<RaftClientReply> watchAsync(RaftClientRequest request)
   private CompletableFuture<RaftClientReply> staleReadAsync(RaftClientRequest request) {
     final long minIndex = request.getType().getStaleRead().getMinIndex();
     final long commitIndex = state.getLog().getLastCommittedIndex();
-    LOG.debug("{}: minIndex={}, commitIndex={}", getMemberId(), minIndex, commitIndex);
+    LOG.debug("{}: minIndex={}, commitIndex={} from {}", getMemberId(), minIndex, commitIndex, request.getClientId());
     if (commitIndex < minIndex) {
       final StaleReadException e = new StaleReadException(
           "Unable to serve stale-read due to server commit index = " + commitIndex + " < min = " + minIndex);

ratis-server/src/test/java/org/apache/ratis/LinearizableReadTests.java

@@ -88,10 +88,9 @@ public void setup() {
     RaftServerConfigKeys.Read.setOption(p, LINEARIZABLE);
     RaftServerConfigKeys.Read.setLeaderLeaseEnabled(p, isLeaderLeaseEnabled());
     RaftServerConfigKeys.Read.ReadIndex.setType(p, readIndexType());
-    // Disable dummy request since currently the request is implemented as a watch request
-    // which can cause follower client to trigger failover to leader which will cause the
-    // all reads to be sent to the leader, making the follower read moot.
-    RaftClientConfigKeys.Async.Experimental.setSendDummyRequest(p, false);
+
+    // Enable dummy request so linearizable-read tests exercise the default ordered-async bootstrap path.
+    RaftClientConfigKeys.Async.Experimental.setSendDummyRequest(p, true);
   }
 
   @Test
@@ -151,13 +150,16 @@ static <C extends MiniRaftCluster> void runTestFollowerLinearizableRead(C cluste
     final int n = 100;
     final List<Reply> f0Replies = new ArrayList<>(n);
     final List<Reply> f1Replies = new ArrayList<>(n);
-    try (RaftClient client = cluster.createClient(leaderId)) {
+    try (RaftClient client = cluster.createClient(leaderId);
+         RaftClient c0 = cluster.createClient(f0);
+         RaftClient c1 = cluster.createClient(f1);
+         ) {
       for (int i = 0; i < n; i++) {
         final int count = i + 1;
         assertReplyExact(count, client.io().send(INCREMENT));
 
-        f0Replies.add(new Reply(count, client.async().sendReadOnly(QUERY, f0)));
-        f1Replies.add(new Reply(count, client.async().sendReadOnly(QUERY, f1)));
+        f0Replies.add(new Reply(count, c0.async().sendReadOnly(QUERY, f0)));
+        f1Replies.add(new Reply(count, c1.async().sendReadOnly(QUERY, f1)));
       }
 
       for (int i = 0; i < n; i++) {

ratis-server/src/test/java/org/apache/ratis/server/impl/MiniRaftCluster.java

@@ -67,6 +67,7 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.Timer;
+import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -93,6 +94,8 @@ public abstract class MiniRaftCluster implements Closeable {
       TimeDuration.valueOf(100, TimeUnit.MILLISECONDS);
   static final AtomicInteger THREAD_COUNT = new AtomicInteger(0);
 
+  static final AtomicInteger CLIENT_ID = new AtomicInteger(0);
+
   public abstract static class Factory<CLUSTER extends MiniRaftCluster> {
     public interface Get<CLUSTER extends MiniRaftCluster> {
       Supplier<RaftProperties> PROPERTIES = JavaUtils.memoize(RaftProperties::new);
@@ -747,6 +750,7 @@ public RaftClient createClient(RaftPeerId leaderId, RaftGroup raftGroup, RetryPo
 
   public RaftClient createClient(RaftPeerId leaderId, RaftGroup raftGroup, RetryPolicy retryPolicy, RaftPeer primaryServer) {
     RaftClient.Builder builder = RaftClient.newBuilder()
+        .setClientId(ClientId.valueOf(new UUID(0, CLIENT_ID.incrementAndGet())))
         .setRaftGroup(raftGroup)
         .setLeaderId(leaderId)
         .setProperties(properties)