@Override
    public CompletableFuture<Acknowledge> submitJob(JobGraph jobGraph, Time timeout) {
        final JobID jobID = jobGraph.getJobID();
        try (MdcCloseable ignored = MdcUtils.withContext(MdcUtils.asContextData(jobID))) {
            log.info("Received JobGraph submission '{}' ({}).", jobGraph.getName(), jobID);
        }
        return isInGloballyTerminalState(jobID)
                .thenComposeAsync(
                        isTerminated -> {
                            if (isTerminated) {
                                log.warn(
                                        "Ignoring JobGraph submission '{}' ({}) because the job already "
                                                + "reached a globally-terminal state (i.e. {}) in a "
                                                + "previous execution.",
                                        jobGraph.getName(),
                                        jobID,
                                        Arrays.stream(JobStatus.values())
                                                .filter(JobStatus::isGloballyTerminalState)
                                                .map(JobStatus::name)
                                                .collect(Collectors.joining(", ")));
                                return FutureUtils.completedExceptionally(
                                        DuplicateJobSubmissionException.ofGloballyTerminated(
                                                jobID));
                            } else if (jobManagerRunnerRegistry.isRegistered(jobID)
                                    || submittedAndWaitingTerminationJobIDs.contains(jobID)) {
                                // job with the given jobID is not terminated, yet
                                return FutureUtils.completedExceptionally(
                                        DuplicateJobSubmissionException.of(jobID));
                            } else if (isPartialResourceConfigured(jobGraph)) {
                                return FutureUtils.completedExceptionally(
                                        new JobSubmissionException(
                                                jobID,
                                                "Currently jobs is not supported if parts of the vertices "
                                                        + "have resources configured. The limitation will be "
                                                        + "removed in future versions."));
                            } else {
                                return internalSubmitJob(jobGraph);
                            }
                        },
                        getMainThreadExecutor(jobID));
    }
@Test
    public void testOverridingJobVertexParallelisms() throws Exception {
        JobVertex v1 = new JobVertex("v1");
        v1.setParallelism(1);
        JobVertex v2 = new JobVertex("v2");
        v2.setParallelism(2);
        JobVertex v3 = new JobVertex("v3");
        v3.setParallelism(3);
        jobGraph = new JobGraph(jobGraph.getJobID(), "job", v1, v2, v3);

        configuration.set(
                PipelineOptions.PARALLELISM_OVERRIDES,
                ImmutableMap.of(
                        v1.getID().toHexString(), "10",
                        // v2 is omitted
                        v3.getID().toHexString(), "21",
                        // unknown vertex added
                        new JobVertexID().toHexString(), "23"));

        jobGraph.getJobConfiguration()
                .set(
                        PipelineOptions.PARALLELISM_OVERRIDES,
                        ImmutableMap.of(
                                // verifies that job graph configuration has higher priority
                                v3.getID().toHexString(), "42",
                                // unknown vertex added
                                new JobVertexID().toHexString(), "25"));

        dispatcher =
                createAndStartDispatcher(
                        heartbeatServices,
                        haServices,
                        new ExpectedJobIdJobManagerRunnerFactory(jobId));
        DispatcherGateway dispatcherGateway = dispatcher.getSelfGateway(DispatcherGateway.class);

        Assert.assertEquals(jobGraph.findVertexByID(v1.getID()).getParallelism(), 1);
        Assert.assertEquals(jobGraph.findVertexByID(v2.getID()).getParallelism(), 2);
        Assert.assertEquals(jobGraph.findVertexByID(v3.getID()).getParallelism(), 3);

        dispatcherGateway.submitJob(jobGraph, TIMEOUT).get();

        Assert.assertEquals(jobGraph.findVertexByID(v1.getID()).getParallelism(), 10);
        Assert.assertEquals(jobGraph.findVertexByID(v2.getID()).getParallelism(), 2);
        Assert.assertEquals(jobGraph.findVertexByID(v3.getID()).getParallelism(), 42);
    }