GoogleHadoopSyncableOutputStream extends OutputStream implements Syncable { @Override public void close() throws IOException { logger.atFine().log( "close(): Current tail file: %s final destination: %s", curGcsPath, finalGcsPath); if (!isOpen()) { logger.atFinest().log("close(): Ignoring; stream already closed."); return; } commitCurrentFile(); curGcsPath = null; curDelegate = null; logger.atFine().log("close(): Awaiting %s deletionFutures", deletionFutures.size()); for (Future<?> deletion : deletionFutures) { try { deletion.get(); } catch (ExecutionException | InterruptedException e) { if (e instanceof InterruptedException) { Thread.currentThread().interrupt(); } throw new IOException("Failed to delete files while closing stream", e); } } } GoogleHadoopSyncableOutputStream(
      GoogleHadoopFileSystemBase ghfs,
      URI gcsPath,
      FileSystem.Statistics statistics,
      CreateFileOptions createFileOptions,
      SyncableOutputStreamOptions options); @VisibleForTesting  GoogleHadoopSyncableOutputStream(
      GoogleHadoopFileSystemBase ghfs,
      URI gcsPath,
      FileSystem.Statistics statistics,
      CreateFileOptions createFileOptions,
      SyncableOutputStreamOptions options,
      ExecutorService cleanupThreadpool); @Override void write(int b); @Override void write(byte[] b, int offset, int len); @Override void close(); void sync(); @Override void hflush(); @Override void hsync(); static final String TEMPFILE_PREFIX; }
@Test public void testCloseTwice() throws IOException { Path objectPath = new Path(ghfs.getFileSystemRoot(), "dir/object.txt"); FSDataOutputStream fout = ghfs.create(objectPath); fout.close(); fout.close(); }
@Test public void testSyncAfterClose() throws IOException { Path objectPath = new Path(ghfs.getFileSystemRoot(), "dir/object.txt"); FSDataOutputStream fout = ghfs.create(objectPath); fout.close(); assertThrows(ClosedChannelException.class, fout::hsync); }