@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 testRestoringModifiedJobFromSavepointWithAllowNonRestoredStateSucceeds() throws Exception {
        // create savepoint data
        final long savepointId = 42L;
        final JobGraph jobGraphWithNewOperator = createJobGraphWithSavepoint(true, savepointId, 1);

        final ExecutionGraphFactory executionGraphFactory = createExecutionGraphFactory();

        final StandaloneCompletedCheckpointStore completedCheckpointStore =
                new StandaloneCompletedCheckpointStore(1);
        executionGraphFactory.createAndRestoreExecutionGraph(
                jobGraphWithNewOperator,
                completedCheckpointStore,
                new CheckpointsCleaner(),
                new StandaloneCheckpointIDCounter(),
                NoOpCheckpointStatsTracker.INSTANCE,
                TaskDeploymentDescriptorFactory.PartitionLocationConstraint.CAN_BE_UNKNOWN,
                0L,
                new DefaultVertexAttemptNumberStore(),
                SchedulerBase.computeVertexParallelismStore(jobGraphWithNewOperator),
                (execution, previousState, newState) -> {},
                rp -> false,
                log);

        final CompletedCheckpoint savepoint = completedCheckpointStore.getLatestCheckpoint();

        assertThat(savepoint).isNotNull();
        assertThat(savepoint.getCheckpointID()).isEqualTo(savepointId);
    }