@Override
  public SettableFuture<ConfigTaskResult> createFunction(
      Model model,
      String udfName,
      String className,
      Optional<String> stringURI,
      Class<?> baseClazz) {
    SettableFuture<ConfigTaskResult> future = SettableFuture.create();
    if (UDFManagementService.getInstance().checkIsBuiltInFunctionName(model, udfName)) {
      future.setException(
          new IoTDBException(
              String.format(
                  "Failed to create UDF [%s], the given function name conflicts with the built-in function name.",
                  udfName.toUpperCase()),
              TSStatusCode.CREATE_UDF_ERROR.getStatusCode()));
      return future;
    }
    try (ConfigNodeClient client =
        CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
      TCreateFunctionReq tCreateFunctionReq =
          new TCreateFunctionReq(udfName, className, false).setModel(model);
      String libRoot = UDFExecutableManager.getInstance().getLibRoot();
      String jarFileName;
      ByteBuffer jarFile;
      String jarMd5;
      if (stringURI.isPresent()) {
        String uriString = stringURI.get();
        jarFileName = new File(uriString).getName();
        try {
          URI uri = new URI(uriString);
          if (uri.getScheme() == null) {
            future.setException(
                new IoTDBException(
                    "The scheme of URI is not set, please specify the scheme of URI.",
                    TSStatusCode.UDF_DOWNLOAD_ERROR.getStatusCode()));
            return future;
          }
          if (!uri.getScheme().equals("file")) {
            // Download executable
            ExecutableResource resource =
                UDFExecutableManager.getInstance().request(Collections.singletonList(uriString));
            String jarFilePathUnderTempDir =
                UDFExecutableManager.getInstance()
                        .getDirStringUnderTempRootByRequestId(resource.getRequestId())
                    + jarFileName;
            // libRoot should be the path of the specified jar
            libRoot = jarFilePathUnderTempDir;
            jarFile = ExecutableManager.transferToBytebuffer(jarFilePathUnderTempDir);
            jarMd5 = DigestUtils.md5Hex(Files.newInputStream(Paths.get(jarFilePathUnderTempDir)));
          } else {
            // libRoot should be the path of the specified jar
            libRoot = new File(new URI(uriString)).getAbsolutePath();
            // If jarPath is a file path on datanode, we transfer it to ByteBuffer and send it to
            // ConfigNode.
            jarFile = ExecutableManager.transferToBytebuffer(libRoot);
            // Set md5 of the jar file
            jarMd5 = DigestUtils.md5Hex(Files.newInputStream(Paths.get(libRoot)));
          }
        } catch (IOException | URISyntaxException e) {
          LOGGER.warn("Failed to get executable for UDF({}) using URI: {}.", udfName, uriString, e);
          future.setException(
              new IoTDBException(
                  "Failed to get executable for UDF '" + udfName + "', please check the URI.",
                  TSStatusCode.TRIGGER_DOWNLOAD_ERROR.getStatusCode()));
          return future;
        }
        // modify req
        tCreateFunctionReq.setJarFile(jarFile);
        tCreateFunctionReq.setJarMD5(jarMd5);
        tCreateFunctionReq.setIsUsingURI(true);
        int index = jarFileName.lastIndexOf(".");
        if (index < 0) {
          tCreateFunctionReq.setJarName(String.format("%s-%s", jarFileName, jarMd5));
        } else {
          tCreateFunctionReq.setJarName(
              String.format(
                  "%s-%s.%s",
                  jarFileName.substring(0, index), jarMd5, jarFileName.substring(index + 1)));
        }
      }

      FunctionType functionType = FunctionType.NONE;
      // try to create instance, this request will fail if creation is not successful
      try (UDFClassLoader classLoader = new UDFClassLoader(libRoot)) {
        // ensure that jar file contains the class and the class is a UDF
        Class<?> clazz = Class.forName(className, true, classLoader);
        Object o = baseClazz.cast(clazz.getDeclaredConstructor().newInstance());
        if (Model.TABLE.equals(model)) {
          // we check function type for table model
          if (o instanceof ScalarFunction) {
            functionType = FunctionType.SCALAR;
          } else if (o instanceof AggregateFunction) {
            functionType = FunctionType.AGGREGATE;
          } else if (o instanceof TableFunction) {
            functionType = FunctionType.TABLE;
            // check there is no duplicate argument specification for name
            TableFunction tableFunction = (TableFunction) o;
            Set<String> argNames = new HashSet<>();
            for (ParameterSpecification specification :
                tableFunction.getArgumentsSpecifications()) {
              if (!argNames.add(specification.getName().toUpperCase())) {
                future.setException(
                    new IoTDBException(
                        "Failed to create function '"
                            + udfName
                            + "', because there is duplicate argument name '"
                            + specification.getName()
                            + "'.",
                        TSStatusCode.UDF_LOAD_CLASS_ERROR.getStatusCode()));
                return future;
              }
            }
          }
        }
        tCreateFunctionReq.setFunctionType(functionType);
      } catch (ClassNotFoundException
          | NoSuchMethodException
          | InstantiationException
          | IllegalAccessException
          | InvocationTargetException
          | ClassCastException e) {
        LOGGER.warn(
            "Failed to create function when try to create {}({}) instance first.",
            baseClazz.getSimpleName(),
            udfName,
            e);
        future.setException(
            new IoTDBException(
                "Failed to load class '"
                    + className
                    + "', because it's not found in jar file or is invalid: "
                    + stringURI.orElse(null),
                TSStatusCode.UDF_LOAD_CLASS_ERROR.getStatusCode()));
        return future;
      }

      final TSStatus executionStatus = client.createFunction(tCreateFunctionReq);
      if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != executionStatus.getCode()) {
        LOGGER.warn(
            "Failed to create function {}({}) because {}",
            udfName,
            className,
            executionStatus.getMessage());
        future.setException(new IoTDBException(executionStatus.message, executionStatus.code));
      } else {
        future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
      }
    } catch (ClientManagerException | IOException | TException 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