fix mark

apache · May 27, 2024 · 2920062 · 2920062
1 parent 640dfc6
commit 2920062
Show file tree

Hide file tree

Showing 5 changed files with 11 additions and 5 deletions.
diff --git a/.../consensus/src/main/java/org/apache/iotdb/consensus/common/request/IConsensusRequest.java b/.../consensus/src/main/java/org/apache/iotdb/consensus/common/request/IConsensusRequest.java
@@ -35,4 +35,8 @@ public interface IConsensusRequest {
    * changed or an error may occur
    */
   ByteBuffer serializeToByteBuffer();
+
+  default void markAsGeneratedByConsensus() {
+    // do nothing by default
+  }
 }
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
@@ -288,7 +288,7 @@ public TSStatus write(ConsensusGroupId groupId, IConsensusRequest request)
     }
 
     // serialize request into Message
-    Message message = new RequestMessage(request);
+    RequestMessage message = new RequestMessage(request);
 
     // 1. first try the local server
     RaftClientRequest clientRequest;
@@ -319,6 +319,7 @@ && waitUntilLeaderReady(raftGroupId)) {
 
     // 2. try raft client
     TSStatus writeResult;
+    message.getActualRequest().markAsGeneratedByConsensus();
     try (AutoCloseable ignored =
             RatisMetricsManager.getInstance().startWriteRemotelyTimer(consensusGroupType);
         RatisClient client = getRaftClient(raftGroup)) {

diff --git a/...in/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataRegionStateMachine.java b/...in/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataRegionStateMachine.java
@@ -233,9 +233,7 @@ public List<Path> getSnapshotFiles(File latestSnapshotRootDir) {
   @Override
   public TSStatus write(IConsensusRequest request) {
     try {
-      PlanNode planNode = (PlanNode) request;
-      planNode.markAsGeneratedByConsensus();
-      return write(planNode);
+      return write((PlanNode) request);
     } catch (IllegalArgumentException e) {
       logger.error(e.getMessage(), e);
       return new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode());

diff --git a/...apache/iotdb/db/consensus/statemachine/dataregion/IoTConsensusDataRegionStateMachine.java b/...apache/iotdb/db/consensus/statemachine/dataregion/IoTConsensusDataRegionStateMachine.java
@@ -56,7 +56,9 @@ public TSStatus write(IConsensusRequest request) {
         List<TSStatus> subStatus = new LinkedList<>();
         for (IConsensusRequest consensusRequest :
             ((DeserializedBatchIndexedConsensusRequest) request).getInsertNodes()) {
-          subStatus.add(write((PlanNode) consensusRequest));
+          PlanNode writeNode = (PlanNode) consensusRequest;
+          writeNode.markAsGeneratedByConsensus();
+          subStatus.add(write(writeNode));
         }
         return new TSStatus().setSubStatus(subStatus);
       } else {

diff --git a/...tanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanNode.java b/...tanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanNode.java
@@ -76,6 +76,7 @@ public void markAsGeneratedByPipe() {
     isGeneratedByPipe = true;
   }
 
+  @Override
   public void markAsGeneratedByConsensus() {
     isGeneratedByConsensus = true;
   }