@Override
  public SettableFuture<ConfigTaskResult> createTopic(
      final CreateTopicStatement createTopicStatement) {
    final SettableFuture<ConfigTaskResult> future = SettableFuture.create();

    final String topicName = createTopicStatement.getTopicName();
    final Map<String, String> topicAttributes = createTopicStatement.getTopicAttributes();

    // Replace now value with current time (raw timestamp based on system timestamp precision)
    final long currentTime =
        CommonDateTimeUtils.convertMilliTimeWithPrecision(
            System.currentTimeMillis(),
            CommonDescriptor.getInstance().getConfig().getTimestampPrecision());
    topicAttributes.computeIfPresent(
        TopicConstant.START_TIME_KEY,
        (k, v) -> {
          if (TopicConstant.NOW_TIME_VALUE.equals(v)) {
            return String.valueOf(currentTime);
          }
          return v;
        });
    topicAttributes.computeIfPresent(
        TopicConstant.END_TIME_KEY,
        (k, v) -> {
          if (TopicConstant.NOW_TIME_VALUE.equals(v)) {
            return String.valueOf(currentTime);
          }
          return v;
        });

    // Validate topic config
    final TopicMeta temporaryTopicMeta =
        new TopicMeta(topicName, System.currentTimeMillis(), topicAttributes);
    try {
      PipeDataNodeAgent.plugin()
          .validate(
              "fakePipeName",
              // TODO: currently use root to create topic
              temporaryTopicMeta.generateExtractorAttributes(
                  CommonDescriptor.getInstance().getConfig().getAdminName()),
              temporaryTopicMeta.generateProcessorAttributes(),
              temporaryTopicMeta.generateConnectorAttributes("fakeConsumerGroupId"));
    } catch (final Exception e) {
      future.setException(
          new IoTDBException(e.getMessage(), TSStatusCode.CREATE_TOPIC_ERROR.getStatusCode()));
      return future;
    }

    try (final ConfigNodeClient configNodeClient =
        CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
      final TCreateTopicReq req =
          new TCreateTopicReq()
              .setTopicName(topicName)
              .setIfNotExistsCondition(createTopicStatement.hasIfNotExistsCondition())
              .setTopicAttributes(topicAttributes);
      final TSStatus tsStatus = configNodeClient.createTopic(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 (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