@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 testCanBeCanceledViaRegistry() throws Exception {
        CloseableRegistry closeableRegistry = new CloseableRegistry();
        OneShotLatch waitForBlock = new OneShotLatch();
        OneShotLatch unblock = new OneShotLatch();
        OperatorStateHandle blockingRestoreHandle = mock(OperatorStateHandle.class);
        when(blockingRestoreHandle.openInputStream())
                .thenReturn(new BlockingFSDataInputStream(waitForBlock, unblock));

        List<StateObjectCollection<OperatorStateHandle>> sortedRestoreOptions =
                Collections.singletonList(
                        new StateObjectCollection<>(
                                Collections.singletonList(blockingRestoreHandle)));

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

        AtomicReference<Exception> exceptionReference = new AtomicReference<>(null);
        Thread restoreThread =
                new Thread(
                        () -> {
                            try {
                                restorerProcedure.createAndRestore(
                                        sortedRestoreOptions,
                                        StateObject.StateObjectSizeStatsCollector.create());
                            } catch (Exception e) {
                                exceptionReference.set(e);
                            }
                        });

        restoreThread.start();
        waitForBlock.await();
        closeableRegistry.close();
        unblock.trigger();
        restoreThread.join();

        Exception exception = exceptionReference.get();
        assertThat(exception).isInstanceOf(FlinkException.class);
    }