@Override
  public void releaseResource(Resource resource) {
    if (target.runByFlinkRestClient()) {
      Preconditions.checkArgument(
          resource.getProperties().containsKey(SESSION_CLUSTER_JOB_ID),
          "Cannot find {} from optimizer start up stats",
          SESSION_CLUSTER_JOB_ID);
      String jobId = resource.getProperties().get(SESSION_CLUSTER_JOB_ID);
      Configuration configuration =
          FlinkConf.buildFor(loadFlinkConfig(), getContainerProperties())
              .withGroupProperties(resource.getProperties())
              .build()
              .toConfiguration();

      try (RestClusterClient<String> restClusterClient =
          FlinkClientUtil.getRestClusterClient(configuration)) {
        Acknowledge ignore =
            restClusterClient
                .cancel(JobID.fromHexString(jobId))
                .get(flinkClientTimeoutSecond, TimeUnit.SECONDS);
      } catch (TimeoutException timeoutException) {
        throw new RuntimeException(
            String.format(
                "Timed out stopping the optimizer in Flink cluster, "
                    + "please configure a larger timeout via '%s'",
                FLINK_CLIENT_TIMEOUT_SECOND));
      } catch (Exception e) {
        throw new RuntimeException("Failed to release flink optimizer", e);
      }
    } else {
      String releaseCommand;
      switch (target) {
        case YARN_APPLICATION:
        case YARN_PER_JOB:
          releaseCommand = buildReleaseYarnCommand(resource);
          break;
        case KUBERNETES_APPLICATION:
          releaseCommand = buildReleaseKubernetesCommand(resource);
          break;
        default:
          throw new IllegalStateException("Unsupported running target: " + target.getValue());
      }

      try {
        String exportCmd = String.join(" && ", exportSystemProperties());
        String releaseCmd = exportCmd + " && " + releaseCommand;
        String[] cmd = {"/bin/sh", "-c", releaseCmd};
        LOG.info("Releasing flink optimizer using command: {}", releaseCmd);
        Runtime.getRuntime().exec(cmd);
      } catch (IOException e) {
        throw new UncheckedIOException("Failed to release flink optimizer.", e);
      }
    }
  }
memory conversion of units method, supporting m and g units
amoro-ams/src/main/java/org/apache/amoro/server/manager/FlinkOptimizerContainer.java