[IoTConsensusV2 X Region Migration] Start consensus pipe in proper time rather than automatically start it when create it

Author: Pengzna

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/CreatePipeProcedureV2.java

@@ -203,7 +203,9 @@ public void executeFromCalculateInfoForTask(ConfigNodeProcedureEnv env) {
     }
 
     pipeRuntimeMeta = new PipeRuntimeMeta(consensusGroupIdToTaskMetaMap);
-    pipeRuntimeMeta.getStatus().set(PipeStatus.RUNNING);
+    if (!createPipeRequest.needManuallyStart) {
+      pipeRuntimeMeta.getStatus().set(PipeStatus.RUNNING);
+    }
   }
 
   @Override

iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/PipeConsensus.java

@@ -332,9 +332,8 @@ public void addRemotePeer(ConsensusGroupId groupId, Peer peer) throws ConsensusE
       impl.setRemotePeerActive(peer, false);
 
       // step 2: notify all the other Peers to create consensus pipes to newPeer
-      // NOTE: For this step, coordinator(thisNode) will transfer its full data snapshot to target,
-      // while other peers will only transmit data(may contain both historical and realtime data)
-      // after the snapshot progress to target.
+      // NOTE: For this step, coordinator(thisNode) will transfer its full data snapshot to target
+      // while other peers record the coordinator's progress.
       LOGGER.info("[{}] notify current peers to create consensus pipes...", CLASS_NAME);
       impl.notifyPeersToCreateConsensusPipes(peer, impl.getThisNodePeer());
       KillPoint.setKillPoint(DataNodeKillPoints.COORDINATOR_ADD_PEER_TRANSITION);
@@ -343,7 +342,13 @@ public void addRemotePeer(ConsensusGroupId groupId, Peer peer) throws ConsensusE
       LOGGER.info("[{}] wait until all the other peers finish transferring...", CLASS_NAME);
       impl.waitPeersToTargetPeerTransmissionCompleted(peer);
 
-      // step 4: active new Peer to let new Peer receive snapshot
+      // step 4. start other peers' consensus pipe to target peer to transfer remaining data
+      // NOTE: For this step, other peers will start to transmit data(may contain both historical
+      // and realtime data) after the snapshot progress to target.
+      LOGGER.info("[{}] start transfer remaining data from other peers", CLASS_NAME);
+      impl.startOtherConsensusPipesToTargetPeer(peer);
+
+      // step 5: active new Peer to let new Peer receive client requests
       LOGGER.info("[{}] activate new peer...", CLASS_NAME);
       impl.setRemotePeerActive(peer, true);
       KillPoint.setKillPoint(DataNodeKillPoints.COORDINATOR_ADD_PEER_DONE);

iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/PipeConsensusServerImpl.java

