class FsSavepointStore implements SavepointStore {

	@Override
	public <T extends Savepoint> String storeSavepoint(T savepoint) throws IOException {
		Preconditions.checkNotNull(savepoint, "Savepoint");

		Exception latestException = null;
		Path path = null;
		FSDataOutputStream fdos = null;

		// Try to create a FS output stream
		for (int attempt = 0; attempt < 10; attempt++) {
			path = new Path(rootPath, FileUtils.getRandomFilename(prefix));
			try {
				fdos = fileSystem.create(path, false);
				break;
			} catch (Exception e) {
				latestException = e;
			}
		}

		if (fdos == null) {
			throw new IOException("Failed to create file output stream at " + path, latestException);
		}

		boolean success = false;
		try (DataOutputStream dos = new DataOutputStream(fdos)) {
			// Write header
			dos.writeInt(MAGIC_NUMBER);
			dos.writeInt(savepoint.getVersion());

			// Write savepoint
			SavepointSerializer<T> serializer = SavepointSerializers.getSerializer(savepoint);
			serializer.serialize(savepoint, dos);
			success = true;
		} finally {
			if (!success && fileSystem.exists(path)) {
				if (!fileSystem.delete(path, true)) {
					LOG.warn("Failed to delete file " + path + " after failed write.");
				}
			}
		}

		return path.toString();
	}

	  FsSavepointStore(String rootPath, String prefix);

}