public class K8sApplicationClient {

    protected static final String KUBE_CONFIG_FILE = System.getProperty("user.dir") + "/river-flink/src/main/resources/k8s/config";
    protected static final String JOB_MANAGER_SERVICE_ACCOUNT = "flink";

    protected static final String NAMESPACE = "default";
    protected static final String CLUSTER_ID = "my-first-application-cluster";
    protected static final String CONTAINER_IMAGE = "flinkjobdemo:001";
    protected static String JOB_JAR = "local:///opt/flink/usrlib/FlinkJobdemo.jar";
    protected static final KubernetesConfigOptions.ImagePullPolicy CONTAINER_IMAGE_PULL_POLICY =
            KubernetesConfigOptions.ImagePullPolicy.IfNotPresent;
    protected static final int JOB_MANAGER_MEMORY = 768;
    protected static final int TOTAL_FLINK_MEMORY = 768;

    private static KubernetesClusterDescriptor descriptor;
    private static ClusterSpecification clusterSpecification =
            new ClusterSpecification.ClusterSpecificationBuilder().createClusterSpecification();
    private static ApplicationConfiguration appConfig =
            new ApplicationConfiguration(new String[0], null);

    private static Configuration flinkConfig = new Configuration();
    private static File flinkConfDir = new File(System.getProperty("user.dir") + "/river-flink/flinkJobTemp/test-job001");

    public static void main(String[] args) {
        try {
            setup();
            submit();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    private static void submit () throws Exception {
        KubernetesClusterClientFactory factory = new KubernetesClusterClientFactory();
        descriptor = factory.createClusterDescriptor(flinkConfig);
        descriptor.deployApplicationCluster(clusterSpecification, appConfig);
    }

    public static void setup() throws Exception {
        setupFlinkConfig();
        writeFlinkConfiguration();
    }

    protected static void setupFlinkConfig() {
        // 作业镜像地址
        flinkConfig.set(PipelineOptions.JARS, Collections.singletonList(JOB_JAR));
        // 配置k8s
        flinkConfig.set(DeploymentOptions.TARGET, KubernetesDeploymentTarget.APPLICATION.getName());
        // k8s config path
        flinkConfig.setString(KubernetesConfigOptions.KUBE_CONFIG_FILE, KUBE_CONFIG_FILE);
        // k8s 服务账号
        flinkConfig.setString(KubernetesConfigOptions.JOB_MANAGER_SERVICE_ACCOUNT, JOB_MANAGER_SERVICE_ACCOUNT);
        // namespace
        flinkConfig.setString(KubernetesConfigOptions.NAMESPACE, NAMESPACE);
        // cluster_id
        flinkConfig.setString(KubernetesConfigOptions.CLUSTER_ID, CLUSTER_ID);
        // docker镜像
        flinkConfig.setString(KubernetesConfigOptions.CONTAINER_IMAGE, CONTAINER_IMAGE);
        // 镜像拉取策略
//        flinkConfig.set(KubernetesConfigOptions.CONTAINER_IMAGE_PULL_POLICY, CONTAINER_IMAGE_PULL_POLICY);

        // jobManager memory
        flinkConfig.set(JobManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.ofMebiBytes(JOB_MANAGER_MEMORY));
        // taskManager memory
        flinkConfig.set(TaskManagerOptions.TOTAL_FLINK_MEMORY, MemorySize.ofMebiBytes(TOTAL_FLINK_MEMORY));

        // flinkConf 路径
        flinkConfig.set(DeploymentOptionsInternal.CONF_DIR, flinkConfDir.toString());
    }

    protected static void writeFlinkConfiguration() throws IOException {
        BootstrapTools.writeConfiguration(
                flinkConfig, new File(flinkConfDir, "flink-conf.yaml"));
    }

}