@Override
  public SettableFuture<ConfigTaskResult> createPipe(
      final CreatePipeStatement createPipeStatement) {
    final SettableFuture<ConfigTaskResult> future = SettableFuture.create();

    // Verify that Pipe is disabled if TSFile encryption is enabled
    if (!Objects.equals(TSFileDescriptor.getInstance().getConfig().getEncryptType(), "UNENCRYPTED")
        && !Objects.equals(
            TSFileDescriptor.getInstance().getConfig().getEncryptType(),
            "org.apache.tsfile.encrypt.UNENCRYPTED")) {
      future.setException(
          new IoTDBException(
              String.format(
                  "Failed to create Pipe %s because TSFile is configured with encryption, which prohibits the use of Pipe",
                  createPipeStatement.getPipeName()),
              TSStatusCode.PIPE_ERROR.getStatusCode()));
      return future;
    }

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

    // Validate pipe plugin before creation
    try {
      PipeDataNodeAgent.plugin()
          .validate(
              createPipeStatement.getPipeName(),
              createPipeStatement.getExtractorAttributes(),
              createPipeStatement.getProcessorAttributes(),
              createPipeStatement.getConnectorAttributes());
    } catch (final Exception e) {
      future.setException(
          new IoTDBException(e.getMessage(), TSStatusCode.PIPE_ERROR.getStatusCode()));
      return future;
    }

    try (final ConfigNodeClient configNodeClient =
        CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
      final TCreatePipeReq req =
          new TCreatePipeReq()
              .setPipeName(createPipeStatement.getPipeName())
              .setIfNotExistsCondition(createPipeStatement.hasIfNotExistsCondition())
              .setExtractorAttributes(createPipeStatement.getExtractorAttributes())
              .setProcessorAttributes(createPipeStatement.getProcessorAttributes())
              .setConnectorAttributes(createPipeStatement.getConnectorAttributes());
      final TSStatus tsStatus = configNodeClient.createPipe(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