diff --git a/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/DataNodeRemoveHandler.java b/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/DataNodeRemoveHandler.java
index d6b3b2b..d93ca83 100644
--- a/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/DataNodeRemoveHandler.java
+++ b/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/DataNodeRemoveHandler.java
@@ -558,8 +558,15 @@
       return Optional.empty();
     }
 
+    List<TDataNodeLocation> aliveDataNodes =
+        configManager.getNodeManager().filterDataNodeThroughStatus(NodeStatus.Running).stream()
+            .map(TDataNodeConfiguration::getLocation)
+            .collect(Collectors.toList());
+
     // TODO replace findAny() by select the low load node.
-    return regionReplicaNodes.stream().filter(e -> !e.equals(filterLocation)).findAny();
+    return regionReplicaNodes.stream()
+        .filter(e -> aliveDataNodes.contains(e) && !e.equals(filterLocation))
+        .findAny();
   }
 
   private String getIdWithRpcEndpoint(TDataNodeLocation location) {
diff --git a/consensus/src/main/java/org/apache/iotdb/consensus/exception/ConsensusGroupAddPeerException.java b/consensus/src/main/java/org/apache/iotdb/consensus/exception/ConsensusGroupAddPeerException.java
new file mode 100644
index 0000000..0260ba0
--- /dev/null
+++ b/consensus/src/main/java/org/apache/iotdb/consensus/exception/ConsensusGroupAddPeerException.java
@@ -0,0 +1,34 @@
+/*
+ * 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.exception;
+
+public class ConsensusGroupAddPeerException extends Exception {
+  public ConsensusGroupAddPeerException(String message) {
+    super(message);
+  }
+
+  public ConsensusGroupAddPeerException(Throwable cause) {
+    super(cause);
+  }
+
+  public ConsensusGroupAddPeerException(String message, Throwable cause) {
+    super(message, cause);
+  }
+}
diff --git a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/MultiLeaderConsensus.java b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/MultiLeaderConsensus.java
index d091772..2a9da3a 100644
--- a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/MultiLeaderConsensus.java
+++ b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/MultiLeaderConsensus.java
@@ -36,14 +36,19 @@
 import org.apache.iotdb.consensus.common.response.ConsensusWriteResponse;
 import org.apache.iotdb.consensus.config.ConsensusConfig;
 import org.apache.iotdb.consensus.config.MultiLeaderConfig;
+import org.apache.iotdb.consensus.exception.ConsensusException;
+import org.apache.iotdb.consensus.exception.ConsensusGroupAddPeerException;
 import org.apache.iotdb.consensus.exception.ConsensusGroupAlreadyExistException;
 import org.apache.iotdb.consensus.exception.ConsensusGroupNotExistException;
 import org.apache.iotdb.consensus.exception.IllegalPeerEndpointException;
 import org.apache.iotdb.consensus.exception.IllegalPeerNumException;
 import org.apache.iotdb.consensus.multileader.client.AsyncMultiLeaderServiceClient;
 import org.apache.iotdb.consensus.multileader.client.MultiLeaderConsensusClientPool.AsyncMultiLeaderServiceClientPoolFactory;
+import org.apache.iotdb.consensus.multileader.client.MultiLeaderConsensusClientPool.SyncMultiLeaderServiceClientPoolFactory;
+import org.apache.iotdb.consensus.multileader.client.SyncMultiLeaderServiceClient;
 import org.apache.iotdb.consensus.multileader.service.MultiLeaderRPCService;
 import org.apache.iotdb.consensus.multileader.service.MultiLeaderRPCServiceProcessor;
+import org.apache.iotdb.rpc.RpcUtils;
 import org.apache.iotdb.rpc.TSStatusCode;
 
 import org.slf4j.Logger;
@@ -73,6 +78,7 @@
   private final RegisterManager registerManager = new RegisterManager();
   private final MultiLeaderConfig config;
   private final IClientManager<TEndPoint, AsyncMultiLeaderServiceClient> clientManager;
+  private final IClientManager<TEndPoint, SyncMultiLeaderServiceClient> syncClientManager;
 
   public MultiLeaderConsensus(ConsensusConfig config, Registry registry) {
     this.thisNode = config.getThisNode();
@@ -84,6 +90,10 @@
         new IClientManager.Factory<TEndPoint, AsyncMultiLeaderServiceClient>()
             .createClientManager(
                 new AsyncMultiLeaderServiceClientPoolFactory(config.getMultiLeaderConfig()));
+    this.syncClientManager =
+        new IClientManager.Factory<TEndPoint, SyncMultiLeaderServiceClient>()
+            .createClientManager(
+                new SyncMultiLeaderServiceClientPoolFactory(config.getMultiLeaderConfig()));
   }
 
   @Override
@@ -116,6 +126,7 @@
                   new ArrayList<>(),
                   registry.apply(consensusGroupId),
                   clientManager,
+                  syncClientManager,
                   config);
           stateMachineMap.put(consensusGroupId, consensus);
           consensus.start();
@@ -144,6 +155,10 @@
     if (impl.isReadOnly()) {
       status = new TSStatus(TSStatusCode.READ_ONLY_SYSTEM_ERROR.getStatusCode());
       status.setMessage("Fail to do non-query operations because system is read-only.");
+    } else if (!impl.isActive()) {
+      // TODO: (xingtanzjr) whether we need to define a new status to indicate the inactive status ?
+      status = RpcUtils.getStatus(TSStatusCode.WRITE_PROCESS_REJECT);
+      status.setMessage("peer is inactive and not ready to receive sync log request.");
     } else {
       status = impl.write(request);
     }
@@ -191,6 +206,7 @@
                   peers,
                   registry.apply(groupId),
                   clientManager,
+                  syncClientManager,
                   config);
           impl.start();
           return impl;
@@ -225,12 +241,65 @@
 
   @Override
   public ConsensusGenericResponse addPeer(ConsensusGroupId groupId, Peer peer) {
-    return ConsensusGenericResponse.newBuilder().setSuccess(false).build();
+    MultiLeaderServerImpl impl = stateMachineMap.get(groupId);
+    if (impl == null) {
+      return ConsensusGenericResponse.newBuilder()
+          .setException(new ConsensusGroupNotExistException(groupId))
+          .build();
+    }
+    try {
+      // step 1: inactive new Peer to prepare for following steps
+      logger.info("[MultiLeaderConsensus] inactivate new peer: {}", peer);
+      impl.inactivePeer(peer);
+
+      // step 2: notify all the other Peers to build the sync connection to newPeer
+      logger.info("[MultiLeaderConsensus] notify current peers to build sync log...");
+      impl.notifyPeersToBuildSyncLogChannel(peer);
+
+      // step 3: take snapshot
+      logger.info("[MultiLeaderConsensus] start to take snapshot...");
+      impl.takeSnapshot();
+
+      // step 4: transit snapshot
+      logger.info("[MultiLeaderConsensus] start to transit snapshot...");
+      impl.transitSnapshot(peer);
+
+      // step 5: let the new peer load snapshot
+      logger.info("[MultiLeaderConsensus] trigger new peer to load snapshot...");
+      impl.triggerSnapshotLoad(peer);
+
+      // step 6: active new Peer
+      logger.info("[MultiLeaderConsensus] activate new peer...");
+      impl.activePeer(peer);
+
+    } catch (ConsensusGroupAddPeerException e) {
+      logger.error("cannot execute addPeer() for {}", peer, e);
+      return ConsensusGenericResponse.newBuilder()
+          .setSuccess(false)
+          .setException(new ConsensusException(e.getMessage()))
+          .build();
+    }
+
+    return ConsensusGenericResponse.newBuilder().setSuccess(true).build();
   }
 
   @Override
   public ConsensusGenericResponse removePeer(ConsensusGroupId groupId, Peer peer) {
-    return ConsensusGenericResponse.newBuilder().setSuccess(false).build();
+    MultiLeaderServerImpl impl = stateMachineMap.get(groupId);
+    if (impl == null) {
+      return ConsensusGenericResponse.newBuilder()
+          .setException(new ConsensusGroupNotExistException(groupId))
+          .build();
+    }
+    try {
+      impl.notifyPeersToRemoveSyncLogChannel(peer);
+    } catch (ConsensusGroupAddPeerException e) {
+      return ConsensusGenericResponse.newBuilder()
+          .setSuccess(false)
+          .setException(new ConsensusException(e.getMessage()))
+          .build();
+    }
+    return ConsensusGenericResponse.newBuilder().setSuccess(true).build();
   }
 
   @Override
diff --git a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/MultiLeaderServerImpl.java b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/MultiLeaderServerImpl.java
index 1b45b3c..5803b9e 100644
--- a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/MultiLeaderServerImpl.java
+++ b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/MultiLeaderServerImpl.java
@@ -28,14 +28,31 @@
 import org.apache.iotdb.consensus.common.request.IConsensusRequest;
 import org.apache.iotdb.consensus.common.request.IndexedConsensusRequest;
 import org.apache.iotdb.consensus.config.MultiLeaderConfig;
+import org.apache.iotdb.consensus.exception.ConsensusGroupAddPeerException;
 import org.apache.iotdb.consensus.multileader.client.AsyncMultiLeaderServiceClient;
+import org.apache.iotdb.consensus.multileader.client.SyncMultiLeaderServiceClient;
 import org.apache.iotdb.consensus.multileader.logdispatcher.LogDispatcher;
+import org.apache.iotdb.consensus.multileader.snapshot.SnapshotFragmentReader;
+import org.apache.iotdb.consensus.multileader.thrift.TActivatePeerReq;
+import org.apache.iotdb.consensus.multileader.thrift.TActivatePeerRes;
+import org.apache.iotdb.consensus.multileader.thrift.TBuildSyncLogChannelReq;
+import org.apache.iotdb.consensus.multileader.thrift.TBuildSyncLogChannelRes;
+import org.apache.iotdb.consensus.multileader.thrift.TInactivatePeerReq;
+import org.apache.iotdb.consensus.multileader.thrift.TInactivatePeerRes;
+import org.apache.iotdb.consensus.multileader.thrift.TRemoveSyncLogChannelReq;
+import org.apache.iotdb.consensus.multileader.thrift.TRemoveSyncLogChannelRes;
+import org.apache.iotdb.consensus.multileader.thrift.TSendSnapshotFragmentReq;
+import org.apache.iotdb.consensus.multileader.thrift.TSendSnapshotFragmentRes;
+import org.apache.iotdb.consensus.multileader.thrift.TTriggerSnapshotLoadReq;
+import org.apache.iotdb.consensus.multileader.thrift.TTriggerSnapshotLoadRes;
 import org.apache.iotdb.consensus.multileader.wal.ConsensusReqReader;
 import org.apache.iotdb.consensus.multileader.wal.GetConsensusReqReaderPlan;
 import org.apache.iotdb.rpc.RpcUtils;
 import org.apache.iotdb.rpc.TSStatusCode;
 import org.apache.iotdb.tsfile.utils.PublicBAOS;
 
+import org.apache.commons.io.FileUtils;
+import org.apache.thrift.TException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -44,7 +61,10 @@
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.file.Files;
+import java.nio.file.Path;
 import java.nio.file.Paths;
+import java.nio.file.StandardOpenOption;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
@@ -56,6 +76,8 @@
 public class MultiLeaderServerImpl {
 
   private static final String CONFIGURATION_FILE_NAME = "configuration.dat";
+  private static final String CONFIGURATION_TMP_FILE_NAME = "configuration.dat.tmp";
+  private static final String SNAPSHOT_DIR_NAME = "snapshot";
 
   private final Logger logger = LoggerFactory.getLogger(MultiLeaderServerImpl.class);
 
@@ -69,6 +91,9 @@
   private final LogDispatcher logDispatcher;
   private final MultiLeaderConfig config;
   private final ConsensusReqReader reader;
+  private boolean active;
+  private String latestSnapshotId;
+  private final IClientManager<TEndPoint, SyncMultiLeaderServiceClient> syncClientManager;
 
   public MultiLeaderServerImpl(
       String storageDir,
@@ -76,10 +101,13 @@
       List<Peer> configuration,
       IStateMachine stateMachine,
       IClientManager<TEndPoint, AsyncMultiLeaderServiceClient> clientManager,
+      IClientManager<TEndPoint, SyncMultiLeaderServiceClient> syncClientManager,
       MultiLeaderConfig config) {
+    this.active = true;
     this.storageDir = storageDir;
     this.thisNode = thisNode;
     this.stateMachine = stateMachine;
+    this.syncClientManager = syncClientManager;
     this.configuration = configuration;
     if (configuration.isEmpty()) {
       recoverConfiguration();
@@ -175,36 +203,302 @@
     return stateMachine.read(request);
   }
 
-  public boolean takeSnapshot(File snapshotDir) {
-    return stateMachine.takeSnapshot(snapshotDir);
+  public void takeSnapshot() throws ConsensusGroupAddPeerException {
+    try {
+      latestSnapshotId =
+          String.format(
+              "%s_%s_%d",
+              SNAPSHOT_DIR_NAME, thisNode.getGroupId().getId(), System.currentTimeMillis());
+      File snapshotDir = new File(storageDir, latestSnapshotId);
+      if (snapshotDir.exists()) {
+        FileUtils.deleteDirectory(snapshotDir);
+      }
+      if (!snapshotDir.mkdirs()) {
+        throw new ConsensusGroupAddPeerException(
+            String.format("%s: cannot mkdir for snapshot", thisNode.getGroupId()));
+      }
+      if (!stateMachine.takeSnapshot(snapshotDir)) {
+        throw new ConsensusGroupAddPeerException("unknown error when taking snapshot");
+      }
+    } catch (IOException e) {
+      throw new ConsensusGroupAddPeerException("error when taking snapshot", e);
+    }
   }
 
-  public void loadSnapshot(File latestSnapshotRootDir) {
-    stateMachine.loadSnapshot(latestSnapshotRootDir);
+  public void transitSnapshot(Peer targetPeer) throws ConsensusGroupAddPeerException {
+    File snapshotDir = new File(storageDir, latestSnapshotId);
+    List<Path> snapshotPaths = stateMachine.getSnapshotFiles(snapshotDir);
+    System.out.println(snapshotPaths);
+    try (SyncMultiLeaderServiceClient client =
+        syncClientManager.borrowClient(targetPeer.getEndpoint())) {
+      for (Path path : snapshotPaths) {
+        SnapshotFragmentReader reader = new SnapshotFragmentReader(latestSnapshotId, path);
+        while (reader.hasNext()) {
+          TSendSnapshotFragmentReq req = reader.next().toTSendSnapshotFragmentReq();
+          req.setConsensusGroupId(targetPeer.getGroupId().convertToTConsensusGroupId());
+          //          receiveSnapshotFragment(latestSnapshotId, req.filePath, req.fileChunk);
+          TSendSnapshotFragmentRes res = client.sendSnapshotFragment(req);
+          if (!isSuccess(res.getStatus())) {
+            throw new ConsensusGroupAddPeerException(
+                String.format("error when sending snapshot fragment to %s", targetPeer));
+          }
+        }
+        reader.close();
+      }
+    } catch (IOException | TException e) {
+      throw new ConsensusGroupAddPeerException(
+          String.format("error when send snapshot file to %s", targetPeer), e);
+    }
+  }
+
+  public void receiveSnapshotFragment(
+      String snapshotId, String originalFilePath, ByteBuffer fileChunk)
+      throws ConsensusGroupAddPeerException {
+    try {
+      String targetFilePath = calculateSnapshotPath(snapshotId, originalFilePath);
+      File targetFile = new File(storageDir, targetFilePath);
+      Path parentDir = Paths.get(targetFile.getParent());
+      if (!Files.exists(parentDir)) {
+        Files.createDirectories(parentDir);
+      }
+      Files.write(
+          Paths.get(targetFile.getAbsolutePath()),
+          fileChunk.array(),
+          StandardOpenOption.CREATE,
+          StandardOpenOption.APPEND);
+    } catch (IOException e) {
+      throw new ConsensusGroupAddPeerException(
+          String.format("error when receiving snapshot %s", snapshotId), e);
+    }
+  }
+
+  private String calculateSnapshotPath(String snapshotId, String originalFilePath)
+      throws ConsensusGroupAddPeerException {
+    if (!originalFilePath.contains(snapshotId)) {
+      throw new ConsensusGroupAddPeerException(
+          String.format(
+              "invalid snapshot file. snapshotId: %s, filePath: %s", snapshotId, originalFilePath));
+    }
+    return originalFilePath.substring(originalFilePath.indexOf(snapshotId));
+  }
+
+  public void loadSnapshot(String snapshotId) {
+    // TODO: (xingtanzjr) throw exception if the snapshot load failed
+    stateMachine.loadSnapshot(new File(storageDir, snapshotId));
+  }
+
+  public void inactivePeer(Peer peer) throws ConsensusGroupAddPeerException {
+    try (SyncMultiLeaderServiceClient client = syncClientManager.borrowClient(peer.getEndpoint())) {
+      TInactivatePeerRes res =
+          client.inactivatePeer(
+              new TInactivatePeerReq(peer.getGroupId().convertToTConsensusGroupId()));
+      if (!isSuccess(res.status)) {
+        throw new ConsensusGroupAddPeerException(
+            String.format("error when inactivating %s. %s", peer, res.getStatus()));
+      }
+    } catch (IOException | TException e) {
+      throw new ConsensusGroupAddPeerException(
+          String.format("error when inactivating %s", peer), e);
+    }
+  }
+
+  public void triggerSnapshotLoad(Peer peer) throws ConsensusGroupAddPeerException {
+    try (SyncMultiLeaderServiceClient client = syncClientManager.borrowClient(peer.getEndpoint())) {
+      TTriggerSnapshotLoadRes res =
+          client.triggerSnapshotLoad(
+              new TTriggerSnapshotLoadReq(
+                  thisNode.getGroupId().convertToTConsensusGroupId(), latestSnapshotId));
+      if (!isSuccess(res.status)) {
+        throw new ConsensusGroupAddPeerException(
+            String.format("error when triggering snapshot load %s. %s", peer, res.getStatus()));
+      }
+    } catch (IOException | TException e) {
+      throw new ConsensusGroupAddPeerException(String.format("error when activating %s", peer), e);
+    }
+  }
+
+  public void activePeer(Peer peer) throws ConsensusGroupAddPeerException {
+    try (SyncMultiLeaderServiceClient client = syncClientManager.borrowClient(peer.getEndpoint())) {
+      TActivatePeerRes res =
+          client.activatePeer(new TActivatePeerReq(peer.getGroupId().convertToTConsensusGroupId()));
+      if (!isSuccess(res.status)) {
+        throw new ConsensusGroupAddPeerException(
+            String.format("error when activating %s. %s", peer, res.getStatus()));
+      }
+    } catch (IOException | TException e) {
+      throw new ConsensusGroupAddPeerException(String.format("error when activating %s", peer), e);
+    }
+  }
+
+  public void notifyPeersToBuildSyncLogChannel(Peer targetPeer)
+      throws ConsensusGroupAddPeerException {
+    // The configuration will be modified during iterating because we will add the targetPeer to
+    // configuration
+    List<Peer> currentMembers = new ArrayList<>(this.configuration);
+    logger.info(
+        "[MultiLeaderConsensus] notify current peers to build sync log. group member: {}, target: {}",
+        currentMembers,
+        targetPeer);
+    for (Peer peer : currentMembers) {
+      logger.info("[MultiLeaderConsensus] build sync log channel from {}", peer);
+      if (peer.equals(thisNode)) {
+        // use searchIndex for thisNode as the initialSyncIndex because targetPeer will load the
+        // snapshot produced by thisNode
+        buildSyncLogChannel(targetPeer, index.get());
+      } else {
+        // use RPC to tell other peers to build sync log channel to target peer
+        try (SyncMultiLeaderServiceClient client =
+            syncClientManager.borrowClient(peer.getEndpoint())) {
+          TBuildSyncLogChannelRes res =
+              client.buildSyncLogChannel(
+                  new TBuildSyncLogChannelReq(
+                      targetPeer.getGroupId().convertToTConsensusGroupId(),
+                      targetPeer.getEndpoint()));
+          if (!isSuccess(res.status)) {
+            throw new ConsensusGroupAddPeerException(
+                String.format("build sync log channel failed from %s to %s", peer, targetPeer));
+          }
+        } catch (IOException | TException e) {
+          // We use a simple way to deal with the connection issue when notifying other nodes to
+          // build sync log. If the un-responsible peer is the peer which will be removed, we cannot
+          // suspend the operation and need to skip it. In order to keep the mechanism works fine,
+          // we will skip the peer which cannot be reached.
+          // If following error message appears, the un-responsible peer should be removed manually
+          // after current operation
+          // TODO: (xingtanzjr) design more reliable way for MultiLeaderConsensus
+          logger.error(
+              "cannot notify {} to build sync log channel. Please check the status of this node manually",
+              peer,
+              e);
+        }
+      }
+    }
+  }
+
+  public void notifyPeersToRemoveSyncLogChannel(Peer targetPeer)
+      throws ConsensusGroupAddPeerException {
+    // The configuration will be modified during iterating because we will add the targetPeer to
+    // configuration
+    List<Peer> currentMembers = new ArrayList<>(this.configuration);
+    for (Peer peer : currentMembers) {
+      if (peer.equals(targetPeer)) {
+        // if the targetPeer is the same as current peer, skip it because removing itself is illegal
+        continue;
+      }
+      if (peer.equals(thisNode)) {
+        removeSyncLogChannel(targetPeer);
+      } else {
+        // use RPC to tell other peers to build sync log channel to target peer
+        try (SyncMultiLeaderServiceClient client =
+            syncClientManager.borrowClient(peer.getEndpoint())) {
+          TRemoveSyncLogChannelRes res =
+              client.removeSyncLogChannel(
+                  new TRemoveSyncLogChannelReq(
+                      targetPeer.getGroupId().convertToTConsensusGroupId(),
+                      targetPeer.getEndpoint()));
+          if (!isSuccess(res.status)) {
+            throw new ConsensusGroupAddPeerException(
+                String.format("remove sync log channel failed from %s to %s", peer, targetPeer));
+          }
+        } catch (IOException | TException e) {
+          throw new ConsensusGroupAddPeerException(
+              String.format("error when removing sync log channel to %s", peer), e);
+        }
+      }
+    }
+  }
+
+  private boolean isSuccess(TSStatus status) {
+    return status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode();
+  }
+
+  /** build SyncLog channel with safeIndex as the default initial sync index */
+  public void buildSyncLogChannel(Peer targetPeer) throws ConsensusGroupAddPeerException {
+    buildSyncLogChannel(targetPeer, getCurrentSafelyDeletedSearchIndex());
+  }
+
+  public void buildSyncLogChannel(Peer targetPeer, long initialSyncIndex)
+      throws ConsensusGroupAddPeerException {
+    // step 1, build sync channel in LogDispatcher
+    logger.info(
+        "[MultiLeaderConsensus] build sync log channel to {} with initialSyncIndex {}",
+        targetPeer,
+        initialSyncIndex);
+    logDispatcher.addLogDispatcherThread(targetPeer, initialSyncIndex);
+    // step 2, update configuration
+    configuration.add(targetPeer);
+    logger.info("[MultiLeaderConsensus] persist new configuration: {}", configuration);
+    persistConfigurationUpdate();
+  }
+
+  public void removeSyncLogChannel(Peer targetPeer) throws ConsensusGroupAddPeerException {
+    try {
+      logDispatcher.removeLogDispatcherThread(targetPeer);
+      logger.info("[MultiLeaderConsensus] log dispatcher to {} removed and cleanup", targetPeer);
+      configuration.remove(targetPeer);
+      persistConfigurationUpdate();
+      logger.info("[MultiLeaderConsensus] configuration updated to {}", this.configuration);
+    } catch (IOException e) {
+      throw new ConsensusGroupAddPeerException("error when remove LogDispatcherThread", e);
+    }
   }
 
   public void persistConfiguration() {
     try (PublicBAOS publicBAOS = new PublicBAOS();
         DataOutputStream outputStream = new DataOutputStream(publicBAOS)) {
-      outputStream.writeInt(configuration.size());
-      for (Peer peer : configuration) {
-        peer.serialize(outputStream);
-      }
+      serializeConfigurationTo(outputStream);
       Files.write(
           Paths.get(new File(storageDir, CONFIGURATION_FILE_NAME).getAbsolutePath()),
           publicBAOS.getBuf());
     } catch (IOException e) {
+      // TODO: (xingtanzjr) need to handle the IOException because the MultiLeaderConsensus won't
+      // work expectedly
+      //  if the exception occurs
       logger.error("Unexpected error occurs when persisting configuration", e);
     }
   }
 
+  public void persistConfigurationUpdate() throws ConsensusGroupAddPeerException {
+    try (PublicBAOS publicBAOS = new PublicBAOS();
+        DataOutputStream outputStream = new DataOutputStream(publicBAOS)) {
+      serializeConfigurationTo(outputStream);
+      Path tmpConfigurationPath =
+          Paths.get(new File(storageDir, CONFIGURATION_TMP_FILE_NAME).getAbsolutePath());
+      Path configurationPath =
+          Paths.get(new File(storageDir, CONFIGURATION_FILE_NAME).getAbsolutePath());
+      Files.write(tmpConfigurationPath, publicBAOS.getBuf());
+      Files.delete(configurationPath);
+      Files.move(tmpConfigurationPath, configurationPath);
+    } catch (IOException e) {
+      throw new ConsensusGroupAddPeerException(
+          "Unexpected error occurs when update configuration", e);
+    }
+  }
+
+  private void serializeConfigurationTo(DataOutputStream outputStream) throws IOException {
+    outputStream.writeInt(configuration.size());
+    for (Peer peer : configuration) {
+      peer.serialize(outputStream);
+    }
+  }
+
   public void recoverConfiguration() {
     ByteBuffer buffer;
     try {
-      buffer =
-          ByteBuffer.wrap(
-              Files.readAllBytes(
-                  Paths.get(new File(storageDir, CONFIGURATION_FILE_NAME).getAbsolutePath())));
+      Path tmpConfigurationPath =
+          Paths.get(new File(storageDir, CONFIGURATION_TMP_FILE_NAME).getAbsolutePath());
+      Path configurationPath =
+          Paths.get(new File(storageDir, CONFIGURATION_FILE_NAME).getAbsolutePath());
+      // If the tmpConfigurationPath exists, it means the `persistConfigurationUpdate` is
+      // interrupted
+      // unexpectedly, we need substitute configuration with tmpConfiguration file
+      if (Files.exists(tmpConfigurationPath)) {
+        if (Files.exists(configurationPath)) {
+          Files.delete(configurationPath);
+        }
+        Files.move(tmpConfigurationPath, configurationPath);
+      }
+      buffer = ByteBuffer.wrap(Files.readAllBytes(configurationPath));
       int size = buffer.getInt();
       for (int i = 0; i < size; i++) {
         configuration.add(Peer.deserialize(buffer));
@@ -278,4 +572,13 @@
   public boolean isReadOnly() {
     return stateMachine.isReadOnly();
   }
+
+  public boolean isActive() {
+    return active;
+  }
+
+  public void setActive(boolean active) {
+    logger.info("set {} active status to {}", this.thisNode, active);
+    this.active = active;
+  }
 }
diff --git a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/client/DispatchLogHandler.java b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/client/DispatchLogHandler.java
index b4cefed..958ccd1 100644
--- a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/client/DispatchLogHandler.java
+++ b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/client/DispatchLogHandler.java
@@ -45,9 +45,7 @@
 
   @Override
   public void onComplete(TSyncLogRes response) {
-    if (response.getStatus().size() == 1
-        && response.getStatus().get(0).getCode()
-            == TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode()) {
+    if (response.getStatus().size() == 1 && needRetry(response.getStatus().get(0).getCode())) {
       logger.warn(
           "Can not send {} to peer {} for {} times because {}",
           batch,
@@ -62,6 +60,12 @@
     }
   }
 
+  private boolean needRetry(int statusCode) {
+    return statusCode == TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode()
+        || statusCode == TSStatusCode.READ_ONLY_SYSTEM_ERROR.getStatusCode()
+        || statusCode == TSStatusCode.WRITE_PROCESS_REJECT.getStatusCode();
+  }
+
   @Override
   public void onError(Exception exception) {
     logger.warn(
diff --git a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/client/MultiLeaderConsensusClientPool.java b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/client/MultiLeaderConsensusClientPool.java
index 44d1002..8e2d23a 100644
--- a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/client/MultiLeaderConsensusClientPool.java
+++ b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/client/MultiLeaderConsensusClientPool.java
@@ -33,6 +33,30 @@
 
   private MultiLeaderConsensusClientPool() {}
 
+  public static class SyncMultiLeaderServiceClientPoolFactory
+      implements IClientPoolFactory<TEndPoint, SyncMultiLeaderServiceClient> {
+    private final MultiLeaderConfig config;
+
+    public SyncMultiLeaderServiceClientPoolFactory(MultiLeaderConfig config) {
+      this.config = config;
+    }
+
+    @Override
+    public KeyedObjectPool<TEndPoint, SyncMultiLeaderServiceClient> createClientPool(
+        ClientManager<TEndPoint, SyncMultiLeaderServiceClient> manager) {
+      return new GenericKeyedObjectPool<>(
+          new SyncMultiLeaderServiceClient.Factory(
+              manager,
+              new ClientFactoryProperty.Builder()
+                  .setConnectionTimeoutMs(config.getRpc().getConnectionTimeoutInMs())
+                  .setRpcThriftCompressionEnabled(config.getRpc().isRpcThriftCompressionEnabled())
+                  .setSelectorNumOfAsyncClientManager(
+                      config.getRpc().getSelectorNumOfClientManager())
+                  .build()),
+          new ClientPoolProperty.Builder<SyncMultiLeaderServiceClient>().build().getConfig());
+    }
+  }
+
   public static class AsyncMultiLeaderServiceClientPoolFactory
       implements IClientPoolFactory<TEndPoint, AsyncMultiLeaderServiceClient> {
 
diff --git a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/client/SyncMultiLeaderServiceClient.java b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/client/SyncMultiLeaderServiceClient.java
new file mode 100644
index 0000000..3cbb3ae
--- /dev/null
+++ b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/client/SyncMultiLeaderServiceClient.java
@@ -0,0 +1,144 @@
+/*
+ * 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.multileader.client;
+
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.commons.client.BaseClientFactory;
+import org.apache.iotdb.commons.client.ClientFactoryProperty;
+import org.apache.iotdb.commons.client.ClientManager;
+import org.apache.iotdb.commons.client.sync.SyncThriftClient;
+import org.apache.iotdb.commons.client.sync.SyncThriftClientWithErrorHandler;
+import org.apache.iotdb.commons.utils.TestOnly;
+import org.apache.iotdb.consensus.multileader.thrift.MultiLeaderConsensusIService;
+import org.apache.iotdb.rpc.RpcTransportFactory;
+import org.apache.iotdb.rpc.TConfigurationConst;
+import org.apache.iotdb.rpc.TimeoutChangeableTransport;
+
+import org.apache.commons.pool2.PooledObject;
+import org.apache.commons.pool2.impl.DefaultPooledObject;
+import org.apache.thrift.protocol.TProtocolFactory;
+import org.apache.thrift.transport.TSocket;
+import org.apache.thrift.transport.TTransportException;
+
+import java.lang.reflect.Constructor;
+import java.net.SocketException;
+
+public class SyncMultiLeaderServiceClient extends MultiLeaderConsensusIService.Client
+    implements SyncThriftClient, AutoCloseable {
+
+  private final TEndPoint endPoint;
+  private final ClientManager<TEndPoint, SyncMultiLeaderServiceClient> clientManager;
+
+  public SyncMultiLeaderServiceClient(
+      TProtocolFactory protocolFactory,
+      int connectionTimeout,
+      TEndPoint endPoint,
+      ClientManager<TEndPoint, SyncMultiLeaderServiceClient> clientManager)
+      throws TTransportException {
+    super(
+        protocolFactory.getProtocol(
+            RpcTransportFactory.INSTANCE.getTransport(
+                new TSocket(
+                    TConfigurationConst.defaultTConfiguration,
+                    endPoint.getIp(),
+                    endPoint.getPort(),
+                    connectionTimeout))));
+    this.endPoint = endPoint;
+    this.clientManager = clientManager;
+    getInputProtocol().getTransport().open();
+  }
+
+  @TestOnly
+  public TEndPoint getTEndpoint() {
+    return endPoint;
+  }
+
+  @TestOnly
+  public ClientManager<TEndPoint, SyncMultiLeaderServiceClient> getClientManager() {
+    return clientManager;
+  }
+
+  public void close() {
+    if (clientManager != null) {
+      clientManager.returnClient(endPoint, this);
+    }
+  }
+
+  public void setTimeout(int timeout) {
+    // the same transport is used in both input and output
+    ((TimeoutChangeableTransport) (getInputProtocol().getTransport())).setTimeout(timeout);
+  }
+
+  public void invalidate() {
+    getInputProtocol().getTransport().close();
+  }
+
+  @Override
+  public void invalidateAll() {
+    clientManager.clear(endPoint);
+  }
+
+  public int getTimeout() throws SocketException {
+    return ((TimeoutChangeableTransport) getInputProtocol().getTransport()).getTimeOut();
+  }
+
+  @Override
+  public String toString() {
+    return String.format("SyncMultiLeaderServiceClient{%s}", endPoint);
+  }
+
+  public static class Factory extends BaseClientFactory<TEndPoint, SyncMultiLeaderServiceClient> {
+
+    public Factory(
+        ClientManager<TEndPoint, SyncMultiLeaderServiceClient> clientManager,
+        ClientFactoryProperty clientFactoryProperty) {
+      super(clientManager, clientFactoryProperty);
+    }
+
+    @Override
+    public void destroyObject(
+        TEndPoint endpoint, PooledObject<SyncMultiLeaderServiceClient> pooledObject) {
+      pooledObject.getObject().invalidate();
+    }
+
+    @Override
+    public PooledObject<SyncMultiLeaderServiceClient> makeObject(TEndPoint endpoint)
+        throws Exception {
+      Constructor<SyncMultiLeaderServiceClient> constructor =
+          SyncMultiLeaderServiceClient.class.getConstructor(
+              TProtocolFactory.class, int.class, endpoint.getClass(), clientManager.getClass());
+      return new DefaultPooledObject<>(
+          SyncThriftClientWithErrorHandler.newErrorHandler(
+              SyncMultiLeaderServiceClient.class,
+              constructor,
+              clientFactoryProperty.getProtocolFactory(),
+              clientFactoryProperty.getConnectionTimeoutMs(),
+              endpoint,
+              clientManager));
+    }
+
+    @Override
+    public boolean validateObject(
+        TEndPoint endpoint, PooledObject<SyncMultiLeaderServiceClient> pooledObject) {
+      return pooledObject.getObject() != null
+          && pooledObject.getObject().getInputProtocol().getTransport().isOpen();
+    }
+  }
+}
diff --git a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/logdispatcher/IndexController.java b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/logdispatcher/IndexController.java
index 027bf6a..f43f0a4 100644
--- a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/logdispatcher/IndexController.java
+++ b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/logdispatcher/IndexController.java
@@ -45,13 +45,15 @@
 
   private final String storageDir;
   private final String prefix;
+  private final long initialIndex;
 
   private final long checkpointGap;
 
-  public IndexController(String storageDir, String prefix, long checkpointGap) {
+  public IndexController(String storageDir, String prefix, long initialIndex, long checkpointGap) {
     this.storageDir = storageDir;
     this.prefix = prefix + '-';
     this.checkpointGap = checkpointGap;
+    this.initialIndex = initialIndex;
     restore();
   }
 
@@ -100,11 +102,17 @@
     try {
       if (oldFile.exists()) {
         FileUtils.moveFile(oldFile, newFile);
+        logger.info(
+            "version file updated, previous: {}, current: {}",
+            oldFile.getAbsolutePath(),
+            newFile.getAbsolutePath());
+      } else {
+        // In the normal state, this branch should not be triggered.
+        logger.error(
+            "failed to flush sync index. cannot find previous version file. previous: {}",
+            lastFlushedIndex);
       }
-      logger.info(
-          "Version file updated, previous: {}, current: {}",
-          oldFile.getAbsolutePath(),
-          newFile.getAbsolutePath());
+
       lastFlushedIndex = flushIndex;
     } catch (IOException e) {
       logger.error("Error occurred when flushing next version", e);
@@ -138,12 +146,26 @@
       }
       currentIndex = lastFlushedIndex;
     } else {
-      versionFile = new File(directory, prefix + "0");
+      currentIndex = initialIndex;
+      versionFile = new File(directory, prefix + initialIndex);
       try {
         Files.createFile(versionFile.toPath());
+        lastFlushedIndex = initialIndex;
       } catch (IOException e) {
+        // TODO: (xingtanzjr) we need to handle the situation that file creation failed.
+        //  Or the dispatcher won't run correctly
         logger.error("Error occurred when creating new file {}", versionFile.getAbsolutePath(), e);
       }
     }
   }
+
+  public void cleanupVersionFiles() throws IOException {
+    File directory = new File(storageDir);
+    File[] versionFiles = directory.listFiles((dir, name) -> name.startsWith(prefix));
+    if (versionFiles != null && versionFiles.length > 0) {
+      for (File versionFile : versionFiles) {
+        Files.delete(versionFile.toPath());
+      }
+    }
+  }
 }
diff --git a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/logdispatcher/LogDispatcher.java b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/logdispatcher/LogDispatcher.java
index 0574d12..ccf11bf 100644
--- a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/logdispatcher/LogDispatcher.java
+++ b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/logdispatcher/LogDispatcher.java
@@ -55,15 +55,16 @@
 
 /** Manage all asynchronous replication threads and corresponding async clients */
 public class LogDispatcher {
-
   private final Logger logger = LoggerFactory.getLogger(LogDispatcher.class);
-
+  private static final long DEFAULT_INITIAL_SYNC_INDEX = 0L;
   private final MultiLeaderServerImpl impl;
   private final List<LogDispatcherThread> threads;
   private final String selfPeerId;
   private final IClientManager<TEndPoint, AsyncMultiLeaderServiceClient> clientManager;
   private ExecutorService executorService;
 
+  private boolean stopped = false;
+
   public LogDispatcher(
       MultiLeaderServerImpl impl,
       IClientManager<TEndPoint, AsyncMultiLeaderServiceClient> clientManager) {
@@ -73,22 +74,26 @@
     this.threads =
         impl.getConfiguration().stream()
             .filter(x -> !Objects.equals(x, impl.getThisNode()))
-            .map(x -> new LogDispatcherThread(x, impl.getConfig()))
+            .map(x -> new LogDispatcherThread(x, impl.getConfig(), DEFAULT_INITIAL_SYNC_INDEX))
             .collect(Collectors.toList());
     if (!threads.isEmpty()) {
+      // We use cached thread pool here because each LogDispatcherThread will occupy one thread.
+      // And every LogDispatcherThread won't release its thread in this pool because it won't stop
+      // unless LogDispatcher stop.
+      // Thus, the size of this threadPool will be the same as the count of LogDispatcherThread.
       this.executorService =
-          IoTDBThreadPoolFactory.newFixedThreadPool(
-              threads.size(), "LogDispatcher-" + impl.getThisNode().getGroupId());
+          IoTDBThreadPoolFactory.newCachedThreadPool(
+              "LogDispatcher-" + impl.getThisNode().getGroupId());
     }
   }
 
-  public void start() {
+  public synchronized void start() {
     if (!threads.isEmpty()) {
       threads.forEach(executorService::submit);
     }
   }
 
-  public void stop() {
+  public synchronized void stop() {
     if (!threads.isEmpty()) {
       threads.forEach(LogDispatcherThread::stop);
       executorService.shutdownNow();
@@ -102,31 +107,65 @@
         logger.error("Unexpected Interruption when closing LogDispatcher service ");
       }
     }
+    stopped = true;
   }
 
-  public OptionalLong getMinSyncIndex() {
+  public synchronized void addLogDispatcherThread(Peer peer, long initialSyncIndex) {
+    if (stopped) {
+      return;
+    }
+    //
+    LogDispatcherThread thread = new LogDispatcherThread(peer, impl.getConfig(), initialSyncIndex);
+    threads.add(thread);
+    executorService.submit(thread);
+  }
+
+  public synchronized void removeLogDispatcherThread(Peer peer) throws IOException {
+    if (stopped) {
+      return;
+    }
+    int threadIndex = -1;
+    for (int i = 0; i < threads.size(); i++) {
+      if (threads.get(i).peer.equals(peer)) {
+        threadIndex = i;
+        break;
+      }
+    }
+    if (threadIndex == -1) {
+      return;
+    }
+    threads.get(threadIndex).stop();
+    threads.get(threadIndex).cleanup();
+    threads.remove(threadIndex);
+  }
+
+  public synchronized OptionalLong getMinSyncIndex() {
     return threads.stream().mapToLong(LogDispatcherThread::getCurrentSyncIndex).min();
   }
 
   public void offer(IndexedConsensusRequest request) {
     List<ByteBuffer> serializedRequests = request.buildSerializedRequests();
-    threads.forEach(
-        thread -> {
-          logger.debug(
-              "{}->{}: Push a log to the queue, where the queue length is {}",
-              impl.getThisNode().getGroupId(),
-              thread.getPeer().getEndpoint().getIp(),
-              thread.getPendingRequest().size());
-          if (!thread
-              .getPendingRequest()
-              .offer(new IndexedConsensusRequest(serializedRequests, request.getSearchIndex()))) {
+    // we put the serialization step outside the synchronized block because it is stateless and
+    // time-consuming
+    synchronized (this) {
+      threads.forEach(
+          thread -> {
             logger.debug(
-                "{}: Log queue of {} is full, ignore the log to this node, searchIndex: {}",
+                "{}->{}: Push a log to the queue, where the queue length is {}",
                 impl.getThisNode().getGroupId(),
-                thread.getPeer(),
-                request.getSearchIndex());
-          }
-        });
+                thread.getPeer().getEndpoint().getIp(),
+                thread.getPendingRequest().size());
+            if (!thread
+                .getPendingRequest()
+                .offer(new IndexedConsensusRequest(serializedRequests, request.getSearchIndex()))) {
+              logger.debug(
+                  "{}: Log queue of {} is full, ignore the log to this node, searchIndex: {}",
+                  impl.getThisNode().getGroupId(),
+                  thread.getPeer(),
+                  request.getSearchIndex());
+            }
+          });
+    }
   }
 
   public class LogDispatcherThread implements Runnable {
@@ -148,7 +187,7 @@
 
     private ConsensusReqReader.ReqIterator walEntryiterator;
 
-    public LogDispatcherThread(Peer peer, MultiLeaderConfig config) {
+    public LogDispatcherThread(Peer peer, MultiLeaderConfig config, long initialSyncIndex) {
       this.peer = peer;
       this.config = config;
       this.pendingRequest =
@@ -157,6 +196,7 @@
           new IndexController(
               impl.getStorageDir(),
               Utils.fromTEndPointToString(peer.getEndpoint()),
+              initialSyncIndex,
               config.getReplication().getCheckpointGap());
       this.syncStatus = new SyncStatus(controller, config);
       this.walEntryiterator = reader.getReqIterator(START_INDEX);
@@ -186,6 +226,10 @@
       stopped = true;
     }
 
+    public void cleanup() throws IOException {
+      this.controller.cleanupVersionFiles();
+    }
+
     public boolean isStopped() {
       return stopped;
     }
diff --git a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/service/MultiLeaderRPCServiceProcessor.java b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/service/MultiLeaderRPCServiceProcessor.java
index 86c3b0f..2544451 100644
--- a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/service/MultiLeaderRPCServiceProcessor.java
+++ b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/service/MultiLeaderRPCServiceProcessor.java
@@ -21,19 +21,33 @@
 
 import org.apache.iotdb.common.rpc.thrift.TSStatus;
 import org.apache.iotdb.commons.consensus.ConsensusGroupId;
-import org.apache.iotdb.commons.exception.IoTDBException;
+import org.apache.iotdb.consensus.common.Peer;
 import org.apache.iotdb.consensus.common.request.BatchIndexedConsensusRequest;
 import org.apache.iotdb.consensus.common.request.ByteBufferConsensusRequest;
 import org.apache.iotdb.consensus.common.request.IConsensusRequest;
 import org.apache.iotdb.consensus.common.request.MultiLeaderConsensusRequest;
+import org.apache.iotdb.consensus.exception.ConsensusGroupAddPeerException;
 import org.apache.iotdb.consensus.multileader.MultiLeaderConsensus;
 import org.apache.iotdb.consensus.multileader.MultiLeaderServerImpl;
 import org.apache.iotdb.consensus.multileader.thrift.MultiLeaderConsensusIService;
+import org.apache.iotdb.consensus.multileader.thrift.TActivatePeerReq;
+import org.apache.iotdb.consensus.multileader.thrift.TActivatePeerRes;
+import org.apache.iotdb.consensus.multileader.thrift.TBuildSyncLogChannelReq;
+import org.apache.iotdb.consensus.multileader.thrift.TBuildSyncLogChannelRes;
+import org.apache.iotdb.consensus.multileader.thrift.TInactivatePeerReq;
+import org.apache.iotdb.consensus.multileader.thrift.TInactivatePeerRes;
 import org.apache.iotdb.consensus.multileader.thrift.TLogBatch;
+import org.apache.iotdb.consensus.multileader.thrift.TRemoveSyncLogChannelReq;
+import org.apache.iotdb.consensus.multileader.thrift.TRemoveSyncLogChannelRes;
+import org.apache.iotdb.consensus.multileader.thrift.TSendSnapshotFragmentReq;
+import org.apache.iotdb.consensus.multileader.thrift.TSendSnapshotFragmentRes;
 import org.apache.iotdb.consensus.multileader.thrift.TSyncLogReq;
 import org.apache.iotdb.consensus.multileader.thrift.TSyncLogRes;
+import org.apache.iotdb.consensus.multileader.thrift.TTriggerSnapshotLoadReq;
+import org.apache.iotdb.consensus.multileader.thrift.TTriggerSnapshotLoadRes;
 import org.apache.iotdb.rpc.TSStatusCode;
 
+import org.apache.thrift.TException;
 import org.apache.thrift.async.AsyncMethodCallback;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -61,7 +75,7 @@
       if (impl == null) {
         String message =
             String.format(
-                "Unexpected consensusGroupId %s for TSyncLogReq which size is %s",
+                "unexpected consensusGroupId %s for TSyncLogReq which size is %s",
                 groupId, req.getBatches().size());
         logger.error(message);
         TSStatus status = new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode());
@@ -70,10 +84,17 @@
         return;
       }
       if (impl.isReadOnly()) {
-        String message = "Fail to sync log because system is read-only.";
+        String message = "fail to sync log because system is read-only.";
         logger.error(message);
-        resultHandler.onError(
-            new IoTDBException(message, TSStatusCode.READ_ONLY_SYSTEM_ERROR.getStatusCode()));
+        TSStatus status = new TSStatus(TSStatusCode.READ_ONLY_SYSTEM_ERROR.getStatusCode());
+        status.setMessage(message);
+        resultHandler.onComplete(new TSyncLogRes(Collections.singletonList(status)));
+        return;
+      }
+      if (!impl.isActive()) {
+        TSStatus status = new TSStatus(TSStatusCode.WRITE_PROCESS_REJECT.getStatusCode());
+        status.setMessage("peer is inactive and not ready to receive sync log request");
+        resultHandler.onComplete(new TSyncLogRes(Collections.singletonList(status)));
         return;
       }
       BatchIndexedConsensusRequest requestsInThisBatch =
@@ -106,12 +127,155 @@
       }
       TSStatus writeStatus = impl.getStateMachine().write(requestsInThisBatch);
       logger.debug(
-          "Execute TSyncLogReq for {} with result {}", req.consensusGroupId, writeStatus.subStatus);
+          "execute TSyncLogReq for {} with result {}", req.consensusGroupId, writeStatus.subStatus);
       resultHandler.onComplete(new TSyncLogRes(writeStatus.subStatus));
     } catch (Exception e) {
       resultHandler.onError(e);
     }
   }
 
+  @Override
+  public void inactivatePeer(
+      TInactivatePeerReq req, AsyncMethodCallback<TInactivatePeerRes> resultHandler)
+      throws TException {
+    ConsensusGroupId groupId =
+        ConsensusGroupId.Factory.createFromTConsensusGroupId(req.getConsensusGroupId());
+    MultiLeaderServerImpl impl = consensus.getImpl(groupId);
+    if (impl == null) {
+      String message =
+          String.format("unexpected consensusGroupId %s for inactivatePeer request", groupId);
+      logger.error(message);
+      TSStatus status = new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode());
+      status.setMessage(message);
+      resultHandler.onComplete(new TInactivatePeerRes(status));
+      return;
+    }
+    impl.setActive(false);
+    resultHandler.onComplete(
+        new TInactivatePeerRes(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())));
+  }
+
+  @Override
+  public void activatePeer(
+      TActivatePeerReq req, AsyncMethodCallback<TActivatePeerRes> resultHandler) throws TException {
+    ConsensusGroupId groupId =
+        ConsensusGroupId.Factory.createFromTConsensusGroupId(req.getConsensusGroupId());
+    MultiLeaderServerImpl impl = consensus.getImpl(groupId);
+    if (impl == null) {
+      String message =
+          String.format("unexpected consensusGroupId %s for inactivatePeer request", groupId);
+      logger.error(message);
+      TSStatus status = new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode());
+      status.setMessage(message);
+      resultHandler.onComplete(new TActivatePeerRes(status));
+      return;
+    }
+    impl.setActive(true);
+    resultHandler.onComplete(
+        new TActivatePeerRes(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())));
+  }
+
+  @Override
+  public void buildSyncLogChannel(
+      TBuildSyncLogChannelReq req, AsyncMethodCallback<TBuildSyncLogChannelRes> resultHandler)
+      throws TException {
+    ConsensusGroupId groupId =
+        ConsensusGroupId.Factory.createFromTConsensusGroupId(req.getConsensusGroupId());
+    MultiLeaderServerImpl impl = consensus.getImpl(groupId);
+    if (impl == null) {
+      String message =
+          String.format("unexpected consensusGroupId %s for buildSyncLogChannel request", groupId);
+      logger.error(message);
+      TSStatus status = new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode());
+      status.setMessage(message);
+      resultHandler.onComplete(new TBuildSyncLogChannelRes(status));
+      return;
+    }
+    TSStatus responseStatus;
+    try {
+      impl.buildSyncLogChannel(new Peer(groupId, req.endPoint));
+      responseStatus = new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+    } catch (ConsensusGroupAddPeerException e) {
+      responseStatus = new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode());
+      responseStatus.setMessage(e.getMessage());
+    }
+    resultHandler.onComplete(new TBuildSyncLogChannelRes(responseStatus));
+  }
+
+  @Override
+  public void removeSyncLogChannel(
+      TRemoveSyncLogChannelReq req, AsyncMethodCallback<TRemoveSyncLogChannelRes> resultHandler)
+      throws TException {
+    ConsensusGroupId groupId =
+        ConsensusGroupId.Factory.createFromTConsensusGroupId(req.getConsensusGroupId());
+    MultiLeaderServerImpl impl = consensus.getImpl(groupId);
+    if (impl == null) {
+      String message =
+          String.format("unexpected consensusGroupId %s for buildSyncLogChannel request", groupId);
+      logger.error(message);
+      TSStatus status = new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode());
+      status.setMessage(message);
+      resultHandler.onComplete(new TRemoveSyncLogChannelRes(status));
+      return;
+    }
+    TSStatus responseStatus;
+    try {
+      impl.removeSyncLogChannel(new Peer(groupId, req.endPoint));
+      responseStatus = new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+    } catch (ConsensusGroupAddPeerException e) {
+      responseStatus = new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode());
+      responseStatus.setMessage(e.getMessage());
+    }
+    resultHandler.onComplete(new TRemoveSyncLogChannelRes(responseStatus));
+  }
+
+  @Override
+  public void sendSnapshotFragment(
+      TSendSnapshotFragmentReq req, AsyncMethodCallback<TSendSnapshotFragmentRes> resultHandler)
+      throws TException {
+    ConsensusGroupId groupId =
+        ConsensusGroupId.Factory.createFromTConsensusGroupId(req.getConsensusGroupId());
+    MultiLeaderServerImpl impl = consensus.getImpl(groupId);
+    if (impl == null) {
+      String message =
+          String.format("unexpected consensusGroupId %s for buildSyncLogChannel request", groupId);
+      logger.error(message);
+      TSStatus status = new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode());
+      status.setMessage(message);
+      resultHandler.onComplete(new TSendSnapshotFragmentRes(status));
+      return;
+    }
+    TSStatus responseStatus;
+    try {
+      impl.receiveSnapshotFragment(req.snapshotId, req.filePath, req.fileChunk);
+      responseStatus = new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+    } catch (ConsensusGroupAddPeerException e) {
+      responseStatus = new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode());
+      responseStatus.setMessage(e.getMessage());
+    }
+    resultHandler.onComplete(new TSendSnapshotFragmentRes(responseStatus));
+  }
+
+  @Override
+  public void triggerSnapshotLoad(
+      TTriggerSnapshotLoadReq req, AsyncMethodCallback<TTriggerSnapshotLoadRes> resultHandler)
+      throws TException {
+    ConsensusGroupId groupId =
+        ConsensusGroupId.Factory.createFromTConsensusGroupId(req.getConsensusGroupId());
+    MultiLeaderServerImpl impl = consensus.getImpl(groupId);
+    if (impl == null) {
+      String message =
+          String.format("unexpected consensusGroupId %s for buildSyncLogChannel request", groupId);
+      logger.error(message);
+      TSStatus status = new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode());
+      status.setMessage(message);
+      resultHandler.onComplete(new TTriggerSnapshotLoadRes(status));
+      return;
+    }
+    impl.loadSnapshot(req.snapshotId);
+    resultHandler.onComplete(
+        new TTriggerSnapshotLoadRes(new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode())));
+  }
+
   public void handleClientExit() {}
 }
