@Nonnull
    public T createAndRestore(
            @Nonnull List<? extends Collection<S>> restoreOptions,
            @Nonnull StateObject.StateObjectSizeStatsCollector stats)
            throws Exception {

        if (restoreOptions.isEmpty()) {
            restoreOptions = Collections.singletonList(Collections.emptyList());
        }

        int alternativeIdx = 0;

        Exception collectedException = null;

        while (alternativeIdx < restoreOptions.size()) {

            Collection<S> restoreState = restoreOptions.get(alternativeIdx);

            ++alternativeIdx;

            // IMPORTANT: please be careful when modifying the log statements because they are used
            // for validation in
            // the automatic end-to-end tests. Those tests might fail if they are not aligned with
            // the log message!
            if (restoreState.isEmpty()) {
                LOG.debug("Creating {} with empty state.", logDescription);
            } else {
                if (LOG.isTraceEnabled()) {
                    LOG.trace(
                            "Creating {} and restoring with state {} from alternative ({}/{}).",
                            logDescription,
                            restoreState,
                            alternativeIdx,
                            restoreOptions.size());
                } else {
                    LOG.debug(
                            "Creating {} and restoring with state from alternative ({}/{}).",
                            logDescription,
                            alternativeIdx,
                            restoreOptions.size());
                }
            }

            try {
                T successfullyRestored = attemptCreateAndRestore(restoreState);
                // Obtain and report stats for the state objects used in our successful restore
                restoreState.forEach(handle -> handle.collectSizeStats(stats));
                return successfullyRestored;
            } catch (Exception ex) {

                collectedException = ExceptionUtils.firstOrSuppressed(ex, collectedException);

                if (backendCloseableRegistry.isClosed()) {
                    throw new FlinkException(
                            "Stopping restore attempts for already cancelled task.",
                            collectedException);
                }

                LOG.warn(
                        "Exception while restoring {} from alternative ({}/{}), will retry while more "
                                + "alternatives are available.",
                        logDescription,
                        alternativeIdx,
                        restoreOptions.size(),
                        ex);
            }
        }

        throw new FlinkException(
                "Could not restore "
                        + logDescription
                        + " from any of the "
                        + restoreOptions.size()
                        + " provided restore options.",
                collectedException);
    }
@Test
    void testRestoreProcedureOrderAndFailure() throws Exception {

        CloseableRegistry closeableRegistry = new CloseableRegistry();
        CheckpointStreamFactory checkpointStreamFactory = new MemCheckpointStreamFactory(1024);

        ListStateDescriptor<Integer> stateDescriptor =
                new ListStateDescriptor<>("test-state", Integer.class);
        OperatorStateBackend originalBackend = backendSupplier.apply(Collections.emptyList());
        SnapshotResult<OperatorStateHandle> snapshotResult;

        try {
            ListState<Integer> listState = originalBackend.getListState(stateDescriptor);

            listState.add(0);
            listState.add(1);
            listState.add(2);
            listState.add(3);

            RunnableFuture<SnapshotResult<OperatorStateHandle>> snapshot =
                    originalBackend.snapshot(
                            0L,
                            0L,
                            checkpointStreamFactory,
                            CheckpointOptions.forCheckpointWithDefaultLocation());

            snapshot.run();
            snapshotResult = snapshot.get();

        } finally {
            originalBackend.close();
            originalBackend.dispose();
        }

        OperatorStateHandle firstFailHandle = mock(OperatorStateHandle.class);
        OperatorStateHandle secondSuccessHandle = spy(snapshotResult.getJobManagerOwnedSnapshot());
        OperatorStateHandle thirdNotUsedHandle = mock(OperatorStateHandle.class);

        List<StateObjectCollection<OperatorStateHandle>> sortedRestoreOptions =
                Arrays.asList(
                        new StateObjectCollection<>(Collections.singletonList(firstFailHandle)),
                        new StateObjectCollection<>(Collections.singletonList(secondSuccessHandle)),
                        new StateObjectCollection<>(Collections.singletonList(thirdNotUsedHandle)));

        BackendRestorerProcedure<OperatorStateBackend, OperatorStateHandle> restorerProcedure =
                new BackendRestorerProcedure<>(
                        backendSupplier, closeableRegistry, "test op state backend");

        OperatorStateBackend restoredBackend =
                restorerProcedure.createAndRestore(
                        sortedRestoreOptions, StateObject.StateObjectSizeStatsCollector.create());
        assertThat(restoredBackend).isNotNull();

        try {
            verify(firstFailHandle).openInputStream();
            verify(secondSuccessHandle).openInputStream();
            verify(thirdNotUsedHandle, times(0)).openInputStream();

            ListState<Integer> listState = restoredBackend.getListState(stateDescriptor);

            Iterator<Integer> stateIterator = listState.get().iterator();
            assertThat(stateIterator.next()).isZero();
            assertThat(stateIterator.next()).isOne();
            assertThat(stateIterator.next()).isEqualTo(2);
            assertThat(stateIterator.next()).isEqualTo(3);
            assertThat(stateIterator).isExhausted();

        } finally {
            restoredBackend.close();
            restoredBackend.dispose();
        }
    }