public CompletableFuture<JobClient> submitJob(
            JobGraph jobGraph, ClassLoader userCodeClassloader) throws Exception {
        MiniClusterConfiguration miniClusterConfig =
                getMiniClusterConfig(jobGraph.getMaximumParallelism());
        MiniCluster miniCluster = miniClusterFactory.apply(miniClusterConfig);
        miniCluster.start();

        return miniCluster
                .submitJob(jobGraph)
                .thenApplyAsync(
                        FunctionUtils.uncheckedFunction(
                                submissionResult -> {
                                    org.apache.flink.client.ClientUtils
                                            .waitUntilJobInitializationFinished(
                                                    () ->
                                                            miniCluster
                                                                    .getJobStatus(
                                                                            submissionResult
                                                                                    .getJobID())
                                                                    .get(),
                                                    () ->
                                                            miniCluster
                                                                    .requestJobResult(
                                                                            submissionResult
                                                                                    .getJobID())
                                                                    .get(),
                                                    userCodeClassloader);
                                    return submissionResult;
                                }))
                .thenApply(
                        result ->
                                new MiniClusterJobClient(
                                        result.getJobID(),
                                        miniCluster,
                                        userCodeClassloader,
                                        MiniClusterJobClient.JobFinalizationBehavior
                                                .SHUTDOWN_CLUSTER))
                .whenComplete(
                        (ignored, throwable) -> {
                            if (throwable != null) {
                                // We failed to create the JobClient and must shutdown to ensure
                                // cleanup.
                                shutDownCluster(miniCluster);
                            }
                        })
                .thenApply(Function.identity());
    }
@Test
    void testJobClientSavepoint() throws Exception {
        PerJobMiniClusterFactory perJobMiniClusterFactory = initializeMiniCluster();
        JobClient jobClient =
                perJobMiniClusterFactory
                        .submitJob(getCancellableJobGraph(), ClassLoader.getSystemClassLoader())
                        .get();

        while (jobClient.getJobStatus().get() != JobStatus.RUNNING) {
            Thread.sleep(50);
        }

        assertThatThrownBy(
                        () -> jobClient.triggerSavepoint(null, SavepointFormatType.DEFAULT).get(),
                        "is not a streaming job.")
                .isInstanceOf(ExecutionException.class);

        assertThatFuture(jobClient.stopWithSavepoint(true, null, SavepointFormatType.DEFAULT))
                .eventuallyFailsWith(ExecutionException.class)
                .withMessageContaining("is not a streaming job.");
    }