flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/main/java/org/apache/flink/runtime/state/AbstractStateBackend.java
/**
	 * Loads the state backend from the configuration, from the parameter 'state.backend', as defined
	 * in {@link CoreOptions#STATE_BACKEND}.
	 * 
	 * <p>The state backends can be specified either via their shortcut name, or via the class name
	 * of a {@link StateBackendFactory}. If a StateBackendFactory class name is specified, the factory
	 * is instantiated (via its zero-argument constructor) and its
	 * {@link StateBackendFactory#createFromConfig(Configuration)} method is called.
	 *
	 * <p>Recognized shortcut names are '{@value AbstractStateBackend#MEMORY_STATE_BACKEND_NAME}',
	 * '{@value AbstractStateBackend#FS_STATE_BACKEND_NAME}', and
	 * '{@value AbstractStateBackend#ROCKSDB_STATE_BACKEND_NAME}'.
	 * 
	 * @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
	 */
Loads the state backend from the configuration, from the parameter 'state.backend', as defined
in  CoreOptions#STATE_BACKEND.

The state backends can be specified either via their shortcut name, or via the class name
of a  StateBackendFactory. If a StateBackendFactory class name is specified, the factory
is instantiated (via its zero-argument constructor) and its
StateBackendFactory#createFromConfig(Configuration) method is called.



@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
Loads the state backend from the configuration from the parameter ' state . backend ' as defined in CoreOptions#STATE_BACKEND . The state backends can be specified either via their shortcut name or via the class name of a StateBackendFactory . If a StateBackendFactory class name is specified the factory is instantiated ( via its zero - argument constructor ) and its StateBackendFactory#createFromConfig ( Configuration ) method is called . @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
public static StateBackend loadStateBackendFromConfig(
			Configuration config,
			ClassLoader classLoader,
			@Nullable Logger logger) throws IllegalConfigurationException, DynamicCodeLoadingException, IOException {
		checkNotNull(config, "config");
		checkNotNull(classLoader, "classLoader");
		final String backendName = config.getString(CoreOptions.STATE_BACKEND);
		if (backendName == null) {
			return null;
		}
		String factoryClassName = backendName;
		switch (backendName.toLowerCase()) {
			case MEMORY_STATE_BACKEND_NAME:
				if (logger != null) {
					logger.info("State backend is set to heap memory (checkpoint to JobManager)");
				}
				return new MemoryStateBackend();
			case FS_STATE_BACKEND_NAME:
				FsStateBackend fsBackend = new FsStateBackendFactory().createFromConfig(config);
				if (logger != null) {
					logger.info("State backend is set to heap memory (checkpoints to filesystem \"{}\")",
							fsBackend.getBasePath());
				}
				return fsBackend;
			case ROCKSDB_STATE_BACKEND_NAME:
				factoryClassName = "org.apache.flink.contrib.streaming.state.RocksDBStateBackendFactory";
			default:
				if (logger != null) {
					logger.info("Loading state backend via factory {}", factoryClassName);
				}
				StateBackendFactory<?> factory;
				try {
					@SuppressWarnings("rawtypes")
					Class<? extends StateBackendFactory> clazz = 
							Class.forName(factoryClassName, false, classLoader)
									.asSubclass(StateBackendFactory.class);
					factory = clazz.newInstance();
				}
				catch (ClassNotFoundException e) {
					throw new DynamicCodeLoadingException(
							"Cannot find configured state backend factory class: " + backendName, e);
				}
				catch (ClassCastException | InstantiationException | IllegalAccessException e) {
					throw new DynamicCodeLoadingException("The class configured under '" +
							CoreOptions.STATE_BACKEND.key() + "' is not a valid state backend factory (" +
							backendName + ')', e);
				}
				return factory.createFromConfig(config);
		}
	}
public static StateBackend loadStateBackendFromConfig ( Configuration config , ClassLoader classLoader , @ Nullable Logger logger ) throws IllegalConfigurationException , DynamicCodeLoadingException , IOException { checkNotNull ( config , "config" ) ; checkNotNull ( classLoader , "classLoader" ) ; final String backendName = config . getString ( CoreOptions . STATE_BACKEND ) ; if ( backendName == null ) { return null ; } String factoryClassName = backendName ; switch ( backendName . toLowerCase ( ) ) { case MEMORY_STATE_BACKEND_NAME : if ( logger != null ) { logger . info ( "State backend is set to heap memory (checkpoint to JobManager)" ) ; } return new MemoryStateBackend ( ) ; case FS_STATE_BACKEND_NAME : FsStateBackend fsBackend = new FsStateBackendFactory ( ) . createFromConfig ( config ) ; if ( logger != null ) { logger . info ( "State backend is set to heap memory (checkpoints to filesystem \"{}\")" , fsBackend . getBasePath ( ) ) ; } return fsBackend ; case ROCKSDB_STATE_BACKEND_NAME : factoryClassName = "org.apache.flink.contrib.streaming.state.RocksDBStateBackendFactory" ; default : if ( logger != null ) { logger . info ( "Loading state backend via factory {}" , factoryClassName ) ; } StateBackendFactory < ? > factory ; try { @ SuppressWarnings ( "rawtypes" ) Class < ? extends StateBackendFactory > clazz = Class . forName ( factoryClassName , false , classLoader ) . asSubclass ( StateBackendFactory . class ) ; factory = clazz . newInstance ( ) ; } catch ( ClassNotFoundException e ) { throw new DynamicCodeLoadingException ( "Cannot find configured state backend factory class: " + backendName , e ) ; } catch ( ClassCastException | InstantiationException | IllegalAccessException e ) { throw new DynamicCodeLoadingException ( "The class configured under '" + CoreOptions . STATE_BACKEND . key ( ) + "' is not a valid state backend factory (" + backendName + ')' , e ) ; } return factory . createFromConfig ( config ) ; } }
Loads the state backend from the configuration, from the parameter 'state.backend', as defined
 in {@link CoreOptions#STATE_BACKEND}.
Loads the state backend from the configuration from the parameter ' state . backend ' as defined in { @link CoreOptions#STATE_BACKEND } .
// by default the factory class is the backend name  // fall through to the 'default' case that uses reflection to load the backend // that way we can keep RocksDB in a separate module