public void start(FlinkInfo flinkInfo) throws Exception {
        try {
            // Start a new task without savepoint
            if (StringUtils.isEmpty(flinkInfo.getJobId())) {
                IntegrationTaskRunner taskRunner = new IntegrationTaskRunner(flinkInfo,
                        TaskCommitType.START_NOW.getCode());
                Future<?> future = TaskRunService.submit(taskRunner);
                future.get();
            } else {
                // Restore an old task with savepoint
                boolean noSavepoint =
                        isNullOrTerminated(flinkInfo) || StringUtils.isEmpty(flinkInfo.getSavepointPath());
                if (noSavepoint) {
                    String message =
                            String.format("restore job failed, as " + JOB_TERMINATED_MSG, flinkInfo.getJobId());
                    log.error(message);
                    throw new Exception(message);
                }

                IntegrationTaskRunner taskRunner = new IntegrationTaskRunner(flinkInfo,
                        TaskCommitType.RESUME.getCode());
                Future<?> future = TaskRunService.submit(taskRunner);
                future.get();
            }
        } catch (Exception e) {
            log.warn("submit flink job failed for {}", flinkInfo, e);
            throw new Exception("submit flink job failed: " + e.getMessage());
        }
    }
Start the Flink job, if the job id was not empty, restore it.
inlong-manager/manager-plugins/base/src/main/java/org/apache/inlong/manager/plugin/flink/FlinkOperation.java