class FixFileFsStateOutputStream extends CheckpointStateOutputStream {

	@Override
	public FileStateHandle closeAndGetHandle() throws IOException {
		synchronized (this) {
			if (!closed) {
				try {
					// make a best effort attempt to figure out the size
					long size = 0;
					try {
						size = out.getPos();
					} catch (Exception ignored) {}

					// close and return
					out.close();

					return new FileStateHandle(path, size);
				}
				catch (Exception e) {
					try {
						fileSystem.delete(path, false);
					}
					catch (Exception deleteException) {
						LOG.warn("Could not delete the checkpoint stream file {}.", path, deleteException);
					}

					throw new IOException("Could not flush and close the file system " +
							"output stream to " + path + " in order to obtain the " +
							"stream state handle", e);
				}
				finally {
					closed = true;
				}
			}
			else {
				throw new IOException("Stream has already been closed and discarded.");
			}
		}
	}

	public  FixFileFsStateOutputStream(FileSystem fileSystem, Path path);

}