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 testTurnUpParallelismByOverwriteParallelism() throws Exception {
        JobVertex jobVertex = getBlockingJobVertex();
        JobGraph jobGraph = JobGraphTestUtils.streamingJobGraph(jobVertex);
        int overwriteParallelism = jobVertex.getParallelism() + 1;
        BlockingInvokable.reset(overwriteParallelism);

        Configuration configuration = new Configuration();
        configuration.set(
                PipelineOptions.PARALLELISM_OVERRIDES,
                ImmutableMap.of(
                        jobVertex.getID().toHexString(), String.valueOf(overwriteParallelism)));

        PerJobMiniClusterFactory perJobMiniClusterFactory = initializeMiniCluster(configuration);
        JobClient jobClient =
                perJobMiniClusterFactory
                        .submitJob(jobGraph, ClassLoader.getSystemClassLoader())
                        .get();

        // wait for tasks to be properly running
        BlockingInvokable.latch.await();

        jobClient.cancel().get();
        assertThatFuture(jobClient.getJobExecutionResult())
                .eventuallyFailsWith(ExecutionException.class)
                .withMessageContaining("Job was cancelled");

        assertThatMiniClusterIsShutdown();
    }