private String buildReleaseKubernetesCommand(Resource resource) {
    Preconditions.checkArgument(
        resource.getProperties().containsKey(KUBERNETES_CLUSTER_ID_PROPERTY),
        "Cannot find {} from optimizer start up stats.",
        KUBERNETES_CLUSTER_ID_PROPERTY);
    String clusterId = resource.getProperties().get(KUBERNETES_CLUSTER_ID_PROPERTY);

    FlinkConf conf =
        FlinkConf.buildFor(loadFlinkConfig(), getContainerProperties())
            .withGroupProperties(resource.getProperties())
            .build();
    conf.putToOptions(FlinkConfKeys.KUBERNETES_CLUSTER_ID, clusterId);

    String options = conf.toCliOptions();

    // job-manager rest api can't be visited from k8s cluster outside, so using kubernetes-session
    // command to kill it
    // echo 'stop' | $FLINK_HOME/bin/kubertnetes-session.sh
    return String.format(
        " echo 'stop' | %s/bin/kubernetes-session.sh %s -Dexecution.attached=true",
        this.flinkHome, options);
  }
memory conversion of units method, supporting m and g units
amoro-ams/src/main/java/org/apache/amoro/server/manager/FlinkOptimizerContainer.java