test_class.file: flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/SavepointStoreFactoryTest.java
test_class.identifier: SavepointStoreFactoryTest
test_case.body: @Test(expected = IllegalConfigurationException.class)
  	public void testUnexpectedSavepointBackend() throws Exception {
  		Configuration config = new Configuration();
  		config.setString(SavepointStoreFactory.SAVEPOINT_BACKEND_KEY, "unexpected");
  		SavepointStoreFactory.createFromConfig(config);
  		fail("Did not throw expected Exception");
  	}
test_case.class_method_signature: SavepointStoreFactoryTest.testUnexpectedSavepointBackend()
test_case.full_signature: @Test(expected = IllegalConfigurationException.class) public void testUnexpectedSavepointBackend()
test_case.identifier: testUnexpectedSavepointBackend
test_case.invocations: setString createFromConfig fail
test_case.modifiers: @Test(expected = IllegalConfigurationException.class) public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testUnexpectedSavepointBackend()
focal_class.declarator: SAVEPOINT_BACKEND_KEY = "savepoints.state.backend"
focal_class.modifier: public static final
focal_class.original_string: public static final String SAVEPOINT_BACKEND_KEY = "savepoints.state.backend";
focal_class.type: String
focal_class.var_name: SAVEPOINT_BACKEND_KEY
focal_class.declarator: SAVEPOINT_DIRECTORY_KEY = "savepoints.state.backend.fs.dir"
focal_class.modifier: public static final
focal_class.original_string: public static final String SAVEPOINT_DIRECTORY_KEY = "savepoints.state.backend.fs.dir";
focal_class.type: String
focal_class.var_name: SAVEPOINT_DIRECTORY_KEY
focal_class.declarator: DEFAULT_SAVEPOINT_BACKEND = "jobmanager"
focal_class.modifier: public static final
focal_class.original_string: public static final String DEFAULT_SAVEPOINT_BACKEND = "jobmanager";
focal_class.type: String
focal_class.var_name: DEFAULT_SAVEPOINT_BACKEND
focal_class.declarator: LOG = LoggerFactory.getLogger(SavepointStoreFactory.class)
focal_class.modifier: public static final
focal_class.original_string: public static final Logger LOG = LoggerFactory.getLogger(SavepointStoreFactory.class);
focal_class.type: Logger
focal_class.var_name: LOG
focal_class.file: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SavepointStoreFactory.java
focal_class.identifier: SavepointStoreFactory
focal_class.class_method_signature: SavepointStoreFactory.createFromConfig(
  			Configuration config)
focal_class.full_signature: public static SavepointStore createFromConfig(
  			Configuration config)
focal_class.identifier: createFromConfig
focal_class.modifiers: public static
focal_class.parameters: (
  			Configuration config)
focal_class.return: SavepointStore
focal_class.signature: SavepointStore createFromConfig(
  			Configuration config)
focal_class.class_method_signature: SavepointStoreFactory.createJobManagerSavepointStore()
focal_class.full_signature: private static SavepointStore createJobManagerSavepointStore()
focal_class.identifier: createJobManagerSavepointStore
focal_class.modifiers: private static
focal_class.parameters: ()
focal_class.return: SavepointStore
focal_class.signature: SavepointStore createJobManagerSavepointStore()
focal_class.class_method_signature: SavepointStoreFactory.createFileSystemSavepointStore(String rootPath)
focal_class.full_signature: private static SavepointStore createFileSystemSavepointStore(String rootPath)
focal_class.identifier: createFileSystemSavepointStore
focal_class.modifiers: private static
focal_class.parameters: (String rootPath)
focal_class.return: SavepointStore
focal_class.signature: SavepointStore createFileSystemSavepointStore(String rootPath)
focal_method.body: public static SavepointStore createFromConfig(
  			Configuration config) throws Exception {
  
  		// Try a the savepoint-specific configuration first.
  		String savepointBackend = config.getString(SAVEPOINT_BACKEND_KEY, DEFAULT_SAVEPOINT_BACKEND);
  
  		if (savepointBackend == null) {
  			LOG.info("No savepoint state backend configured. " +
  					"Using job manager savepoint state backend.");
  			return createJobManagerSavepointStore();
  		} else if (savepointBackend.equals("jobmanager")) {
  			LOG.info("Using job manager savepoint state backend.");
  			return createJobManagerSavepointStore();
  		} else if (savepointBackend.equals("filesystem")) {
  			String rootPath = config.getString(SAVEPOINT_DIRECTORY_KEY, null);
  
  			if (rootPath == null) {
  				throw new IllegalConfigurationException("Using filesystem as savepoint state backend, " +
  						"but did not specify directory. Please set the " +
  						"following configuration key: '" + SAVEPOINT_DIRECTORY_KEY +
  						"' (e.g. " + SAVEPOINT_DIRECTORY_KEY + ": hdfs:///flink/savepoints/). " +
  						"Falling back to job manager savepoint backend.");
  			} else {
  				LOG.info("Using filesystem savepoint backend (root path: {}).", rootPath);
  
  				return createFileSystemSavepointStore(rootPath);
  			}
  		} else {
  			throw new IllegalConfigurationException("Unexpected savepoint backend " +
  					"configuration '" + savepointBackend + "'. " +
  					"Falling back to job manager savepoint state backend.");
  		}
  	}
focal_method.class_method_signature: SavepointStoreFactory.createFromConfig(
  			Configuration config)
focal_method.full_signature: public static SavepointStore createFromConfig(
  			Configuration config)
focal_method.identifier: createFromConfig
focal_method.invocations: getString info createJobManagerSavepointStore equals info createJobManagerSavepointStore equals getString info createFileSystemSavepointStore
focal_method.modifiers: public static
focal_method.parameters: (
  			Configuration config)
focal_method.return: SavepointStore
focal_method.signature: SavepointStore createFromConfig(
  			Configuration config)