[iotdb] branch rel/1.2 updated: [RatisConsensus] Support linearizable read when leader switches (#10699) (#10710)

[tanxinyu]

This is an automated email from the ASF dual-hosted git repository.

tanxinyu pushed a commit to branch rel/1.2
in repository https://gitbox.apache.org/repos/asf/iotdb.git


The following commit(s) were added to refs/heads/rel/1.2 by this push:
     new 315e4916c9a [RatisConsensus] Support linearizable read when leader 
switches (#10699) (#10710)
315e4916c9a is described below

commit 315e4916c9a951cbdf400efa6209307b1d7f38e4
Author: William Song <48054931+szywill...@users.noreply.github.com>
AuthorDate: Fri Jul 28 15:08:44 2023 +0800

    [RatisConsensus] Support linearizable read when leader switches (#10699) 
(#10710)
---
 .../ratis/ApplicationStateMachineProxy.java        | 24 +++++++++++++++++++---
 .../iotdb/consensus/ratis/RatisConsensus.java      |  3 ++-
 2 files changed, 23 insertions(+), 4 deletions(-)

diff --git 
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/ApplicationStateMachineProxy.java
 
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/ApplicationStateMachineProxy.java
index 4adc8cf41da..828d134d231 100644
--- 
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/ApplicationStateMachineProxy.java
+++ 
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/ApplicationStateMachineProxy.java
@@ -21,6 +21,7 @@ package org.apache.iotdb.consensus.ratis;
 
 import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
 import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.consensus.ConsensusGroupId;
 import org.apache.iotdb.commons.service.metric.PerformanceOverviewMetrics;
 import org.apache.iotdb.consensus.IStateMachine;
 import org.apache.iotdb.consensus.common.DataSet;
@@ -52,8 +53,11 @@ import java.io.File;
 import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.StandardCopyOption;
+import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 public class ApplicationStateMachineProxy extends BaseStateMachine {
 
@@ -64,11 +68,22 @@ public class ApplicationStateMachineProxy extends 
BaseStateMachine {
   private final IStateMachine.RetryPolicy retryPolicy;
   private final SnapshotStorage snapshotStorage;
   private final RaftGroupId groupId;
+  private final ConsensusGroupId consensusGroupId;
   private final TConsensusGroupType consensusGroupType;
+  private final ConcurrentHashMap<ConsensusGroupId, AtomicBoolean> 
canStaleRead;
 
-  public ApplicationStateMachineProxy(IStateMachine stateMachine, RaftGroupId 
id) {
-    applicationStateMachine = stateMachine;
-    groupId = id;
+  ApplicationStateMachineProxy(IStateMachine stateMachine, RaftGroupId id) {
+    this(stateMachine, id, null);
+  }
+
+  ApplicationStateMachineProxy(
+      IStateMachine stateMachine,
+      RaftGroupId id,
+      ConcurrentHashMap<ConsensusGroupId, AtomicBoolean> canStaleRead) {
+    this.applicationStateMachine = stateMachine;
+    this.canStaleRead = canStaleRead;
+    this.groupId = id;
+    this.consensusGroupId = Utils.fromRaftGroupIdToConsensusGroupId(id);
     retryPolicy =
         applicationStateMachine instanceof IStateMachine.RetryPolicy
             ? (IStateMachine.RetryPolicy) applicationStateMachine
@@ -293,6 +308,9 @@ public class ApplicationStateMachineProxy extends 
BaseStateMachine {
 
   @Override
   public void notifyLeaderChanged(RaftGroupMemberId groupMemberId, RaftPeerId 
newLeaderId) {
+    Optional.ofNullable(canStaleRead)
+        .ifPresent(
+            m -> m.computeIfAbsent(consensusGroupId, id -> new 
AtomicBoolean(false)).set(false));
     applicationStateMachine
         .event()
         .notifyLeaderChanged(
diff --git 
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java
 
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java
index f8a4b498353..27c8bb0d271 100644
--- 
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java
+++ 
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java
@@ -172,7 +172,8 @@ class RatisConsensus implements IConsensus {
                 raftGroupId ->
                     new ApplicationStateMachineProxy(
                         
registry.apply(Utils.fromRaftGroupIdToConsensusGroupId(raftGroupId)),
-                        raftGroupId))
+                        raftGroupId,
+                        canServeStaleRead))
             .build();
   }