HDFS-5042. Completed files lost after power failure. Contributed by Vinayakumar 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;
+    }
+  }
 }

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

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) {

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.

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(

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) {

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);
 

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

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