public static PipeMeta copyAndFilterOutNonWorkingDataRegionPipeTasks(PipeMeta originalPipeMeta)
      throws IOException {
    final PipeMeta copiedPipeMeta = originalPipeMeta.deepCopy4TaskAgent();

    copiedPipeMeta
        .getRuntimeMeta()
        .getConsensusGroupId2TaskMetaMap()
        .entrySet()
        .removeIf(
            consensusGroupId2TaskMeta -> {
              if (originalPipeMeta.getStaticMeta().isSourceExternal()) {
                // should keep the external source tasks
                return false;
              }
              final String database;
              try {
                database =
                    ConfigNode.getInstance()
                        .getConfigManager()
                        .getPartitionManager()
                        .getRegionDatabase(
                            new TConsensusGroupId(
                                // We assume that the consensus group id is a data region id.
                                TConsensusGroupType.DataRegion,
                                consensusGroupId2TaskMeta.getKey()));
                if (database == null) {
                  // If the consensus group id is not a data region id, we keep it.
                  // If the consensus group id is a data region id, but the database is not found,
                  // we keep it.
                  return false;
                }
              } catch (final Exception ignore) {
                // In case of any exception, we keep the consensus group id.
                return false;
              }

              final boolean isTableModel;
              try {
                final TDatabaseSchema schema =
                    ConfigNode.getInstance()
                        .getConfigManager()
                        .getClusterSchemaManager()
                        .getDatabaseSchemaByName(database);
                if (schema == null) {
                  // If the database is not found, we keep it.
                  return false;
                }
                isTableModel = schema.isIsTableModel();
              } catch (final Exception ignore) {
                // If the database is not found, we keep it.
                return false;
              }

              try {
                return !DataRegionListeningFilter.shouldDatabaseBeListened(
                    copiedPipeMeta.getStaticMeta().getExtractorParameters(),
                    isTableModel,
                    database);
              } catch (final Exception e) {
                return false;
              }
            });

    return copiedPipeMeta;
  }
Drop a pipe on all the dataNodes.

@param pipeName pipe name of the pipe to drop
@param env ConfigNodeProcedureEnv
@return The responseMap after pushing pipe meta
iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/AbstractOperatePipeProcedureV2.java