@Test public void testCloseTwice() throws IOException { DistributedFileSystem fs = cluster.getFileSystem(); FSDataOutputStream os = fs.create(new Path("/test")); DFSOutputStream dos = (DFSOutputStream) Whitebox.getInternalState(os, "wrappedStream"); @SuppressWarnings("unchecked") AtomicReference<IOException> ex = (AtomicReference<IOException>) Whitebox .getInternalState(dos, "lastException"); Assert.assertEquals(null, ex.get()); dos.close(); IOException dummy = new IOException("dummy"); ex.set(dummy); try { dos.close(); } catch (IOException e) { Assert.assertEquals(e, dummy); } Assert.assertEquals(null, ex.get()); dos.close(); }
@Override public void close() throws IOException { synchronized (this) { if (closed) { IOException e = lastException.getAndSet(null); if (e == null) return; else throw e; } try { flushBuffer(); if (currentPacket != null) { waitAndQueueCurrentPacket(); } if (bytesCurBlock != 0) { currentPacket = createPacket(0, 0, bytesCurBlock, currentSeqno++); currentPacket.lastPacketInBlock = true; currentPacket.syncBlock = shouldSyncBlock; } flushInternal(); ExtendedBlock lastBlock = streamer.getBlock(); closeThreads(false); completeFile(lastBlock); } catch (ClosedChannelException e) { } finally { closed = true; } } dfsClient.endFileLease(fileId); }
DFSOutputStream extends FSOutputSummer implements Syncable, CanSetDropBehind { @Override public void close() throws IOException { synchronized (this) { if (closed) { IOException e = lastException.getAndSet(null); if (e == null) return; else throw e; } try { flushBuffer(); if (currentPacket != null) { waitAndQueueCurrentPacket(); } if (bytesCurBlock != 0) { currentPacket = createPacket(0, 0, bytesCurBlock, currentSeqno++); currentPacket.lastPacketInBlock = true; currentPacket.syncBlock = shouldSyncBlock; } flushInternal(); ExtendedBlock lastBlock = streamer.getBlock(); closeThreads(false); completeFile(lastBlock); } catch (ClosedChannelException e) { } finally { closed = true; } } dfsClient.endFileLease(fileId); } }
DFSOutputStream extends FSOutputSummer implements Syncable, CanSetDropBehind { @Override public void close() throws IOException { synchronized (this) { if (closed) { IOException e = lastException.getAndSet(null); if (e == null) return; else throw e; } try { flushBuffer(); if (currentPacket != null) { waitAndQueueCurrentPacket(); } if (bytesCurBlock != 0) { currentPacket = createPacket(0, 0, bytesCurBlock, currentSeqno++); currentPacket.lastPacketInBlock = true; currentPacket.syncBlock = shouldSyncBlock; } flushInternal(); ExtendedBlock lastBlock = streamer.getBlock(); closeThreads(false); completeFile(lastBlock); } catch (ClosedChannelException e) { } finally { closed = true; } } dfsClient.endFileLease(fileId); } private  DFSOutputStream(DFSClient dfsClient, String src, Progressable progress,
      HdfsFileStatus stat, DataChecksum checksum); private  DFSOutputStream(DFSClient dfsClient, String src, HdfsFileStatus stat,
      EnumSet<CreateFlag> flag, Progressable progress,
      DataChecksum checksum, String[] favoredNodes); private  DFSOutputStream(DFSClient dfsClient, String src,
      Progressable progress, LocatedBlock lastBlock, HdfsFileStatus stat,
      DataChecksum checksum); }
DFSOutputStream extends FSOutputSummer implements Syncable, CanSetDropBehind { @Override public void close() throws IOException { synchronized (this) { if (closed) { IOException e = lastException.getAndSet(null); if (e == null) return; else throw e; } try { flushBuffer(); if (currentPacket != null) { waitAndQueueCurrentPacket(); } if (bytesCurBlock != 0) { currentPacket = createPacket(0, 0, bytesCurBlock, currentSeqno++); currentPacket.lastPacketInBlock = true; currentPacket.syncBlock = shouldSyncBlock; } flushInternal(); ExtendedBlock lastBlock = streamer.getBlock(); closeThreads(false); completeFile(lastBlock); } catch (ClosedChannelException e) { } finally { closed = true; } } dfsClient.endFileLease(fileId); } private  DFSOutputStream(DFSClient dfsClient, String src, Progressable progress,
      HdfsFileStatus stat, DataChecksum checksum); private  DFSOutputStream(DFSClient dfsClient, String src, HdfsFileStatus stat,
      EnumSet<CreateFlag> flag, Progressable progress,
      DataChecksum checksum, String[] favoredNodes); private  DFSOutputStream(DFSClient dfsClient, String src,
      Progressable progress, LocatedBlock lastBlock, HdfsFileStatus stat,
      DataChecksum checksum); @VisibleForTesting synchronized DatanodeInfo[] getPipeline(); @Override @Deprecated void sync(); @Override void hflush(); @Override void hsync(); void hsync(EnumSet<SyncFlag> syncFlags); @Deprecated synchronized int getNumCurrentReplicas(); synchronized int getCurrentBlockReplication(); @Override void close(); @VisibleForTesting void setArtificialSlowdown(long period); @VisibleForTesting synchronized void setChunksPerPacket(int value); long getInitialLen(); FileEncryptionInfo getFileEncryptionInfo(); @Override void setDropBehind(Boolean dropBehind); @VisibleForTesting long getFileId(); }
DFSOutputStream extends FSOutputSummer implements Syncable, CanSetDropBehind { @Override public void close() throws IOException { synchronized (this) { if (closed) { IOException e = lastException.getAndSet(null); if (e == null) return; else throw e; } try { flushBuffer(); if (currentPacket != null) { waitAndQueueCurrentPacket(); } if (bytesCurBlock != 0) { currentPacket = createPacket(0, 0, bytesCurBlock, currentSeqno++); currentPacket.lastPacketInBlock = true; currentPacket.syncBlock = shouldSyncBlock; } flushInternal(); ExtendedBlock lastBlock = streamer.getBlock(); closeThreads(false); completeFile(lastBlock); } catch (ClosedChannelException e) { } finally { closed = true; } } dfsClient.endFileLease(fileId); } private  DFSOutputStream(DFSClient dfsClient, String src, Progressable progress,
      HdfsFileStatus stat, DataChecksum checksum); private  DFSOutputStream(DFSClient dfsClient, String src, HdfsFileStatus stat,
      EnumSet<CreateFlag> flag, Progressable progress,
      DataChecksum checksum, String[] favoredNodes); private  DFSOutputStream(DFSClient dfsClient, String src,
      Progressable progress, LocatedBlock lastBlock, HdfsFileStatus stat,
      DataChecksum checksum); @VisibleForTesting synchronized DatanodeInfo[] getPipeline(); @Override @Deprecated void sync(); @Override void hflush(); @Override void hsync(); void hsync(EnumSet<SyncFlag> syncFlags); @Deprecated synchronized int getNumCurrentReplicas(); synchronized int getCurrentBlockReplication(); @Override void close(); @VisibleForTesting void setArtificialSlowdown(long period); @VisibleForTesting synchronized void setChunksPerPacket(int value); long getInitialLen(); FileEncryptionInfo getFileEncryptionInfo(); @Override void setDropBehind(Boolean dropBehind); @VisibleForTesting long getFileId();  }