SavepointStoreFactory { public static SavepointStore createFromConfig( Configuration config) throws Exception { 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: "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."); } }  static SavepointStore createFromConfig(
			Configuration config); static final String SAVEPOINT_BACKEND_KEY; static final String SAVEPOINT_DIRECTORY_KEY; static final String DEFAULT_SAVEPOINT_BACKEND; static final Logger LOG; }
@Test public void testStateStoreWithDefaultConfig() throws Exception { SavepointStore store = SavepointStoreFactory.createFromConfig(new Configuration()); assertTrue(store.getStateStore() instanceof HeapStateStore); }
@Test public void testSavepointBackendJobManager() throws Exception { Configuration config = new Configuration(); config.setString(SavepointStoreFactory.SAVEPOINT_BACKEND_KEY, "jobmanager"); SavepointStore store = SavepointStoreFactory.createFromConfig(config); assertTrue(store.getStateStore() instanceof HeapStateStore); }
@Test public void testSavepointBackendFileSystem() throws Exception { Configuration config = new Configuration(); String rootPath = System.getProperty("java.io.tmpdir"); config.setString(ConfigConstants.STATE_BACKEND, "filesystem"); config.setString(SavepointStoreFactory.SAVEPOINT_BACKEND_KEY, "filesystem"); config.setString(SavepointStoreFactory.SAVEPOINT_DIRECTORY_KEY, rootPath); SavepointStore store = SavepointStoreFactory.createFromConfig(config); assertTrue(store.getStateStore() instanceof FileSystemStateStore); FileSystemStateStore<CompletedCheckpoint> stateStore = (FileSystemStateStore<CompletedCheckpoint>) store.getStateStore(); assertEquals(new Path(rootPath), stateStore.getRootPath()); }
@Test public void testSavepointBackendFileSystemButCheckpointBackendJobManager() throws Exception { Configuration config = new Configuration(); String rootPath = System.getProperty("java.io.tmpdir"); config.setString(ConfigConstants.STATE_BACKEND, "jobmanager"); config.setString(SavepointStoreFactory.SAVEPOINT_BACKEND_KEY, "filesystem"); config.setString(SavepointStoreFactory.SAVEPOINT_DIRECTORY_KEY, rootPath); SavepointStore store = SavepointStoreFactory.createFromConfig(config); assertTrue(store.getStateStore() instanceof FileSystemStateStore); FileSystemStateStore<CompletedCheckpoint> stateStore = (FileSystemStateStore<CompletedCheckpoint>) store.getStateStore(); assertEquals(new Path(rootPath), stateStore.getRootPath()); }
@Test(expected = IllegalConfigurationException.class) public void testSavepointBackendFileSystemButNoDirectory() throws Exception { Configuration config = new Configuration(); config.setString(SavepointStoreFactory.SAVEPOINT_BACKEND_KEY, "filesystem"); SavepointStoreFactory.createFromConfig(config); fail("Did not throw expected Exception"); }
@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"); }