public static <R extends ResourceVersion<R>>
            Collection<CompletedCheckpoint> retrieveCompletedCheckpoints(
                    StateHandleStore<CompletedCheckpoint, R> checkpointStateHandleStore,
                    CheckpointStoreUtil completedCheckpointStoreUtil)
                    throws Exception {

        LOG.info("Recovering checkpoints from {}.", checkpointStateHandleStore);

        // Get all there is first.
        final List<Tuple2<RetrievableStateHandle<CompletedCheckpoint>, String>> initialCheckpoints =
                checkpointStateHandleStore.getAllAndLock();

        // Sort checkpoints by name.
        initialCheckpoints.sort(Comparator.comparing(o -> o.f1));

        final int numberOfInitialCheckpoints = initialCheckpoints.size();

        LOG.info(
                "Found {} checkpoints in {}.",
                numberOfInitialCheckpoints,
                checkpointStateHandleStore);
        final List<CompletedCheckpoint> retrievedCheckpoints =
                new ArrayList<>(numberOfInitialCheckpoints);
        LOG.info("Trying to fetch {} checkpoints from storage.", numberOfInitialCheckpoints);

        for (Tuple2<RetrievableStateHandle<CompletedCheckpoint>, String> checkpointStateHandle :
                initialCheckpoints) {
            retrievedCheckpoints.add(
                    checkNotNull(
                            retrieveCompletedCheckpoint(
                                    completedCheckpointStoreUtil, checkpointStateHandle)));
        }
        return Collections.unmodifiableList(retrievedCheckpoints);
    }
@Test
    void testRetrievingCheckpointsFailsIfRetrievalOfAnyCheckpointFails() throws Exception {
        final TestingRetrievableStateStorageHelper<CompletedCheckpoint> storageHelper =
                new TestingRetrievableStateStorageHelper<>();
        final List<Tuple2<RetrievableStateHandle<CompletedCheckpoint>, String>> handles =
                new ArrayList<>();
        handles.add(Tuple2.of(storageHelper.store(createCompletedCheckpoint(0L)), "checkpoint-0"));
        handles.add(Tuple2.of(new FailingRetrievableStateHandle<>(), "checkpoint-1"));
        handles.add(Tuple2.of(storageHelper.store(createCompletedCheckpoint(2L)), "checkpoint-2"));
        Collections.shuffle(handles);
        final TestingStateHandleStore<CompletedCheckpoint> stateHandleStore =
                TestingStateHandleStore.<CompletedCheckpoint>newBuilder()
                        .setGetAllSupplier(() -> handles)
                        .build();
        assertThatExceptionOfType(FlinkException.class)
                .isThrownBy(
                        () ->
                                DefaultCompletedCheckpointStoreUtils.retrieveCompletedCheckpoints(
                                        stateHandleStore, new SimpleCheckpointStoreUtil()));
    }