Repository: hadoop Updated Branches: refs/heads/branch-2.7 38bc12c68 -> 01cdea732
HDFS-5042. Completed files lost after power failure. Contributed by Vinayakumar B. Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/01cdea73 Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/01cdea73 Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/01cdea73 Branch: refs/heads/branch-2.7 Commit: 01cdea732f3294d1b30ec89df4ad4477754fe53d Parents: 38bc12c Author: Kihwal Lee <[email protected]> Authored: Thu Jun 1 12:41:00 2017 -0500 Committer: Kihwal Lee <[email protected]> Committed: Thu Jun 1 12:41:00 2017 -0500 ---------------------------------------------------------------------- .../main/java/org/apache/hadoop/io/IOUtils.java | 55 +++++++++++++++++++- hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt | 2 + .../hdfs/server/datanode/BlockReceiver.java | 9 +++- .../server/datanode/fsdataset/FsDatasetSpi.java | 4 +- .../datanode/fsdataset/impl/FsDatasetImpl.java | 36 ++++++++++--- .../server/datanode/SimulatedFSDataset.java | 3 +- .../datanode/TestDataNodeHotSwapVolumes.java | 6 ++- .../server/datanode/TestSimulatedFSDataset.java | 4 +- .../extdataset/ExternalDatasetImpl.java | 3 +- 9 files changed, 105 insertions(+), 17 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hadoop/blob/01cdea73/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/IOUtils.java ---------------------------------------------------------------------- diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/IOUtils.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/IOUtils.java index d4d5862..25c5f11 100644 --- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/IOUtils.java +++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/IOUtils.java @@ -27,6 +27,7 @@ import java.nio.file.DirectoryStream; import java.nio.file.DirectoryIteratorException; import java.nio.file.Files; import java.nio.file.Path; +import java.nio.file.StandardOpenOption; import java.util.ArrayList; import java.util.List; @@ -35,7 +36,7 @@ import org.apache.commons.logging.Log; import org.apache.hadoop.classification.InterfaceAudience; import org.apache.hadoop.classification.InterfaceStability; import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.util.ChunkedArrayList; +import org.apache.hadoop.util.Shell; /** * An utility class for I/O related functionality. @@ -346,4 +347,56 @@ public class IOUtils { } return list; } + + /** + * Ensure that any writes to the given file is written to the storage device + * that contains it. This method opens channel on given File and closes it + * once the sync is done.<br> + * Borrowed from Uwe Schindler in LUCENE-5588 + * @param fileToSync the file to fsync + */ + public static void fsync(File fileToSync) throws IOException { + if (!fileToSync.exists()) { + throw new FileNotFoundException( + "File/Directory " + fileToSync.getAbsolutePath() + " does not exist"); + } + boolean isDir = fileToSync.isDirectory(); + // If the file is a directory we have to open read-only, for regular files + // we must open r/w for the fsync to have an effect. See + // http://blog.httrack.com/blog/2013/11/15/ + // everything-you-always-wanted-to-know-about-fsync/ + try(FileChannel channel = FileChannel.open(fileToSync.toPath(), + isDir ? StandardOpenOption.READ : StandardOpenOption.WRITE)){ + fsync(channel, isDir); + } + } + + /** + * Ensure that any writes to the given file is written to the storage device + * that contains it. This method opens channel on given File and closes it + * once the sync is done. + * Borrowed from Uwe Schindler in LUCENE-5588 + * @param channel Channel to sync + * @param isDir if true, the given file is a directory (Channel should be + * opened for read and ignore IOExceptions, because not all file + * systems and operating systems allow to fsync on a directory) + * @throws IOException + */ + public static void fsync(FileChannel channel, boolean isDir) + throws IOException { + try { + channel.force(true); + } catch (IOException ioe) { + if (isDir) { + assert !(Shell.LINUX + || Shell.MAC) : "On Linux and MacOSX fsyncing a directory" + + " should not throw IOException, we just don't want to rely" + + " on that in production (undocumented)" + ". Got: " + ioe; + // Ignore exception if it is a directory + return; + } + // Throw original exception + throw ioe; + } + } } http://git-wip-us.apache.org/repos/asf/hadoop/blob/01cdea73/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt b/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt index 8b032eb..bac3499 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt +++ b/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt @@ -338,6 +338,8 @@ Release 2.7.4 - UNRELEASED HDFS-11893. Fix TestDFSShell.testMoveWithTargetPortEmpty failure. (Contributed by Brahma Reddy Battula) + HDFS-5042. Completed files lost after power failure. (vinayakumarb via kihwal) + Release 2.7.3 - 2016-08-25 INCOMPATIBLE CHANGES http://git-wip-us.apache.org/repos/asf/hadoop/blob/01cdea73/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/BlockReceiver.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/BlockReceiver.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/BlockReceiver.java index a961bd6..45b6ce2 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/BlockReceiver.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/BlockReceiver.java @@ -122,6 +122,7 @@ class BlockReceiver implements Closeable { private final boolean isTransfer; private boolean syncOnClose; + private volatile boolean dirSyncOnFinalize; private long restartBudget; /** the reference of the volume where the block receiver writes to */ private ReplicaHandler replicaHandler; @@ -528,6 +529,9 @@ class BlockReceiver implements Closeable { // avoid double sync'ing on close if (syncBlock && lastPacketInBlock) { this.syncOnClose = false; + // sync directory for finalize irrespective of syncOnClose config since + // sync is requested. + this.dirSyncOnFinalize = true; } // update received bytes @@ -879,6 +883,7 @@ class BlockReceiver implements Closeable { boolean isReplaceBlock) throws IOException { syncOnClose = datanode.getDnConf().syncOnClose; + dirSyncOnFinalize = syncOnClose; boolean responderClosed = false; mirrorOut = mirrOut; mirrorAddr = mirrAddr; @@ -920,7 +925,7 @@ class BlockReceiver implements Closeable { } else { // for isDatnode or TRANSFER_FINALIZED // Finalize the block. - datanode.data.finalizeBlock(block); + datanode.data.finalizeBlock(block, dirSyncOnFinalize); } } datanode.metrics.incrBlocksWritten(); @@ -1418,7 +1423,7 @@ class BlockReceiver implements Closeable { BlockReceiver.this.close(); endTime = ClientTraceLog.isInfoEnabled() ? System.nanoTime() : 0; block.setNumBytes(replicaInfo.getNumBytes()); - datanode.data.finalizeBlock(block); + datanode.data.finalizeBlock(block, dirSyncOnFinalize); } if (pinning) { http://git-wip-us.apache.org/repos/asf/hadoop/blob/01cdea73/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/FsDatasetSpi.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/FsDatasetSpi.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/FsDatasetSpi.java index 1181ad6..8378725 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/FsDatasetSpi.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/FsDatasetSpi.java @@ -292,12 +292,14 @@ public interface FsDatasetSpi<V extends FsVolumeSpi> extends FSDatasetMBean { * Finalizes the block previously opened for writing using writeToBlock. * The block size is what is in the parameter b and it must match the amount * of data written + * @param block Block to be finalized + * @param fsyncDir whether to sync the directory changes to durable device. * @throws IOException * @throws ReplicaNotFoundException if the replica can not be found when the * block is been finalized. For instance, the block resides on an HDFS volume * that has been removed. */ - public void finalizeBlock(ExtendedBlock b) throws IOException; + void finalizeBlock(ExtendedBlock b, boolean fsyncDir) throws IOException; /** * Unfinalizes the block previously opened for writing using writeToBlock. http://git-wip-us.apache.org/repos/asf/hadoop/blob/01cdea73/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsDatasetImpl.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsDatasetImpl.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsDatasetImpl.java index f92e2a5..76867b1 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsDatasetImpl.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/fsdataset/impl/FsDatasetImpl.java @@ -825,6 +825,17 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> { return dstfile; } + private void fsyncDirectory(File... dirs) + throws IOException { + for (File dir : dirs) { + try { + IOUtils.fsync(dir); + } catch (IOException e) { + throw new IOException("Failed to sync " + dir, e); + } + } + } + /** * Copy the block and meta files for the given block to the given destination. * @return the new meta and block files. @@ -917,7 +928,8 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> { targetVolume, blockFiles[0].getParentFile(), 0); newReplicaInfo.setNumBytes(blockFiles[1].length()); // Finalize the copied files - newReplicaInfo = finalizeReplica(block.getBlockPoolId(), newReplicaInfo); + newReplicaInfo = finalizeReplica(block.getBlockPoolId(), newReplicaInfo, + false); removeOldReplica(replicaInfo, newReplicaInfo, oldBlockFile, oldMetaFile, oldBlockFile.length(), oldMetaFile.length(), block.getBlockPoolId()); @@ -1242,7 +1254,7 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> { bumpReplicaGS(replicaInfo, newGS); // finalize the replica if RBW if (replicaInfo.getState() == ReplicaState.RBW) { - finalizeReplica(b.getBlockPoolId(), replicaInfo); + finalizeReplica(b.getBlockPoolId(), replicaInfo, false); } return replicaInfo; } @@ -1539,7 +1551,8 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> { * Complete the block write! */ @Override // FsDatasetSpi - public synchronized void finalizeBlock(ExtendedBlock b) throws IOException { + public synchronized void finalizeBlock(ExtendedBlock b, boolean fsyncDir) + throws IOException { if (Thread.interrupted()) { // Don't allow data modifications from interrupted threads throw new IOException("Cannot finalize block from Interrupted Thread"); @@ -1550,11 +1563,11 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> { // been opened for append but never modified return; } - finalizeReplica(b.getBlockPoolId(), replicaInfo); + finalizeReplica(b.getBlockPoolId(), replicaInfo, fsyncDir); } private synchronized FinalizedReplica finalizeReplica(String bpid, - ReplicaInfo replicaInfo) throws IOException { + ReplicaInfo replicaInfo, boolean fsyncDir) throws IOException { FinalizedReplica newReplicaInfo = null; if (replicaInfo.getState() == ReplicaState.RUR && ((ReplicaUnderRecovery)replicaInfo).getOriginalReplica().getState() == @@ -1573,6 +1586,15 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> { bpid, replicaInfo, f, replicaInfo.getBytesReserved()); newReplicaInfo = new FinalizedReplica(replicaInfo, v, dest.getParentFile()); + /* + * Sync the directory after rename from tmp/rbw to Finalized if + * configured. Though rename should be atomic operation, sync on both + * dest and src directories are done because IOUtils.fsync() calls + * directory's channel sync, not the journal itself. + */ + if (fsyncDir) { + fsyncDirectory(dest.getParentFile(), f.getParentFile()); + } if (v.isTransientStorage()) { ramDiskReplicaTracker.addReplica(bpid, replicaInfo.getBlockId(), v); datanode.getMetrics().addRamDiskBytesWrite(replicaInfo.getNumBytes()); @@ -2486,12 +2508,12 @@ class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl> { // but it is immediately converted to finalized state within the same // lock, so no need to update it. volumeMap.add(bpid, newReplicaInfo); - finalizeReplica(bpid, newReplicaInfo); + finalizeReplica(bpid, newReplicaInfo, false); } } // finalize the block - return finalizeReplica(bpid, rur); + return finalizeReplica(bpid, rur, false); } private File[] copyReplicaWithNewBlockIdAndGS( http://git-wip-us.apache.org/repos/asf/hadoop/blob/01cdea73/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/SimulatedFSDataset.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/SimulatedFSDataset.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/SimulatedFSDataset.java index 99bd3ae..7c40bdd 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/SimulatedFSDataset.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/SimulatedFSDataset.java @@ -580,7 +580,8 @@ public class SimulatedFSDataset implements FsDatasetSpi<FsVolumeSpi> { } @Override // FsDatasetSpi - public synchronized void finalizeBlock(ExtendedBlock b) throws IOException { + public synchronized void finalizeBlock(ExtendedBlock b, boolean fsyncDir) + throws IOException { final Map<Block, BInfo> map = getMap(b.getBlockPoolId()); BInfo binfo = map.get(b.getLocalBlock()); if (binfo == null) { http://git-wip-us.apache.org/repos/asf/hadoop/blob/01cdea73/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeHotSwapVolumes.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeHotSwapVolumes.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeHotSwapVolumes.java index 48b2e5a..e5583a0 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeHotSwapVolumes.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataNodeHotSwapVolumes.java @@ -633,10 +633,12 @@ public class TestDataNodeHotSwapVolumes { // Bypass the argument to FsDatasetImpl#finalizeBlock to verify that // the block is not removed, since the volume reference should not // be released at this point. - data.finalizeBlock((ExtendedBlock) invocation.getArguments()[0]); + data.finalizeBlock((ExtendedBlock) invocation.getArguments()[0], + (boolean) invocation.getArguments()[1]); return null; } - }).when(dn.data).finalizeBlock(any(ExtendedBlock.class)); + }).when(dn.data).finalizeBlock(any(ExtendedBlock.class), + Mockito.anyBoolean()); final CyclicBarrier barrier = new CyclicBarrier(2); http://git-wip-us.apache.org/repos/asf/hadoop/blob/01cdea73/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestSimulatedFSDataset.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestSimulatedFSDataset.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestSimulatedFSDataset.java index dd24685..62d378f 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestSimulatedFSDataset.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestSimulatedFSDataset.java @@ -82,7 +82,7 @@ public class TestSimulatedFSDataset { out.close(); } b.setNumBytes(blockIdToLen(i)); - fsdataset.finalizeBlock(b); + fsdataset.finalizeBlock(b, false); assertEquals(blockIdToLen(i), fsdataset.getLength(b)); } return bytesAdded; @@ -272,7 +272,7 @@ public class TestSimulatedFSDataset { } try { - fsdataset.finalizeBlock(b); + fsdataset.finalizeBlock(b, false); assertTrue("Expected an IO exception", false); } catch (IOException e) { // ok - as expected http://git-wip-us.apache.org/repos/asf/hadoop/blob/01cdea73/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/extdataset/ExternalDatasetImpl.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/extdataset/ExternalDatasetImpl.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/extdataset/ExternalDatasetImpl.java index 6593c22..23072ce 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/extdataset/ExternalDatasetImpl.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/extdataset/ExternalDatasetImpl.java @@ -183,7 +183,8 @@ public class ExternalDatasetImpl implements FsDatasetSpi<ExternalVolumeImpl> { } @Override - public void finalizeBlock(ExtendedBlock b) throws IOException { + public void finalizeBlock(ExtendedBlock b, boolean fsyncDir) + throws IOException { } @Override --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
