@Override
  public SettableFuture<ConfigTaskResult> removeDataNode(
      final RemoveDataNodeStatement removeDataNodeStatement) {
    final SettableFuture<ConfigTaskResult> future = SettableFuture.create();
    try (ConfigNodeClient configNodeClient =
        CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
      Set<Integer> nodeIds = removeDataNodeStatement.getNodeIds();

      Set<Integer> validNodeIds =
          configNodeClient.getDataNodeConfiguration(-1).getDataNodeConfigurationMap().keySet();

      Set<Integer> invalidNodeIds = new HashSet<>(nodeIds);
      invalidNodeIds.removeAll(validNodeIds);

      if (!invalidNodeIds.isEmpty()) {
        LOGGER.info("Cannot remove invalid nodeIds:{}", invalidNodeIds);
        nodeIds.removeAll(invalidNodeIds);
      }

      if (nodeIds.size() != 1) {
        LOGGER.error(
            "The DataNode to be removed is not in the cluster, or the input format is incorrect.");
        future.setException(
            new IOException(
                "The DataNode to be removed is not in the cluster, or the input format is incorrect."));
      }

      LOGGER.info("Starting to remove DataNode with nodeIds: {}", nodeIds);

      final Set<Integer> finalNodeIds = nodeIds;
      List<TDataNodeLocation> removeDataNodeLocations =
          configNodeClient
              .getDataNodeConfiguration(-1)
              .getDataNodeConfigurationMap()
              .values()
              .stream()
              .map(TDataNodeConfiguration::getLocation)
              .filter(location -> finalNodeIds.contains(location.getDataNodeId()))
              .collect(Collectors.toList());

      List<String> simplifiedLocations = new ArrayList<>();
      for (TDataNodeLocation dataNodeLocation : removeDataNodeLocations) {
        simplifiedLocations.add(
            dataNodeLocation.getDataNodeId()
                + "@"
                + dataNodeLocation.getInternalEndPoint().getIp());
      }

      LOGGER.info("Start to remove datanode, removed DataNodes endpoint: {}", simplifiedLocations);
      TDataNodeRemoveReq removeReq = new TDataNodeRemoveReq(removeDataNodeLocations);
      TDataNodeRemoveResp removeResp = configNodeClient.removeDataNode(removeReq);
      LOGGER.info("Submit Remove DataNodes result {} ", removeResp);
      if (removeResp.getStatus().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
        future.setException(
            new IoTDBException(
                removeResp.getStatus().toString(), removeResp.getStatus().getCode()));
        return future;
      } else {
        LOGGER.info(
            "Submit remove-datanode request successfully, but the process may fail. "
                + "more details are shown in the logs of confignode-leader and removed-datanode, "
                + "and after the process of removing datanode ends successfully, "
                + "you are supposed to delete directory and data of the removed-datanode manually");
        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