private TSStatus handleLeaderChangeInternal(final PipeHandleLeaderChangePlan plan) {
    plan.getConsensusGroupId2NewLeaderIdMap()
        .forEach(
            (consensusGroupId, newLeader) ->
                pipeMetaKeeper
                    .getPipeMetaList()
                    .forEach(
                        pipeMeta -> {
                          if (PipeType.CONSENSUS.equals(pipeMeta.getStaticMeta().getPipeType())) {
                            return; // pipe consensus pipe task will not change
                          }

                          if (pipeMeta.getStaticMeta().isSourceExternal()) {
                            // external source pipe tasks are not balanced here since non-leaders
                            // don't know about RegionLeader Map and will be balanced in the meta
                            // sync procedure
                            LOGGER.info(
                                "Pipe {} is using external source, skip region leader change. PipeHandleLeaderChangePlan: {}",
                                pipeMeta.getStaticMeta().getPipeName(),
                                plan.getConsensusGroupId2NewLeaderIdMap());
                            return;
                          }

                          final Map<Integer, PipeTaskMeta> consensusGroupIdToTaskMetaMap =
                              pipeMeta.getRuntimeMeta().getConsensusGroupId2TaskMetaMap();

                          if (consensusGroupIdToTaskMetaMap.containsKey(consensusGroupId.getId())) {
                            // If the region leader is -1, it means the region is
                            // removed
                            if (newLeader != -1) {
                              consensusGroupIdToTaskMetaMap
                                  .get(consensusGroupId.getId())
                                  .setLeaderNodeId(newLeader);
                              // New region leader may contain un-transferred events
                              ((PipeTemporaryMetaInCoordinator) pipeMeta.getTemporaryMeta())
                                  .markDataNodeUncompleted(newLeader);
                            } else {
                              consensusGroupIdToTaskMetaMap.remove(consensusGroupId.getId());
                            }
                          } else {
                            // If CN does not contain the region group, it means the data
                            // region group is newly added.
                            if (newLeader != -1) {
                              consensusGroupIdToTaskMetaMap.put(
                                  consensusGroupId.getId(),
                                  new PipeTaskMeta(
                                      MinimumProgressIndex.INSTANCE,
                                      newLeader,
                                      consensusGroupId.getId(),
                                      false));
                            }
                            // else:
                            // "The pipe task meta does not contain the data region group {} or
                            // the data region group has already been removed"
                          }
                        }));

    return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
  }
Handle the region leader change event and update the pipe task meta accordingly.
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/pipe/PipeTaskInfo.java