class SavepointStoreFactory {

	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.");
		}
	}

}