@@ -425,7 +425,7 @@ public void notifyPeersToCreateConsensusPipes(Peer targetPeer, Peer coordinatorP
     try {
       // This node which acts as coordinator will transfer complete historical snapshot to new
       // target.
-      createConsensusPipeToTargetPeer(targetPeer, thisNode);
+      createConsensusPipeToTargetPeer(targetPeer, thisNode, false);
     } catch (Exception e) {
       LOGGER.warn(
           "{} cannot create consensus pipe to {}, may because target peer is unknown currently, please manually check!",
@@ -437,11 +437,11 @@ public void notifyPeersToCreateConsensusPipes(Peer targetPeer, Peer coordinatorP
   }
 
   public synchronized void createConsensusPipeToTargetPeer(
-      Peer targetPeer, Peer regionMigrationCoordinatorPeer)
+      Peer targetPeer, Peer regionMigrationCoordinatorPeer, boolean needManuallyStart)
       throws ConsensusGroupModifyPeerException {
     try {
       consensusPipeManager.createConsensusPipe(
-          thisNode, targetPeer, regionMigrationCoordinatorPeer);
+          thisNode, targetPeer, regionMigrationCoordinatorPeer, needManuallyStart);
       peerManager.addAndPersist(targetPeer);
     } catch (IOException e) {
       LOGGER.warn("{} cannot persist peer {}", thisNode, targetPeer, e);
@@ -510,6 +510,24 @@ public synchronized void dropConsensusPipeToTargetPeer(Peer targetPeer)
     }
   }
 
+  public void startOtherConsensusPipesToTargetPeer(Peer targetPeer)
+      throws ConsensusGroupModifyPeerException {
+    final List<Peer> otherPeers = peerManager.getOtherPeers(thisNode);
+    for (Peer peer : otherPeers) {
+      if (peer.equals(targetPeer)) {
+        continue;
+      }
+      try {
+        consensusPipeManager.updateConsensusPipe(
+            new ConsensusPipeName(peer, targetPeer), PipeStatus.RUNNING);
+      } catch (Exception e) {
+        LOGGER.warn("{} cannot start consensus pipe to {}", peer, targetPeer, e);
+        throw new ConsensusGroupModifyPeerException(
+            String.format("%s cannot start consensus pipe to %s", peer, targetPeer), e);
+      }
+    }
+  }
+
   public void waitPeersToTargetPeerTransmissionCompleted(Peer targetPeer)
       throws ConsensusGroupModifyPeerException {
     boolean isTransmissionCompleted = false;

iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/consensuspipe/ConsensusPipeDispatcher.java

@@ -26,7 +26,8 @@ void createPipe(
       String pipeName,
       Map<String, String> extractorAttributes,
       Map<String, String> processorAttributes,
-      Map<String, String> connectorAttributes)
+      Map<String, String> connectorAttributes,
+      boolean needManuallyStart)
       throws Exception;
 
   void startPipe(String pipeName) throws Exception;

iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/consensuspipe/ConsensusPipeManager.java

@@ -71,17 +71,29 @@ public void createConsensusPipe(Peer senderPeer, Peer receiverPeer) throws Excep
     Triple<ImmutableMap<String, String>, ImmutableMap<String, String>, ImmutableMap<String, String>>
         params = buildPipeParams(senderPeer, receiverPeer, senderPeer);
     dispatcher.createPipe(
-        consensusPipeName.toString(), params.getLeft(), params.getMiddle(), params.getRight());
+        consensusPipeName.toString(),
+        params.getLeft(),
+        params.getMiddle(),
+        params.getRight(),
+        false);
   }
 
   /** This method is used when executing region migration */
   public void createConsensusPipe(
-      Peer senderPeer, Peer receiverPeer, Peer regionMigrationCoordinatorPeer) throws Exception {
+      Peer senderPeer,
+      Peer receiverPeer,
+      Peer regionMigrationCoordinatorPeer,
+      boolean needManuallyStart)
+      throws Exception {
     ConsensusPipeName consensusPipeName = new ConsensusPipeName(senderPeer, receiverPeer);
     Triple<ImmutableMap<String, String>, ImmutableMap<String, String>, ImmutableMap<String, String>>
         params = buildPipeParams(senderPeer, receiverPeer, regionMigrationCoordinatorPeer);
     dispatcher.createPipe(
-        consensusPipeName.toString(), params.getLeft(), params.getMiddle(), params.getRight());
+        consensusPipeName.toString(),
+        params.getLeft(),
+        params.getMiddle(),
+        params.getRight(),
+        needManuallyStart);
   }
 
   public Triple<

iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/service/PipeConsensusRPCServiceProcessor.java

@@ -117,7 +117,8 @@ public TNotifyPeerToCreateConsensusPipeResp notifyPeerToCreateConsensusPipe(
           new Peer(
               ConsensusGroupId.Factory.createFromTConsensusGroupId(req.targetPeerConsensusGroupId),
               req.coordinatorPeerNodeId,
-              req.coordinatorPeerEndPoint));
+              req.coordinatorPeerEndPoint),
+          true);
       responseStatus = new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
     } catch (ConsensusGroupModifyPeerException e) {
       responseStatus = new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode());

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/ConsensusPipeDataNodeDispatcher.java

@@ -49,13 +49,15 @@ public void createPipe(
       String pipeName,
       Map<String, String> extractorAttributes,
       Map<String, String> processorAttributes,
-      Map<String, String> connectorAttributes)
+      Map<String, String> connectorAttributes,
+      boolean needManuallyStart)
       throws Exception {
     try (ConfigNodeClient configNodeClient =
         CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
       TCreatePipeReq req =
           new TCreatePipeReq()
               .setPipeName(pipeName)
+              .setNeedManuallyStart(needManuallyStart)
               .setExtractorAttributes(extractorAttributes)
               .setProcessorAttributes(processorAttributes)
               .setConnectorAttributes(connectorAttributes);

iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift

@@ -737,6 +737,7 @@ struct TCreatePipeReq {
     3: optional map<string, string> processorAttributes
     4: required map<string, string> connectorAttributes
     5: optional bool ifNotExistsCondition
+    6: optional bool needManuallyStart
 }
 
 struct TAlterPipeReq {