@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 testRestoringModifiedJobFromSavepointFails() throws Exception {
        final JobGraph jobGraphWithNewOperator = createJobGraphWithSavepoint(false, 42L, 1);

        final ExecutionGraphFactory executionGraphFactory = createExecutionGraphFactory();

        assertThatThrownBy(
                        () ->
                                executionGraphFactory.createAndRestoreExecutionGraph(
                                        jobGraphWithNewOperator,
                                        new StandaloneCompletedCheckpointStore(1),
                                        new CheckpointsCleaner(),
                                        new StandaloneCheckpointIDCounter(),
                                        NoOpCheckpointStatsTracker.INSTANCE,
                                        TaskDeploymentDescriptorFactory.PartitionLocationConstraint
                                                .CAN_BE_UNKNOWN,
                                        0L,
                                        new DefaultVertexAttemptNumberStore(),
                                        SchedulerBase.computeVertexParallelismStore(
                                                jobGraphWithNewOperator),
                                        (execution, previousState, newState) -> {},
                                        rp -> false,
                                        log))
                .withFailMessage(
                        "Expected ExecutionGraph creation to fail because of non restored state.")
                .isInstanceOf(Exception.class)
                .hasMessageContaining("Failed to rollback to checkpoint/savepoint");
    }