@Override
  public SettableFuture<ConfigTaskResult> alterPipe(final AlterPipeStatement alterPipeStatement) {
    final SettableFuture<ConfigTaskResult> future = SettableFuture.create();

    // Validate pipe name
    if (alterPipeStatement.getPipeName().startsWith(PipeStaticMeta.SYSTEM_PIPE_PREFIX)) {
      future.setException(
          new IoTDBException(
              String.format(
                  "Failed to alter pipe %s, pipe name starting with \"%s\" are not allowed to be altered.",
                  alterPipeStatement.getPipeName(), PipeStaticMeta.SYSTEM_PIPE_PREFIX),
              TSStatusCode.PIPE_ERROR.getStatusCode()));
      return future;
    }

    // Validate pipe existence
    final PipeMeta pipeMetaFromCoordinator;
    try (final ConfigNodeClient configNodeClient =
        ConfigNodeClientManager.getInstance().borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
      final TGetAllPipeInfoResp getAllPipeInfoResp = configNodeClient.getAllPipeInfo();
      if (getAllPipeInfoResp.getStatus().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
        future.setException(
            new IoTDBException(
                String.format(
                    "Failed to get pipe info from config node, status is %s.",
                    getAllPipeInfoResp.getStatus()),
                TSStatusCode.PIPE_ERROR.getStatusCode()));
        return future;
      }

      pipeMetaFromCoordinator =
          getAllPipeInfoResp.getAllPipeInfo().stream()
              .map(PipeMeta::deserialize4Coordinator)
              .filter(
                  pipeMeta ->
                      pipeMeta
                          .getStaticMeta()
                          .getPipeName()
                          .equals(alterPipeStatement.getPipeName()))
              .findFirst()
              .orElse(null);
      if (pipeMetaFromCoordinator == null) {
        future.setException(
            new IoTDBException(
                String.format(
                    "Failed to alter pipe %s, pipe not found in system.",
                    alterPipeStatement.getPipeName()),
                TSStatusCode.PIPE_ERROR.getStatusCode()));
        return future;
      }
    } catch (final Exception e) {
      future.setException(
          new IoTDBException(
              String.format(
                  "Failed to alter pipe %s, because %s",
                  alterPipeStatement.getPipeName(), e.getMessage()),
              TSStatusCode.PIPE_ERROR.getStatusCode()));
      return future;
    }

    // Construct temporary pipe static meta for validation
    final String pipeName = alterPipeStatement.getPipeName();
    final Map<String, String> extractorAttributes;
    final Map<String, String> processorAttributes;
    final Map<String, String> connectorAttributes;
    try {
      if (!alterPipeStatement.getExtractorAttributes().isEmpty()) {
        // We don't allow changing the extractor plugin type
        if (alterPipeStatement
                .getExtractorAttributes()
                .containsKey(PipeExtractorConstant.EXTRACTOR_KEY)
            || alterPipeStatement
                .getExtractorAttributes()
                .containsKey(PipeExtractorConstant.SOURCE_KEY)
            || alterPipeStatement.isReplaceAllExtractorAttributes()) {
          checkIfSourcePluginChanged(
              pipeMetaFromCoordinator.getStaticMeta().getExtractorParameters(),
              new PipeParameters(alterPipeStatement.getExtractorAttributes()));
        }
        if (alterPipeStatement.isReplaceAllExtractorAttributes()) {
          extractorAttributes = alterPipeStatement.getExtractorAttributes();
        } else {
          final boolean onlyContainsUser =
              onlyContainsUser(alterPipeStatement.getExtractorAttributes());
          pipeMetaFromCoordinator
              .getStaticMeta()
              .getExtractorParameters()
              .addOrReplaceEquivalentAttributes(
                  new PipeParameters(alterPipeStatement.getExtractorAttributes()));
          extractorAttributes =
              pipeMetaFromCoordinator.getStaticMeta().getExtractorParameters().getAttribute();
          if (onlyContainsUser) {
            checkSourceType(alterPipeStatement.getPipeName(), extractorAttributes);
          }
        }
      } else {
        extractorAttributes =
            pipeMetaFromCoordinator.getStaticMeta().getExtractorParameters().getAttribute();
      }

      if (!alterPipeStatement.getProcessorAttributes().isEmpty()) {
        if (alterPipeStatement.isReplaceAllProcessorAttributes()) {
          processorAttributes = alterPipeStatement.getProcessorAttributes();
        } else {
          pipeMetaFromCoordinator
              .getStaticMeta()
              .getProcessorParameters()
              .addOrReplaceEquivalentAttributes(
                  new PipeParameters(alterPipeStatement.getProcessorAttributes()));
          processorAttributes =
              pipeMetaFromCoordinator.getStaticMeta().getProcessorParameters().getAttribute();
        }
      } else {
        processorAttributes =
            pipeMetaFromCoordinator.getStaticMeta().getProcessorParameters().getAttribute();
      }

      if (!alterPipeStatement.getConnectorAttributes().isEmpty()) {
        if (alterPipeStatement.isReplaceAllConnectorAttributes()) {
          connectorAttributes = alterPipeStatement.getConnectorAttributes();
        } else {
          final boolean onlyContainsUser =
              onlyContainsUser(alterPipeStatement.getConnectorAttributes());
          pipeMetaFromCoordinator
              .getStaticMeta()
              .getConnectorParameters()
              .addOrReplaceEquivalentAttributes(
                  new PipeParameters(alterPipeStatement.getConnectorAttributes()));
          connectorAttributes =
              pipeMetaFromCoordinator.getStaticMeta().getConnectorParameters().getAttribute();
          if (onlyContainsUser) {
            checkSinkType(alterPipeStatement.getPipeName(), connectorAttributes);
          }
        }
      } else {
        connectorAttributes =
            pipeMetaFromCoordinator.getStaticMeta().getConnectorParameters().getAttribute();
      }

      PipeDataNodeAgent.plugin()
          .validate(pipeName, extractorAttributes, processorAttributes, connectorAttributes);
    } catch (final Exception e) {
      future.setException(
          new IoTDBException(e.getMessage(), TSStatusCode.PIPE_ERROR.getStatusCode()));
      return future;
    }

    try (ConfigNodeClient configNodeClient =
        CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
      final TAlterPipeReq req =
          new TAlterPipeReq(
              pipeName,
              alterPipeStatement.getProcessorAttributes(),
              alterPipeStatement.getConnectorAttributes(),
              alterPipeStatement.isReplaceAllProcessorAttributes(),
              alterPipeStatement.isReplaceAllConnectorAttributes());
      req.setExtractorAttributes(alterPipeStatement.getExtractorAttributes());
      req.setIsReplaceAllExtractorAttributes(alterPipeStatement.isReplaceAllExtractorAttributes());
      req.setIfExistsCondition(alterPipeStatement.hasIfExistsCondition());
      req.setIsTableModel(alterPipeStatement.isTableModel());
      final TSStatus tsStatus = configNodeClient.alterPipe(req);
      if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != tsStatus.getCode()) {
        future.setException(new IoTDBException(tsStatus.message, tsStatus.code));
      } else {
        future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
      }
    } catch (final Exception e) {
      future.setException(e);
    }
    return future;
  }
FIXME Consolidate this clientManager with the upper one.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java