public boolean restoreLatestCheckpointedStateToAll(
            final Set<ExecutionJobVertex> tasks, final boolean allowNonRestoredState)
            throws Exception {

        final OptionalLong restoredCheckpointId =
                restoreLatestCheckpointedStateInternal(
                        tasks,
                        OperatorCoordinatorRestoreBehavior
                                .RESTORE_OR_RESET, // global recovery restores coordinators, or
                        // resets them to empty
                        false, // recovery might come before first successful checkpoint
                        allowNonRestoredState,
                        false);

        return restoredCheckpointId.isPresent();
    }
@Test
    void testCheckpointStatsTrackerRestoreCallback() throws Exception {
        StandaloneCompletedCheckpointStore store = new StandaloneCompletedCheckpointStore(1);

        // set up the coordinator and validate the initial state
        CheckpointStatsTracker tracker =
                new DefaultCheckpointStatsTracker(
                        10, UnregisteredMetricGroups.createUnregisteredJobManagerJobMetricGroup());
        CheckpointCoordinator checkpointCoordinator =
                new CheckpointCoordinatorBuilder()
                        .setCompletedCheckpointStore(store)
                        .setTimer(manuallyTriggeredScheduledExecutor)
                        .setCheckpointStatsTracker(tracker)
                        .build(EXECUTOR_RESOURCE.getExecutor());

        store.addCheckpointAndSubsumeOldestOne(
                new CompletedCheckpoint(
                        new JobID(),
                        42,
                        0,
                        0,
                        Collections.<OperatorID, OperatorState>emptyMap(),
                        Collections.<MasterState>emptyList(),
                        CheckpointProperties.forCheckpoint(
                                CheckpointRetentionPolicy.NEVER_RETAIN_AFTER_TERMINATION),
                        new TestCompletedCheckpointStorageLocation(),
                        null),
                new CheckpointsCleaner(),
                () -> {});

        assertThat(
                        checkpointCoordinator.restoreLatestCheckpointedStateToAll(
                                Collections.emptySet(), true))
                .isTrue();

        assertThat(tracker.createSnapshot().getLatestRestoredCheckpoint().getCheckpointId())
                .isEqualTo(42);
    }