docs: @Nonnull
  	public T createAndRestore(@Nonnull List<? extends Collection<S>> restoreOptions) 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 {
  				return attemptCreateAndRestore(restoreState);
  			} catch (Exception ex) {
  
  				collectedException = ExceptionUtils.firstOrSuppressed(ex, collectedException);
  
  				LOG.warn("Exception while restoring {} from alternative ({}/{}), will retry while more " +
  					"alternatives are available.", logDescription, alternativeIdx, restoreOptions.size(), ex);
  
  				if (backendCloseableRegistry.isClosed()) {
  					throw new FlinkException("Stopping restore attempts for already cancelled task.", collectedException);
  				}
  			}
  		}
  
  		throw new FlinkException("Could not restore " + logDescription + " from any of the " + restoreOptions.size() +
  			" provided restore options.", collectedException);
  	}