[TRaft]Raft Optimization for Time series Workloads

Author: LittleHealth

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/RouteBalancer.java

@@ -87,6 +87,8 @@ public class RouteBalancer implements IClusterStatusSubscriber {
               && ConsensusFactory.RATIS_CONSENSUS.equals(DATA_REGION_CONSENSUS_PROTOCOL_CLASS))
           || (CONF.isEnableAutoLeaderBalanceForIoTConsensus()
               && ConsensusFactory.IOT_CONSENSUS.equals(DATA_REGION_CONSENSUS_PROTOCOL_CLASS))
+          || (CONF.isEnableAutoLeaderBalanceForIoTConsensus()
+              && ConsensusFactory.TRAFT_CONSENSUS.equals(DATA_REGION_CONSENSUS_PROTOCOL_CLASS))
           || (CONF.isEnableAutoLeaderBalanceForIoTConsensus()
               && ConsensusFactory.IOT_CONSENSUS_V2.equals(DATA_REGION_CONSENSUS_PROTOCOL_CLASS))
           // The simple consensus protocol will always automatically designate itself as the leader
@@ -200,6 +202,7 @@ private void balanceRegionLeader(
                 newLeaderId);
             switch (consensusProtocolClass) {
               case ConsensusFactory.IOT_CONSENSUS:
+              case ConsensusFactory.TRAFT_CONSENSUS:
               case ConsensusFactory.SIMPLE_CONSENSUS:
                 // For IoTConsensus or SimpleConsensus protocol, change
                 // RegionRouteMap is enough

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandler.java

@@ -68,6 +68,7 @@
 import static org.apache.iotdb.consensus.ConsensusFactory.IOT_CONSENSUS;
 import static org.apache.iotdb.consensus.ConsensusFactory.IOT_CONSENSUS_V2;
 import static org.apache.iotdb.consensus.ConsensusFactory.RATIS_CONSENSUS;
+import static org.apache.iotdb.consensus.ConsensusFactory.TRAFT_CONSENSUS;
 
 public class RegionMaintainHandler {
 
@@ -148,7 +149,8 @@ public TSStatus createNewRegionPeer(TConsensusGroupId regionId, TDataNodeLocatio
     List<TDataNodeLocation> currentPeerNodes;
     if (TConsensusGroupType.DataRegion.equals(regionId.getType())
         && (IOT_CONSENSUS.equals(CONF.getDataRegionConsensusProtocolClass())
-            || IOT_CONSENSUS_V2.equals(CONF.getDataRegionConsensusProtocolClass()))) {
+            || IOT_CONSENSUS_V2.equals(CONF.getDataRegionConsensusProtocolClass())
+            || TRAFT_CONSENSUS.equals(CONF.getDataRegionConsensusProtocolClass()))) {
       // parameter of createPeer for MultiLeader should be all peers
       currentPeerNodes = new ArrayList<>(regionReplicaNodes);
       currentPeerNodes.add(destDataNode);

iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ConsensusFactory.java

@@ -37,6 +37,7 @@ public class ConsensusFactory {
   public static final String SIMPLE_CONSENSUS = "org.apache.iotdb.consensus.simple.SimpleConsensus";
   public static final String RATIS_CONSENSUS = "org.apache.iotdb.consensus.ratis.RatisConsensus";
   public static final String IOT_CONSENSUS = "org.apache.iotdb.consensus.iot.IoTConsensus";
+  public static final String TRAFT_CONSENSUS = "org.apache.iotdb.consensus.traft.TRaftConsensus";
   public static final String REAL_PIPE_CONSENSUS = "org.apache.iotdb.consensus.pipe.PipeConsensus";
   public static final String IOT_CONSENSUS_V2 = "org.apache.iotdb.consensus.iot.IoTConsensusV2";
   public static final String IOT_CONSENSUS_V2_BATCH_MODE = "batch";

iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/common/request/IConsensusRequest.java

@@ -36,6 +36,20 @@ public interface IConsensusRequest {
    */
   ByteBuffer serializeToByteBuffer();
 
+  default boolean hasTime() {
+    return false;
+  }
+
+  /**
+   * Return the primary timestamp carried by this request.
+   *
+   * <p>Callers should check {@link #hasTime()} before calling this method.
+   */
+  default long getTime() {
+    throw new UnsupportedOperationException(
+        String.format("%s does not carry timestamp", getClass().getName()));
+  }
+
   default long getMemorySize() {
     // return 0 by default
     return 0;

iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/config/ConsensusConfig.java

@@ -33,6 +33,7 @@ public class ConsensusConfig {
   private final RatisConfig ratisConfig;
   private final IoTConsensusConfig iotConsensusConfig;
   private final PipeConsensusConfig pipeConsensusConfig;
+  private final TRaftConfig tRaftConfig;
 
   private ConsensusConfig(
       TEndPoint thisNode,
@@ -41,14 +42,16 @@ private ConsensusConfig(
       TConsensusGroupType consensusGroupType,
       RatisConfig ratisConfig,
       IoTConsensusConfig iotConsensusConfig,
-      PipeConsensusConfig pipeConsensusConfig) {
+      PipeConsensusConfig pipeConsensusConfig,
+      TRaftConfig tRaftConfig) {
     this.thisNodeEndPoint = thisNode;
     this.thisNodeId = thisNodeId;
     this.storageDir = storageDir;
     this.consensusGroupType = consensusGroupType;
     this.ratisConfig = ratisConfig;
     this.iotConsensusConfig = iotConsensusConfig;
     this.pipeConsensusConfig = pipeConsensusConfig;
+    this.tRaftConfig = tRaftConfig;
   }
 
   public TEndPoint getThisNodeEndPoint() {
@@ -79,6 +82,10 @@ public PipeConsensusConfig getPipeConsensusConfig() {
     return pipeConsensusConfig;
   }
 
+  public TRaftConfig getTRaftConfig() {
+    return tRaftConfig;
+  }
+
   public static ConsensusConfig.Builder newBuilder() {
     return new ConsensusConfig.Builder();
   }
@@ -92,6 +99,7 @@ public static class Builder {
     private RatisConfig ratisConfig;
     private IoTConsensusConfig iotConsensusConfig;
     private PipeConsensusConfig pipeConsensusConfig;
+    private TRaftConfig tRaftConfig;
 
     public ConsensusConfig build() {
       return new ConsensusConfig(
@@ -103,7 +111,8 @@ public ConsensusConfig build() {
           Optional.ofNullable(iotConsensusConfig)
               .orElseGet(() -> IoTConsensusConfig.newBuilder().build()),
           Optional.ofNullable(pipeConsensusConfig)
-              .orElseGet(() -> PipeConsensusConfig.newBuilder().build()));
+              .orElseGet(() -> PipeConsensusConfig.newBuilder().build()),
+          Optional.ofNullable(tRaftConfig).orElseGet(() -> TRaftConfig.newBuilder().build()));
     }
 
     public Builder setThisNode(TEndPoint thisNode) {
@@ -140,5 +149,10 @@ public Builder setPipeConsensusConfig(PipeConsensusConfig pipeConsensusConfig) {
       this.pipeConsensusConfig = pipeConsensusConfig;
       return this;
     }
+
+    public Builder setTRaftConfig(TRaftConfig tRaftConfig) {
+      this.tRaftConfig = tRaftConfig;
+      return this;
+    }
   }
 }

iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/config/TRaftConfig.java

@@ -0,0 +1,141 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.consensus.config;
+
+import java.util.Optional;
+
+public class TRaftConfig {
+
+  private final Replication replication;
+  private final Election election;
+
+  private TRaftConfig(Replication replication, Election election) {
+    this.replication = replication;
+    this.election = election;
+  }
+
+  public Replication getReplication() {
+    return replication;
+  }
+
+  public Election getElection() {
+    return election;
+  }
+
+  public static Builder newBuilder() {
+    return new Builder();
+  }
+
+  public static class Builder {
+
+    private Replication replication;
+    private Election election;
+
+    public Builder setReplication(Replication replication) {
+      this.replication = replication;
+      return this;
+    }
+
+    public Builder setElection(Election election) {
+      this.election = election;
+      return this;
+    }
+
+    public TRaftConfig build() {
+      return new TRaftConfig(
+          Optional.ofNullable(replication).orElseGet(() -> Replication.newBuilder().build()),
+          Optional.ofNullable(election).orElseGet(() -> Election.newBuilder().build()));
+    }
+  }
+
+  public static class Replication {
+
+    private final int maxPendingRetryEntriesPerFollower;
+    private final long waitingReplicationTimeMs;
+
+    private Replication(int maxPendingRetryEntriesPerFollower, long waitingReplicationTimeMs) {
+      this.maxPendingRetryEntriesPerFollower = maxPendingRetryEntriesPerFollower;
+      this.waitingReplicationTimeMs = waitingReplicationTimeMs;
+    }
+
+    public int getMaxPendingRetryEntriesPerFollower() {
+      return maxPendingRetryEntriesPerFollower;
+    }
+
+    public long getWaitingReplicationTimeMs() {
+      return waitingReplicationTimeMs;
+    }
+
+    public static Builder newBuilder() {
+      return new Builder();
+    }
+
+    public static class Builder {
+
+      private int maxPendingRetryEntriesPerFollower = 100_000;
+      private long waitingReplicationTimeMs = 1L;
+
+      public Builder setMaxPendingRetryEntriesPerFollower(int maxPendingRetryEntriesPerFollower) {
+        this.maxPendingRetryEntriesPerFollower = maxPendingRetryEntriesPerFollower;
+        return this;
+      }
+
+      public Builder setWaitingReplicationTimeMs(long waitingReplicationTimeMs) {
+        this.waitingReplicationTimeMs = waitingReplicationTimeMs;
+        return this;
+      }
+
+      public Replication build() {
+        return new Replication(maxPendingRetryEntriesPerFollower, waitingReplicationTimeMs);
+      }
+    }
+  }
+
+  public static class Election {
+
+    private final long randomSeed;
+
+    private Election(long randomSeed) {
+      this.randomSeed = randomSeed;
+    }
+
+    public long getRandomSeed() {
+      return randomSeed;
+    }
+
+    public static Builder newBuilder() {
+      return new Builder();
+    }
+
+    public static class Builder {
+
+      private long randomSeed = System.nanoTime();
+
+      public Builder setRandomSeed(long randomSeed) {
+        this.randomSeed = randomSeed;
+        return this;
+      }
+
+      public Election build() {
+        return new Election(randomSeed);
+      }
+    }
+  }
+}