public String stopJob(String jobId, boolean isDrain, String savepointDirectory) throws Exception {
        try {
            JobID jobID = JobID.fromHexString(jobId);
            CompletableFuture<String> stopResult = flinkClient.stopWithSavepoint(jobID, isDrain, savepointDirectory,
                    SavepointFormatType.CANONICAL);
            return stopResult.get();
        } catch (Exception e) {
            log.error("stop job {} failed and savepoint directory is {} : ", jobId, savepointDirectory, e);
            throw new Exception("stop job " + jobId + " failed: " + e.getMessage());
        }
    }
Stop the Flink job with the savepoint.
inlong-manager/manager-plugins/manager-plugins-flink-v1.18/src/main/java/org/apache/inlong/manager/plugin/flink/FlinkClientService.java