@Override
    public ExecutionGraph createAndRestoreExecutionGraph(
            JobGraph jobGraph,
            CompletedCheckpointStore completedCheckpointStore,
            CheckpointsCleaner checkpointsCleaner,
            CheckpointIDCounter checkpointIdCounter,
            CheckpointStatsTracker checkpointStatsTracker,
            TaskDeploymentDescriptorFactory.PartitionLocationConstraint partitionLocationConstraint,
            long initializationTimestamp,
            VertexAttemptNumberStore vertexAttemptNumberStore,
            VertexParallelismStore vertexParallelismStore,
            ExecutionStateUpdateListener executionStateUpdateListener,
            MarkPartitionFinishedStrategy markPartitionFinishedStrategy,
            Logger log)
            throws Exception {
        ExecutionDeploymentListener executionDeploymentListener =
                new ExecutionDeploymentTrackerDeploymentListenerAdapter(executionDeploymentTracker);
        ExecutionStateUpdateListener combinedExecutionStateUpdateListener =
                (execution, previousState, newState) -> {
                    executionStateUpdateListener.onStateUpdate(execution, previousState, newState);
                    if (newState.isTerminal()) {
                        executionDeploymentTracker.stopTrackingDeploymentOf(execution);
                    }
                };

        final ExecutionGraph newExecutionGraph =
                DefaultExecutionGraphBuilder.buildGraph(
                        jobGraph,
                        configuration,
                        futureExecutor,
                        ioExecutor,
                        userCodeClassLoader,
                        completedCheckpointStore,
                        checkpointsCleaner,
                        checkpointIdCounter,
                        rpcTimeout,
                        blobWriter,
                        log,
                        shuffleMaster,
                        jobMasterPartitionTracker,
                        partitionLocationConstraint,
                        executionDeploymentListener,
                        combinedExecutionStateUpdateListener,
                        initializationTimestamp,
                        vertexAttemptNumberStore,
                        vertexParallelismStore,
                        checkpointStatsTracker,
                        isDynamicGraph,
                        executionJobVertexFactory,
                        markPartitionFinishedStrategy,
                        nonFinishedHybridPartitionShouldBeUnknown,
                        jobManagerJobMetricGroup);

        final CheckpointCoordinator checkpointCoordinator =
                newExecutionGraph.getCheckpointCoordinator();

        if (checkpointCoordinator != null) {
            // check whether we find a valid checkpoint
            if (!checkpointCoordinator.restoreInitialCheckpointIfPresent(
                    new HashSet<>(newExecutionGraph.getAllVertices().values()))) {

                // check whether we can restore from a savepoint
                tryRestoreExecutionGraphFromSavepoint(
                        newExecutionGraph, jobGraph.getSavepointRestoreSettings());
            }
        }

        return newExecutionGraph;
    }
@Test
    void testCheckpointStatsTrackerUpdatedWithNewParallelism() throws Exception {
        final long savepointId = 42L;
        final JobGraph jobGraphWithParallelism2 = createJobGraphWithSavepoint(true, savepointId, 2);

        List<Span> spans = new ArrayList<>();
        final JobManagerJobMetricGroup jobManagerJobMetricGroup =
                new UnregisteredMetricGroups.UnregisteredJobManagerJobMetricGroup() {
                    @Override
                    public void addSpan(SpanBuilder spanBuilder) {
                        spans.add(spanBuilder.build());
                    }
                };
        final ExecutionGraphFactory executionGraphFactory =
                createExecutionGraphFactory(jobManagerJobMetricGroup);

        final StandaloneCompletedCheckpointStore completedCheckpointStore =
                new StandaloneCompletedCheckpointStore(1);
        final CheckpointStatsTracker checkpointStatsTracker =
                new DefaultCheckpointStatsTracker(10, jobManagerJobMetricGroup);
        ExecutionGraph executionGraph =
                executionGraphFactory.createAndRestoreExecutionGraph(
                        jobGraphWithParallelism2,
                        completedCheckpointStore,
                        new CheckpointsCleaner(),
                        new StandaloneCheckpointIDCounter(),
                        checkpointStatsTracker,
                        TaskDeploymentDescriptorFactory.PartitionLocationConstraint.CAN_BE_UNKNOWN,
                        0L,
                        new DefaultVertexAttemptNumberStore(),
                        vertexId ->
                                new DefaultVertexParallelismInfo(
                                        1, 1337, integer -> Optional.empty()),
                        (execution, previousState, newState) -> {},
                        rp -> false,
                        log);

        checkpointStatsTracker.reportRestoredCheckpoint(
                savepointId,
                CheckpointProperties.forSavepoint(false, SavepointFormatType.NATIVE),
                "foo",
                1337);

        final Set<ExecutionAttemptID> executionAttemptIDs =
                IterableUtils.toStream(executionGraph.getAllExecutionVertices())
                        .map(ExecutionVertex::getCurrentExecutionAttempt)
                        .map(Execution::getAttemptId)
                        .collect(Collectors.toSet());
        assertThat(executionAttemptIDs).hasSize(1);
        checkpointStatsTracker.reportInitializationMetrics(
                executionAttemptIDs.iterator().next(),
                new SubTaskInitializationMetricsBuilder(
                                SystemClock.getInstance().absoluteTimeMillis())
                        .build());

        assertThat(spans).hasSize(1);
    }