@Override
  public void executeFromCalculateInfoForTask(final ConfigNodeProcedureEnv env) {
    LOGGER.info(
        "CreatePipeProcedureV2: executeFromCalculateInfoForTask({})",
        createPipeRequest.getPipeName());

    pipeStaticMeta =
        new PipeStaticMeta(
            createPipeRequest.getPipeName(),
            System.currentTimeMillis(),
            createPipeRequest.getExtractorAttributes(),
            createPipeRequest.getProcessorAttributes(),
            createPipeRequest.getConnectorAttributes());

    final ConcurrentMap<Integer, PipeTaskMeta> consensusGroupIdToTaskMetaMap =
        new ConcurrentHashMap<>();

    if (PipeType.CONSENSUS.equals(pipeStaticMeta.getPipeType())) {
      final TConsensusGroupId groupId =
          ConsensusGroupId.Factory.createFromString(
                  createPipeRequest.getExtractorAttributes().get(EXTRACTOR_CONSENSUS_GROUP_ID_KEY))
              .convertToTConsensusGroupId();

      final int senderDataNodeId =
          Integer.parseInt(
              createPipeRequest
                  .getExtractorAttributes()
                  .get(EXTRACTOR_CONSENSUS_SENDER_DATANODE_ID_KEY));
      consensusGroupIdToTaskMetaMap.put(
          groupId.getId(),
          new PipeTaskMeta(
              new RecoverProgressIndex(senderDataNodeId, new SimpleProgressIndex(0, 0)),
              senderDataNodeId,
              groupId.getId(),
              false));
    } else if (pipeStaticMeta.isSourceExternal()) {
      // external source
      final PipeExternalSourceLoadBalancer loadBalancer =
          new PipeExternalSourceLoadBalancer(
              pipeStaticMeta
                  .getExtractorParameters()
                  .getStringOrDefault(
                      Arrays.asList(
                          PipeExtractorConstant.EXTERNAL_EXTRACTOR_BALANCE_STRATEGY_KEY,
                          PipeExtractorConstant.EXTERNAL_SOURCE_BALANCE_STRATEGY_KEY),
                      PipeExtractorConstant.EXTERNAL_EXTRACTOR_BALANCE_PROPORTION_STRATEGY));
      final int parallelism =
          pipeStaticMeta
              .getExtractorParameters()
              .getIntOrDefault(
                  Arrays.asList(
                      EXTERNAL_EXTRACTOR_PARALLELISM_KEY, EXTERNAL_SOURCE_PARALLELISM_KEY),
                  EXTERNAL_EXTRACTOR_PARALLELISM_DEFAULT_VALUE);
      loadBalancer
          .balance(parallelism, pipeStaticMeta, env.getConfigManager())
          .forEach(
              (taskIndex, leaderNodeId) -> {
                consensusGroupIdToTaskMetaMap.put(
                    taskIndex,
                    new PipeTaskMeta(
                        MinimumProgressIndex.INSTANCE, leaderNodeId, taskIndex, false));
              });
    } else {
      // data regions & schema regions
      env.getConfigManager()
          .getLoadManager()
          .getRegionLeaderMap()
          .forEach(
              (regionGroupId, regionLeaderNodeId) -> {
                final String databaseName =
                    env.getConfigManager().getPartitionManager().getRegionDatabase(regionGroupId);
                if (databaseName != null
                    && !databaseName.equals(SchemaConstant.SYSTEM_DATABASE)
                    && !databaseName.startsWith(SchemaConstant.SYSTEM_DATABASE + ".")) {
                  // Pipe only collect user's data, filter out metric database here.
                  consensusGroupIdToTaskMetaMap.put(
                      regionGroupId.getId(),
                      new PipeTaskMeta(
                          MinimumProgressIndex.INSTANCE,
                          regionLeaderNodeId,
                          regionGroupId.getId(),
                          false));
                }
              });

      // config region
      consensusGroupIdToTaskMetaMap.put(
          // 0 is the consensus group id of the config region, but data region id and schema region
          // id also start from 0, so we use Integer.MIN_VALUE to represent the config region
          Integer.MIN_VALUE,
          new PipeTaskMeta(
              MinimumProgressIndex.INSTANCE,
              // The leader of the config region is the config node itself
              ConfigNodeDescriptor.getInstance().getConf().getConfigNodeId(),
              Integer.MIN_VALUE,
              false));
    }

    pipeRuntimeMeta = new PipeRuntimeMeta(consensusGroupIdToTaskMetaMap);
    if (!createPipeRequest.needManuallyStart) {
      pipeRuntimeMeta.getStatus().set(PipeStatus.RUNNING);
    }
  }
Check the {@link PipePlugin} configuration in Pipe. If there is an error, throw {@link
PipeException}. If there is a Pipe with the same name and there is no IfNotExists condition in
{@link #createPipeRequest}, throw {@link PipeException}. If there is an IfNotExists condition,
return {@code false}. If there is no Pipe with the same name, return {@code true}.

@param env the environment for the procedure
@return {@code true} The pipeline does not exist {@code false} The pipeline already exists and
    satisfies the IfNotExists condition
@throws PipeException
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/task/CreatePipeProcedureV2.java