diff --git a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/snapshot/SnapshotFragment.java b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/snapshot/SnapshotFragment.java
new file mode 100644
index 0000000..ea04296
--- /dev/null
+++ b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/snapshot/SnapshotFragment.java
@@ -0,0 +1,81 @@
+/*
+ * 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.multileader.snapshot;
+
+import org.apache.iotdb.consensus.multileader.thrift.TSendSnapshotFragmentReq;
+
+import java.nio.ByteBuffer;
+
+public class SnapshotFragment {
+  private final String snapshotId;
+  private final String filePath;
+  private final long totalSize;
+  private final long startOffset;
+  private final long fragmentSize;
+  private final ByteBuffer fileChunk;
+
+  public SnapshotFragment(
+      String snapshotId,
+      String filePath,
+      long totalSize,
+      long startOffset,
+      long fragmentSize,
+      ByteBuffer fileChunk) {
+    this.snapshotId = snapshotId;
+    this.filePath = filePath;
+    this.totalSize = totalSize;
+    this.startOffset = startOffset;
+    this.fragmentSize = fragmentSize;
+    this.fileChunk = fileChunk;
+  }
+
+  public TSendSnapshotFragmentReq toTSendSnapshotFragmentReq() {
+    TSendSnapshotFragmentReq req = new TSendSnapshotFragmentReq();
+    req.setSnapshotId(snapshotId);
+    req.setFilePath(filePath);
+    req.setChunkLength(fragmentSize);
+    req.setFileChunk(fileChunk);
+    return req;
+  }
+
+  public String getSnapshotId() {
+    return snapshotId;
+  }
+
+  public String getFilePath() {
+    return filePath;
+  }
+
+  public long getTotalSize() {
+    return totalSize;
+  }
+
+  public long getStartOffset() {
+    return startOffset;
+  }
+
+  public long getFragmentSize() {
+    return fragmentSize;
+  }
+
+  public ByteBuffer getFileChunk() {
+    return fileChunk;
+  }
+}
diff --git a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/snapshot/SnapshotFragmentReader.java b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/snapshot/SnapshotFragmentReader.java
new file mode 100644
index 0000000..1509995
--- /dev/null
+++ b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/snapshot/SnapshotFragmentReader.java
@@ -0,0 +1,69 @@
+/*
+ * 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.multileader.snapshot;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.SeekableByteChannel;
+import java.nio.file.Files;
+import java.nio.file.Path;
+
+public class SnapshotFragmentReader {
+
+  private static final int DEFAULT_FILE_FRAGMENT_SIZE = 10 * 1024 * 1024;
+  private final String snapshotId;
+  private final String filePath;
+  private final SeekableByteChannel fileChannel;
+  private final long fileSize;
+  private final ByteBuffer buf;
+  private long totalReadSize;
+  private SnapshotFragment cachedSnapshotFragment;
+
+  public SnapshotFragmentReader(String snapshotId, Path path) throws IOException {
+    this.snapshotId = snapshotId;
+    this.filePath = path.toAbsolutePath().toString();
+    this.fileSize = Files.size(path);
+    this.fileChannel = Files.newByteChannel(path);
+    this.buf = ByteBuffer.allocate(DEFAULT_FILE_FRAGMENT_SIZE);
+  }
+
+  public boolean hasNext() throws IOException {
+    buf.clear();
+    int actualReadSize = fileChannel.read(buf);
+    buf.flip();
+    if (actualReadSize > 0) {
+      cachedSnapshotFragment =
+          new SnapshotFragment(snapshotId, filePath, fileSize, totalReadSize, actualReadSize, buf);
+      totalReadSize += actualReadSize;
+      return true;
+    }
+    return false;
+  }
+
+  public SnapshotFragment next() {
+    return cachedSnapshotFragment;
+  }
+
+  public void close() throws IOException {
+    if (fileChannel != null) {
+      fileChannel.close();
+    }
+  }
+}
diff --git a/consensus/src/test/java/org/apache/iotdb/consensus/multileader/logdispatcher/IndexControllerTest.java b/consensus/src/test/java/org/apache/iotdb/consensus/multileader/logdispatcher/IndexControllerTest.java
index 2675470..761ace4 100644
--- a/consensus/src/test/java/org/apache/iotdb/consensus/multileader/logdispatcher/IndexControllerTest.java
+++ b/consensus/src/test/java/org/apache/iotdb/consensus/multileader/logdispatcher/IndexControllerTest.java
@@ -49,7 +49,7 @@
   @Test
   public void testIncrementIntervalAfterRestart() {
     IndexController controller =
-        new IndexController(storageDir.getAbsolutePath(), prefix, CHECK_POINT_GAP);
+        new IndexController(storageDir.getAbsolutePath(), prefix, 0, CHECK_POINT_GAP);
     Assert.assertEquals(0, controller.getCurrentIndex());
     Assert.assertEquals(0, controller.getLastFlushedIndex());
 
@@ -58,7 +58,7 @@
     Assert.assertEquals(CHECK_POINT_GAP - 1, controller.getCurrentIndex());
     Assert.assertEquals(0, controller.getLastFlushedIndex());
 
-    controller = new IndexController(storageDir.getAbsolutePath(), prefix, CHECK_POINT_GAP);
+    controller = new IndexController(storageDir.getAbsolutePath(), prefix, 0, CHECK_POINT_GAP);
     Assert.assertEquals(0, controller.getCurrentIndex());
     Assert.assertEquals(0, controller.getLastFlushedIndex());
 
@@ -66,7 +66,7 @@
     Assert.assertEquals(CHECK_POINT_GAP + 1, controller.getCurrentIndex());
     Assert.assertEquals(CHECK_POINT_GAP, controller.getLastFlushedIndex());
 
-    controller = new IndexController(storageDir.getAbsolutePath(), prefix, CHECK_POINT_GAP);
+    controller = new IndexController(storageDir.getAbsolutePath(), prefix, 0, CHECK_POINT_GAP);
     Assert.assertEquals(CHECK_POINT_GAP, controller.getCurrentIndex());
     Assert.assertEquals(CHECK_POINT_GAP, controller.getLastFlushedIndex());
 
@@ -74,7 +74,7 @@
     Assert.assertEquals(CHECK_POINT_GAP * 2 - 1, controller.getCurrentIndex());
     Assert.assertEquals(CHECK_POINT_GAP, controller.getLastFlushedIndex());
 
-    controller = new IndexController(storageDir.getAbsolutePath(), prefix, CHECK_POINT_GAP);
+    controller = new IndexController(storageDir.getAbsolutePath(), prefix, 0, CHECK_POINT_GAP);
     Assert.assertEquals(CHECK_POINT_GAP, controller.getCurrentIndex());
     Assert.assertEquals(CHECK_POINT_GAP, controller.getLastFlushedIndex());
 
diff --git a/consensus/src/test/java/org/apache/iotdb/consensus/multileader/logdispatcher/SyncStatusTest.java b/consensus/src/test/java/org/apache/iotdb/consensus/multileader/logdispatcher/SyncStatusTest.java
index f7c9072..a888904 100644
--- a/consensus/src/test/java/org/apache/iotdb/consensus/multileader/logdispatcher/SyncStatusTest.java
+++ b/consensus/src/test/java/org/apache/iotdb/consensus/multileader/logdispatcher/SyncStatusTest.java
@@ -56,7 +56,7 @@
   @Test
   public void sequenceTest() throws InterruptedException {
     IndexController controller =
-        new IndexController(storageDir.getAbsolutePath(), prefix, CHECK_POINT_GAP);
+        new IndexController(storageDir.getAbsolutePath(), prefix, 0, CHECK_POINT_GAP);
     Assert.assertEquals(0, controller.getCurrentIndex());
 
     SyncStatus status = new SyncStatus(controller, config);
@@ -82,7 +82,7 @@
   @Test
   public void reverseTest() throws InterruptedException {
     IndexController controller =
-        new IndexController(storageDir.getAbsolutePath(), prefix, CHECK_POINT_GAP);
+        new IndexController(storageDir.getAbsolutePath(), prefix, 0, CHECK_POINT_GAP);
     Assert.assertEquals(0, controller.getCurrentIndex());
     Assert.assertEquals(0, controller.getLastFlushedIndex());
 
@@ -115,7 +115,7 @@
   @Test
   public void mixedTest() throws InterruptedException {
     IndexController controller =
-        new IndexController(storageDir.getAbsolutePath(), prefix, CHECK_POINT_GAP);
+        new IndexController(storageDir.getAbsolutePath(), prefix, 0, CHECK_POINT_GAP);
     Assert.assertEquals(0, controller.getCurrentIndex());
     Assert.assertEquals(0, controller.getLastFlushedIndex());
 
@@ -160,7 +160,7 @@
   @Test
   public void waitTest() throws InterruptedException, ExecutionException {
     IndexController controller =
-        new IndexController(storageDir.getAbsolutePath(), prefix, CHECK_POINT_GAP);
+        new IndexController(storageDir.getAbsolutePath(), prefix, 0, CHECK_POINT_GAP);
     Assert.assertEquals(0, controller.getCurrentIndex());
 
     SyncStatus status = new SyncStatus(controller, config);
diff --git a/server/src/main/java/org/apache/iotdb/db/consensus/statemachine/DataRegionStateMachine.java b/server/src/main/java/org/apache/iotdb/db/consensus/statemachine/DataRegionStateMachine.java
index 75b18d5..0d90993 100644
--- a/server/src/main/java/org/apache/iotdb/db/consensus/statemachine/DataRegionStateMachine.java
+++ b/server/src/main/java/org/apache/iotdb/db/consensus/statemachine/DataRegionStateMachine.java
@@ -31,6 +31,9 @@
 import org.apache.iotdb.db.conf.IoTDBDescriptor;
 import org.apache.iotdb.db.consensus.statemachine.visitor.DataExecutionVisitor;
 import org.apache.iotdb.db.engine.StorageEngineV2;
+import org.apache.iotdb.db.engine.cache.BloomFilterCache;
+import org.apache.iotdb.db.engine.cache.ChunkCache;
+import org.apache.iotdb.db.engine.cache.TimeSeriesMetadataCache;
 import org.apache.iotdb.db.engine.snapshot.SnapshotLoader;
 import org.apache.iotdb.db.engine.snapshot.SnapshotTaker;
 import org.apache.iotdb.db.engine.storagegroup.DataRegion;
@@ -124,6 +127,9 @@
     try {
       StorageEngineV2.getInstance()
           .setDataRegion(new DataRegionId(Integer.parseInt(region.getDataRegionId())), region);
+      ChunkCache.getInstance().clear();
+      TimeSeriesMetadataCache.getInstance().clear();
+      BloomFilterCache.getInstance().clear();
     } catch (Exception e) {
       logger.error("Exception occurs when replacing data region in storage engine.", e);
     }
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/snapshot/SnapshotLoader.java b/server/src/main/java/org/apache/iotdb/db/engine/snapshot/SnapshotLoader.java
index e3fdf37..97771dc 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/snapshot/SnapshotLoader.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/snapshot/SnapshotLoader.java
@@ -122,6 +122,12 @@
         return null;
       }
       LOGGER.info("Moving snapshot file to data dirs");
+      try {
+        deleteAllFilesInDataDirs();
+        LOGGER.info("Remove all data files in original data dir");
+      } catch (IOException e) {
+        return null;
+      }
       createLinksFromSnapshotDirToDataDirWithoutLog(new File(snapshotPath));
       return loadSnapshot();
     } catch (IOException | DiskSpaceInsufficientException e) {
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/snapshot/SnapshotTaker.java b/server/src/main/java/org/apache/iotdb/db/engine/snapshot/SnapshotTaker.java
index 67b3df9..d90eb08 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/snapshot/SnapshotTaker.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/snapshot/SnapshotTaker.java
@@ -149,6 +149,9 @@
           continue;
         }
         File tsFile = resource.getTsFile();
+        if (!resource.isClosed()) {
+          continue;
+        }
         File snapshotTsFile = getSnapshotFilePathForTsFile(tsFile, snapshotId);
         // create hard link for tsfile, resource, mods
         createHardLink(snapshotTsFile, tsFile);
@@ -169,7 +172,14 @@
   }
 
   private void createHardLink(File target, File source) throws IOException {
-    Files.createLink(target.toPath(), source.toPath());
+    if (!target.getParentFile().exists()) {
+      LOGGER.error("Hard link target dir {} doesn't exist", target.getParentFile());
+    }
+    if (!source.exists()) {
+      LOGGER.error("Hard link source file {} doesn't exist", source);
+    }
+
+    Files.createLink(target.getAbsoluteFile().toPath(), source.getAbsoluteFile().toPath());
     snapshotLogger.logFile(source.getAbsolutePath(), target.getAbsolutePath());
   }
 
@@ -216,6 +226,7 @@
   }
 
   private void cleanUpWhenFail(String snapshotId) {
+    LOGGER.info("Cleaning up snapshot dir for {}", snapshotId);
     for (String dataDir : IoTDBDescriptor.getInstance().getConfig().getDataDirs()) {
       File dataDirForThisSnapshot =
           new File(dataDir + File.separator + "snapshot" + File.separator + snapshotId);
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/TestRPCClient.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/TestRPCClient.java
new file mode 100644
index 0000000..4a5b416
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/TestRPCClient.java
@@ -0,0 +1,176 @@
+/*
+ * 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.db.mpp.plan;
+
+import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.client.IClientManager;
+import org.apache.iotdb.commons.client.sync.SyncDataNodeInternalServiceClient;
+import org.apache.iotdb.commons.consensus.DataRegionId;
+import org.apache.iotdb.consensus.config.MultiLeaderConfig;
+import org.apache.iotdb.consensus.multileader.client.MultiLeaderConsensusClientPool;
+import org.apache.iotdb.consensus.multileader.client.SyncMultiLeaderServiceClient;
+import org.apache.iotdb.consensus.multileader.thrift.TInactivatePeerReq;
+import org.apache.iotdb.consensus.multileader.thrift.TInactivatePeerRes;
+import org.apache.iotdb.consensus.multileader.thrift.TTriggerSnapshotLoadReq;
+import org.apache.iotdb.consensus.multileader.thrift.TTriggerSnapshotLoadRes;
+import org.apache.iotdb.db.client.DataNodeClientPoolFactory;
+import org.apache.iotdb.mpp.rpc.thrift.TCreateDataRegionReq;
+import org.apache.iotdb.mpp.rpc.thrift.TMaintainPeerReq;
+
+import org.apache.thrift.TException;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+public class TestRPCClient {
+  private static final IClientManager<TEndPoint, SyncDataNodeInternalServiceClient>
+      INTERNAL_SERVICE_CLIENT_MANAGER =
+          new IClientManager.Factory<TEndPoint, SyncDataNodeInternalServiceClient>()
+              .createClientManager(
+                  new DataNodeClientPoolFactory.SyncDataNodeInternalServiceClientPoolFactory());
+
+  private final IClientManager<TEndPoint, SyncMultiLeaderServiceClient> syncClientManager;
+
+  public TestRPCClient() {
+    syncClientManager =
+        new IClientManager.Factory<TEndPoint, SyncMultiLeaderServiceClient>()
+            .createClientManager(
+                new MultiLeaderConsensusClientPool.SyncMultiLeaderServiceClientPoolFactory(
+                    new MultiLeaderConfig.Builder().build()));
+  }
+
+  public static void main(String args[]) {
+    TestRPCClient client = new TestRPCClient();
+    //    client.removeRegionPeer();
+    client.addPeer();
+    //    client.loadSnapshot();
+  }
+
+  private void loadSnapshot() {
+    try (SyncMultiLeaderServiceClient client =
+        syncClientManager.borrowClient(new TEndPoint("127.0.0.1", 40011))) {
+      TTriggerSnapshotLoadRes res =
+          client.triggerSnapshotLoad(
+              new TTriggerSnapshotLoadReq(
+                  new DataRegionId(1).convertToTConsensusGroupId(), "snapshot_1_1662370255552"));
+      System.out.println(res.status);
+    } catch (IOException | TException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  private void testAddPeer() {
+    try (SyncMultiLeaderServiceClient client =
+        syncClientManager.borrowClient(new TEndPoint("127.0.0.1", 40012))) {
+      TInactivatePeerRes res =
+          client.inactivatePeer(
+              new TInactivatePeerReq(new DataRegionId(1).convertToTConsensusGroupId()));
+      System.out.println(res.status);
+    } catch (IOException | TException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  private void removeRegionPeer() {
+    try (SyncDataNodeInternalServiceClient client =
+        INTERNAL_SERVICE_CLIENT_MANAGER.borrowClient(new TEndPoint("127.0.0.1", 9003))) {
+      client.removeRegionPeer(
+          new TMaintainPeerReq(new DataRegionId(1).convertToTConsensusGroupId(), getLocation2(3)));
+    } catch (IOException | TException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  private void addPeer() {
+    try (SyncDataNodeInternalServiceClient client =
+        INTERNAL_SERVICE_CLIENT_MANAGER.borrowClient(new TEndPoint("127.0.0.1", 9003))) {
+      client.addRegionPeer(
+          new TMaintainPeerReq(new DataRegionId(1).convertToTConsensusGroupId(), getLocation2(3)));
+    } catch (IOException | TException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  private TDataNodeLocation getLocation3(int dataNodeId) {
+    return new TDataNodeLocation(
+        dataNodeId,
+        new TEndPoint("127.0.0.1", 6669),
+        new TEndPoint("127.0.0.1", 9005),
+        new TEndPoint("127.0.0.1", 8779),
+        new TEndPoint("127.0.0.1", 40012),
+        new TEndPoint("127.0.0.1", 50012));
+  }
+
+  private TDataNodeLocation getLocation2(int dataNodeId) {
+    return new TDataNodeLocation(
+        dataNodeId,
+        new TEndPoint("127.0.0.1", 6668),
+        new TEndPoint("127.0.0.1", 9004),
+        new TEndPoint("127.0.0.1", 8778),
+        new TEndPoint("127.0.0.1", 40011),
+        new TEndPoint("127.0.0.1", 50011));
+  }
+
+  private void createDataRegion() {
+    try (SyncDataNodeInternalServiceClient client =
+        INTERNAL_SERVICE_CLIENT_MANAGER.borrowClient(new TEndPoint("127.0.0.1", 9005))) {
+      TCreateDataRegionReq req = new TCreateDataRegionReq();
+      req.setStorageGroup("root.test.g_0");
+      TRegionReplicaSet regionReplicaSet = new TRegionReplicaSet();
+      regionReplicaSet.setRegionId(new DataRegionId(1).convertToTConsensusGroupId());
+      List<TDataNodeLocation> locationList = new ArrayList<>();
+      locationList.add(
+          new TDataNodeLocation(
+              3,
+              new TEndPoint("127.0.0.1", 6667),
+              new TEndPoint("127.0.0.1", 9003),
+              new TEndPoint("127.0.0.1", 8777),
+              new TEndPoint("127.0.0.1", 40010),
+              new TEndPoint("127.0.0.1", 50010)));
+      locationList.add(
+          new TDataNodeLocation(
+              4,
+              new TEndPoint("127.0.0.1", 6668),
+              new TEndPoint("127.0.0.1", 9004),
+              new TEndPoint("127.0.0.1", 8778),
+              new TEndPoint("127.0.0.1", 40011),
+              new TEndPoint("127.0.0.1", 50011)));
+      locationList.add(
+          new TDataNodeLocation(
+              4,
+              new TEndPoint("127.0.0.1", 6669),
+              new TEndPoint("127.0.0.1", 9005),
+              new TEndPoint("127.0.0.1", 8779),
+              new TEndPoint("127.0.0.1", 40012),
+              new TEndPoint("127.0.0.1", 50012)));
+      regionReplicaSet.setDataNodeLocations(locationList);
+      req.setRegionReplicaSet(regionReplicaSet);
+      TSStatus res = client.createDataRegion(req);
+      System.out.println(res.code + " " + res.message);
+
+    } catch (IOException | TException e) {
+      throw new RuntimeException(e);
+    }
+  }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/SimpleFragmentParallelPlanner.java b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/SimpleFragmentParallelPlanner.java
index e532877..f83730b 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/SimpleFragmentParallelPlanner.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/distribution/SimpleFragmentParallelPlanner.java
@@ -169,7 +169,7 @@
 
   private boolean isAvailableDataNode(TDataNodeLocation dataNodeLocation) {
     for (TEndPoint endPoint : queryContext.getEndPointBlackList()) {
-      if (endPoint.getIp().equals(dataNodeLocation.internalEndPoint.getIp())) {
+      if (endPoint.equals(dataNodeLocation.internalEndPoint)) {
         return false;
       }
     }
diff --git a/thrift-multi-leader-consensus/src/main/thrift/mutlileader.thrift b/thrift-multi-leader-consensus/src/main/thrift/mutlileader.thrift
index 7d315d2..380f30a 100644
--- a/thrift-multi-leader-consensus/src/main/thrift/mutlileader.thrift
+++ b/thrift-multi-leader-consensus/src/main/thrift/mutlileader.thrift
@@ -33,10 +33,71 @@
   3: required list<TLogBatch> batches
 }
 
+struct TInactivatePeerReq {
+  1: required common.TConsensusGroupId consensusGroupId
+}
+
+struct TInactivatePeerRes {
+  1: required common.TSStatus status
+}
+
+struct TActivatePeerReq {
+  1: required common.TConsensusGroupId consensusGroupId
+}
+
+struct TActivatePeerRes {
+  1: required common.TSStatus status
+}
+
 struct TSyncLogRes {
   1: required list<common.TSStatus> status
 }
 
+struct TBuildSyncLogChannelReq {
+  1: required common.TConsensusGroupId consensusGroupId
+  2: required common.TEndPoint endPoint
+}
+
+struct TBuildSyncLogChannelRes {
+  1: required common.TSStatus status
+}
+
+struct TRemoveSyncLogChannelReq {
+  1: required common.TConsensusGroupId consensusGroupId
+  2: required common.TEndPoint endPoint
+}
+
+struct TRemoveSyncLogChannelRes {
+  1: required common.TSStatus status
+}
+
+struct TSendSnapshotFragmentReq {
+  1: required common.TConsensusGroupId consensusGroupId
+  2: required string snapshotId
+  3: required string filePath
+  4: required i64 chunkLength
+  5: required binary fileChunk
+}
+
+struct TSendSnapshotFragmentRes {
+  1: required common.TSStatus status
+}
+
+struct TTriggerSnapshotLoadReq {
+  1: required common.TConsensusGroupId consensusGroupId
+  2: required string snapshotId
+}
+
+struct TTriggerSnapshotLoadRes {
+  1: required common.TSStatus status
+}
+
 service MultiLeaderConsensusIService {
   TSyncLogRes syncLog(TSyncLogReq req)
+  TInactivatePeerRes inactivatePeer(TInactivatePeerReq req)
+  TActivatePeerRes activatePeer(TActivatePeerReq req)
+  TBuildSyncLogChannelRes buildSyncLogChannel(TBuildSyncLogChannelReq req)
+  TRemoveSyncLogChannelRes removeSyncLogChannel(TRemoveSyncLogChannelReq req)
+  TSendSnapshotFragmentRes sendSnapshotFragment(TSendSnapshotFragmentReq req)
+  TTriggerSnapshotLoadRes triggerSnapshotLoad(TTriggerSnapshotLoadReq req)
 }
\ No newline at end of file