public static StateBackend fromApplicationOrConfigOrDefault(
			@Nullable StateBackend fromApplication,
			Configuration config,
			ClassLoader classLoader,
			@Nullable Logger logger) throws IllegalConfigurationException, DynamicCodeLoadingException, IOException {

		checkNotNull(config, "config");
		checkNotNull(classLoader, "classLoader");

		final StateBackend backend;

		// (1) the application defined state backend has precedence
		if (fromApplication != null) {
			if (logger != null) {
				logger.info("Using application-defined state backend: {}", fromApplication);
			}

			// see if this is supposed to pick up additional configuration parameters
			if (fromApplication instanceof ConfigurableStateBackend) {
				// needs to pick up configuration
				if (logger != null) {
					logger.info("Configuring application-defined state backend with job/cluster config");
				}

				backend = ((ConfigurableStateBackend) fromApplication).configure(config, classLoader);
			}
			else {
				// keep as is!
				backend = fromApplication;
			}
		}
		else {
			// (2) check if the config defines a state backend
			final StateBackend fromConfig = loadStateBackendFromConfig(config, classLoader, logger);
			if (fromConfig != null) {
				backend = fromConfig;
			}
			else {
				// (3) use the default
				backend = new MemoryStateBackendFactory().createFromConfig(config, classLoader);
				if (logger != null) {
					logger.info("No state backend has been configured, using default (Memory / JobManager) {}", backend);
				}
			}
		}

		// to keep supporting the old behavior where default (JobManager) Backend + HA mode = checkpoints in HA store
		// we add the HA persistence dir as the checkpoint directory if none other is set

		if (backend instanceof MemoryStateBackend) {
			final MemoryStateBackend memBackend = (MemoryStateBackend) backend;

			if (memBackend.getCheckpointPath() == null && HighAvailabilityMode.isHighAvailabilityModeActivated(config)) {
				final String haStoragePath = config.getString(HighAvailabilityOptions.HA_STORAGE_PATH);

				if (haStoragePath != null) {
					try {
						Path checkpointDirPath = new Path(haStoragePath, UUID.randomUUID().toString());
						if (checkpointDirPath.toUri().getScheme() == null) {
							checkpointDirPath = checkpointDirPath.makeQualified(checkpointDirPath.getFileSystem());
						}
						Configuration tempConfig = new Configuration(config);
						tempConfig.setString(CheckpointingOptions.CHECKPOINTS_DIRECTORY, checkpointDirPath.toString());
						return memBackend.configure(tempConfig, classLoader);
					} catch (Exception ignored) {}
				}
			}
		}

		return backend;
	}
public static StateBackend fromApplicationOrConfigOrDefault ( @ Nullable StateBackend fromApplication , Configuration config , ClassLoader classLoader , @ Nullable Logger logger ) throws IllegalConfigurationException , DynamicCodeLoadingException , IOException { checkNotNull ( config , "config" ) ; checkNotNull ( classLoader , "classLoader" ) ; final StateBackend backend ; if ( fromApplication != null ) { if ( logger != null ) { logger . info ( "Using application-defined state backend: {}" , fromApplication ) ; } if ( fromApplication instanceof ConfigurableStateBackend ) { if ( logger != null ) { logger . info ( "Configuring application-defined state backend with job/cluster config" ) ; } backend = ( ( ConfigurableStateBackend ) fromApplication ) . configure ( config , classLoader ) ; } else { backend = fromApplication ; } } else { final StateBackend fromConfig = loadStateBackendFromConfig ( config , classLoader , logger ) ; if ( fromConfig != null ) { backend = fromConfig ; } else { backend = new MemoryStateBackendFactory ( ) . createFromConfig ( config , classLoader ) ; if ( logger != null ) { logger . info ( "No state backend has been configured, using default (Memory / JobManager) {}" , backend ) ; } } } if ( backend instanceof MemoryStateBackend ) { final MemoryStateBackend memBackend = ( MemoryStateBackend ) backend ; if ( memBackend . getCheckpointPath ( ) == null && HighAvailabilityMode . isHighAvailabilityModeActivated ( config ) ) { final String haStoragePath = config . getString ( HighAvailabilityOptions . HA_STORAGE_PATH ) ; if ( haStoragePath != null ) { try { Path checkpointDirPath = new Path ( haStoragePath , UUID . randomUUID ( ) . toString ( ) ) ; if ( checkpointDirPath . toUri ( ) . getScheme ( ) == null ) { checkpointDirPath = checkpointDirPath . makeQualified ( checkpointDirPath . getFileSystem ( ) ) ; } Configuration tempConfig = new Configuration ( config ) ; tempConfig . setString ( CheckpointingOptions . CHECKPOINTS_DIRECTORY , checkpointDirPath . toString ( ) ) ; return memBackend . configure ( tempConfig , classLoader ) ; } catch ( Exception ignored ) { } } } } return backend ; }
Checks if an application-defined state backend is given, and if not, loads the state
backend from the configuration, from the parameter 'state.backend', as defined
in {@link CheckpointingOptions#STATE_BACKEND}. If no state backend is configured, this instantiates the
default state backend (the {@link MemoryStateBackend}).

<p>If an application-defined state backend is found, and the state backend is a
{@link ConfigurableStateBackend}, this methods calls {@link ConfigurableStateBackend#configure(Configuration, ClassLoader)}
on the state backend.

<p>Refer to {@link #loadStateBackendFromConfig(Configuration, ClassLoader, Logger)} for details on
how the state backend is loaded from the configuration.

@param config The configuration to load the state backend from
@param classLoader The class loader that should be used to load the state backend
@param logger Optionally, a logger to log actions to (may be null)

@return The instantiated state backend.

@throws DynamicCodeLoadingException
Thrown if a state backend factory is configured and the factory class was not
found or the factory could not be instantiated
@throws IllegalConfigurationException
May be thrown by the StateBackendFactory when creating / configuring the state
backend in the factory
@throws IOException
May be thrown by the StateBackendFactory when instantiating the state backend