PackagedProgram packagedProgram = PackagedProgram.newBuilder()
                .setArguments(args)
                .setConfiguration(flinkConfig)
                .setJarFile(new File(flinkJobJarPath))
                .setEntryPointClassName(flinkJobClassName)
                .build();
JobGraph jobGraph = PackagedProgramUtils.createJobGraph(packagedProgram, flinkConfig, flinkJobParallelism, false);
log.info("submitting the job");
flinkClient.submitJob(jobGraph);
CompletableFuture<JobID> result = flinkClient.submitJob(jobGraph);