HDFS-550. DataNode restarts may introduce corrupt/duplicated/lost replicas when handling detached replicas. Contributed by Hairong Kuang.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/hdfs/branches/HDFS-265@819746 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -77,6 +77,9 @@ Append branch (unreleased changes)
 
     HDFS-588. Fix TestFiDataTransferProtocol and TestAppend2 failures. (shv)
 
+    HDFS-550. DataNode restarts may introduce corrupt/duplicated/lost replicas
+    when handling detached replicas. (hairong)
+
 Trunk (unreleased changes)
 
   INCOMPATIBLE CHANGES

src/java/org/apache/hadoop/hdfs/server/datanode/DataStorage.java

@@ -55,6 +55,7 @@ public class DataStorage extends Storage {
   final static String COPY_FILE_PREFIX = "dncp_";
   final static String STORAGE_DIR_RBW = "rbw";
   final static String STORAGE_DIR_FINALIZED = "finalized";
+  final static String STORAGE_DIR_DETACHED = "detach";
   
   private String storageID;
 
@@ -272,6 +273,8 @@ public class DataStorage extends Storage {
     File curDir = sd.getCurrentDir();
     File prevDir = sd.getPreviousDir();
     assert curDir.exists() : "Current directory must exist.";
+    // Cleanup directory "detach"
+    cleanupDetachDir(new File(curDir, STORAGE_DIR_DETACHED));
     // delete previous dir before upgrading
     if (prevDir.exists())
       deleteDir(prevDir);
@@ -292,6 +295,30 @@ public class DataStorage extends Storage {
     LOG.info("Upgrade of " + sd.getRoot()+ " is complete.");
   }
 
+  /**
+   * Cleanup the detachDir. 
+   * 
+   * If the directory is not empty report an error; 
+   * Otherwise remove the directory.
+   * 
+   * @param detachDir detach directory
+   * @throws IOException if the directory is not empty or it can not be removed
+   */
+  private void cleanupDetachDir(File detachDir) throws IOException {
+    if (layoutVersion >= PRE_RBW_LAYOUT_VERSION &&
+        detachDir.exists() && detachDir.isDirectory() ) {
+      
+        if (detachDir.list().length != 0 ) {
+          throw new IOException("Detached directory " + detachDir +
+              " is not empty. Please manually move each file under this " +
+              "directory to the finalized directory if the finalized " +
+              "directory tree does not have the file.");
+        } else if (!detachDir.delete()) {
+          throw new IOException("Cannot remove directory " + detachDir);
+        }
+    }
+  }
+  
   void doRollback( StorageDirectory sd,
                    NamespaceInfo nsInfo
                    ) throws IOException {

src/java/org/apache/hadoop/hdfs/server/datanode/FSDataset.java

@@ -191,16 +191,48 @@ public class FSDataset implements FSConstants, FSDatasetInterface {
       return Block.GRANDFATHER_GENERATION_STAMP;
     }
 
-    void getVolumeMap(ReplicasMap volumeMap, FSVolume volume) {
+    void getVolumeMap(ReplicasMap volumeMap, FSVolume volume) 
+    throws IOException {
       if (children != null) {
         for (int i = 0; i < children.length; i++) {
           children[i].getVolumeMap(volumeMap, volume);
         }
       }
 
+      recoverTempUnlinkedBlock();
       volume.addToReplicasMap(volumeMap, dir, true);
     }
         
+    /**
+     * Recover unlinked tmp files on datanode restart. If the original block
+     * does not exist, then the tmp file is renamed to be the
+     * original file name; otherwise the tmp file is deleted.
+     */
+    private void recoverTempUnlinkedBlock() throws IOException {
+      File files[] = dir.listFiles();
+      for (File file : files) {
+        if (!FSDataset.isUnlinkTmpFile(file)) {
+          continue;
+        }
+        File blockFile = getOrigFile(file);
+        if (blockFile.exists()) {
+          //
+          // If the original block file still exists, then no recovery
+          // is needed.
+          //
+          if (!file.delete()) {
+            throw new IOException("Unable to cleanup unlinked tmp file " +
+                file);
+          }
+        } else {
+          if (!file.renameTo(blockFile)) {
+            throw new IOException("Unable to cleanup detached file " +
+                file);
+          }
+        }
+      }
+    }
+    
     /**
      * check if a data diretory is healthy
      * @throws DiskErrorException
@@ -281,7 +313,6 @@ public class FSDataset implements FSConstants, FSDatasetInterface {
     private FSDir dataDir;      // directory store Finalized replica
     private File rbwDir;        // directory store RBW replica
     private File tmpDir;        // directory store Temporary replica
-    private File detachDir; // copy on write for blocks in snapshot
     private DF usage;
     private DU dfsUsage;
     private long reserved;
@@ -293,11 +324,6 @@ public class FSDataset implements FSConstants, FSDatasetInterface {
       final File finalizedDir = new File(
           currentDir, DataStorage.STORAGE_DIR_FINALIZED);
 
-      this.detachDir = new File(parent, "detach");
-      if (detachDir.exists()) {
-        recoverDetachedBlocks(finalizedDir, detachDir);
-      }
-
       // Files that were being written when the datanode was last shutdown
       // are now moved back to the data directory. It is possible that
       // in the future, we might want to do some sort of datanode-local
@@ -322,11 +348,6 @@ public class FSDataset implements FSConstants, FSDatasetInterface {
           throw new IOException("Mkdirs failed to create " + tmpDir.toString());
         }
       }
-      if (!detachDir.mkdirs()) {
-        if (!detachDir.isDirectory()) {
-          throw new IOException("Mkdirs failed to create " + detachDir.toString());
-        }
-      }
       this.usage = new DF(parent, conf);
       this.dfsUsage = new DU(parent, conf);
       this.dfsUsage.start();
@@ -371,7 +392,7 @@ public class FSDataset implements FSConstants, FSDatasetInterface {
      */
     File createTmpFile(Block b) throws IOException {
       File f = new File(tmpDir, b.getBlockName());
-      return createTmpFile(b, f);
+      return FSDataset.createTmpFile(b, f);
     }
 
     /**
@@ -380,38 +401,9 @@ public class FSDataset implements FSConstants, FSDatasetInterface {
      */
     File createRbwFile(Block b) throws IOException {
       File f = new File(rbwDir, b.getBlockName());
-      return createTmpFile(b, f);
+      return FSDataset.createTmpFile(b, f);
     }
 
-    /**
-     * Files used for copy-on-write. They need recovery when datanode
-     * restarts.
-     */
-    File createDetachFile(Block b, String filename) throws IOException {
-      File f = new File(detachDir, filename);
-      return createTmpFile(b, f);
-    }
-
-    private File createTmpFile(Block b, File f) throws IOException {
-      if (f.exists()) {
-        throw new IOException("Unexpected problem in creating temporary file for "+
-                              b + ".  File " + f + " should not be present, but is.");
-      }
-      // Create the zero-length temp file
-      //
-      boolean fileCreated = false;
-      try {
-        fileCreated = f.createNewFile();
-      } catch (IOException ioe) {
-        throw (IOException)new IOException(DISK_ERROR +f).initCause(ioe);
-      }
-      if (!fileCreated) {
-        throw new IOException("Unexpected problem in creating temporary file for "+
-                              b + ".  File " + f + " should be creatable, but is already present.");
-      }
-      return f;
-    }
-      
     File addBlock(Block b, File f) throws IOException {
       File blockFile = dataDir.addBlock(b, f);
       File metaFile = getMetaFile( blockFile , b);
@@ -425,7 +417,7 @@ public class FSDataset implements FSConstants, FSDatasetInterface {
       DiskChecker.checkDir(rbwDir);
     }
       
-    void getVolumeMap(ReplicasMap volumeMap) {
+    void getVolumeMap(ReplicasMap volumeMap) throws IOException {
       // add finalized replicas
       dataDir.getVolumeMap(volumeMap, this);
       // add rbw replicas
@@ -542,42 +534,6 @@ public class FSDataset implements FSConstants, FSDatasetInterface {
     public String toString() {
       return getDir().getAbsolutePath();
     }
-
-    /**
-     * Recover detached files on datanode restart. If a detached block
-     * does not exist in the original directory, then it is moved to the
-     * original directory.
-     */
-    private void recoverDetachedBlocks(File dataDir, File dir) 
-                                           throws IOException {
-      File contents[] = dir.listFiles();
-      if (contents == null) {
-        return;
-      }
-      for (int i = 0; i < contents.length; i++) {
-        if (!contents[i].isFile()) {
-          throw new IOException ("Found " + contents[i] + " in " + dir +
-                                 " but it is not a file.");
-        }
-
-        //
-        // If the original block file still exists, then no recovery
-        // is needed.
-        //
-        File blk = new File(dataDir, contents[i].getName());
-        if (!blk.exists()) {
-          if (!contents[i].renameTo(blk)) {
-            throw new IOException("Unable to recover detached file " +
-                                  contents[i]);
-          }
-          continue;
-        }
-        if (!contents[i].delete()) {
-            throw new IOException("Unable to cleanup detached file " +
-                                  contents[i]);
-        }
-      }
-    }
   }
     
   static class FSVolumeSet {
@@ -640,7 +596,7 @@ public class FSDataset implements FSConstants, FSDatasetInterface {
       return remaining;
     }
       
-    synchronized void getVolumeMap(ReplicasMap volumeMap) {
+    synchronized void getVolumeMap(ReplicasMap volumeMap) throws IOException {
       for (int idx = 0; idx < volumes.length; idx++) {
         volumes[idx].getVolumeMap(volumeMap);
       }
@@ -717,8 +673,23 @@ public class FSDataset implements FSConstants, FSDatasetInterface {
   //Find better place?
   public static final String METADATA_EXTENSION = ".meta";
   public static final short METADATA_VERSION = 1;
-    
+  static final String UNLINK_BLOCK_SUFFIX = ".unlinked";
 
+  private static boolean isUnlinkTmpFile(File f) {
+    String name = f.getName();
+    return name.endsWith(UNLINK_BLOCK_SUFFIX);
+  }
+  
+  static File getUnlinkTmpFile(File f) {
+    return new File(f.getParentFile(), f.getName()+UNLINK_BLOCK_SUFFIX);
+  }
+  
+  private static File getOrigFile(File unlinkTmpFile) {
+    String fileName = unlinkTmpFile.getName();
+    return new File(unlinkTmpFile.getParentFile(),
+        fileName.substring(0, fileName.length()-UNLINK_BLOCK_SUFFIX.length()));
+  }
+  
   static String getMetaFileName(String blockFileName, long genStamp) {
     return blockFileName + "_" + genStamp + METADATA_EXTENSION;
   }
@@ -818,6 +789,26 @@ public class FSDataset implements FSConstants, FSDatasetInterface {
                                                     checksumFile.length());
   }
 
+  static File createTmpFile(Block b, File f) throws IOException {
+    if (f.exists()) {
+      throw new IOException("Unexpected problem in creating temporary file for "+
+                            b + ".  File " + f + " should not be present, but is.");
+    }
+    // Create the zero-length temp file
+    //
+    boolean fileCreated = false;
+    try {
+      fileCreated = f.createNewFile();
+    } catch (IOException ioe) {
+      throw (IOException)new IOException(DISK_ERROR +f).initCause(ioe);
+    }
+    if (!fileCreated) {
+      throw new IOException("Unexpected problem in creating temporary file for "+
+                            b + ".  File " + f + " should be creatable, but is already present.");
+    }
+    return f;
+  }
+    
   FSVolumeSet volumes;
   private int maxBlocksPerDir = 0;
   ReplicasMap volumeMap = new ReplicasMap();
@@ -953,18 +944,18 @@ public class FSDataset implements FSConstants, FSDatasetInterface {
    * snapshot. This ensures that modifying this block does not modify
    * data in any existing snapshots.
    * @param block Block
-   * @param numLinks Detach if the number of links exceed this value
+   * @param numLinks Unlink if the number of links exceed this value
    * @throws IOException
-   * @return - true if the specified block was detached or the block
+   * @return - true if the specified block was unlinked or the block
    *           is not in any snapshot.
    */
-  public boolean detachBlock(Block block, int numLinks) throws IOException {
+  public boolean unlinkBlock(Block block, int numLinks) throws IOException {
     ReplicaInfo info = null;
 
     synchronized (this) {
       info = getReplicaInfo(block);
     }
-   return info.detachBlock(numLinks);
+   return info.unlinkBlock(numLinks);
   }
 
   /** {@inheritDoc} */
@@ -1139,7 +1130,7 @@ public class FSDataset implements FSConstants, FSDatasetInterface {
   private synchronized ReplicaBeingWritten append(FinalizedReplica replicaInfo, 
       long newGS, long estimateBlockLen) throws IOException {
     // unlink the finalized replica
-    replicaInfo.detachBlock(1);
+    replicaInfo.unlinkBlock(1);
     
     // construct a RBW replica with the new GS
     File blkfile = replicaInfo.getBlockFile();
@@ -2086,7 +2077,7 @@ public class FSDataset implements FSConstants, FSDatasetInterface {
           + ", rur=" + rur);
     }
     if (rur.getNumBytes() > newlength) {
-      rur.detachBlock(1);
+      rur.unlinkBlock(1);
       truncateBlock(replicafile, rur.getMetaFile(), rur.getNumBytes(), newlength);
       // update RUR with the new length
       rur.setNumBytes(newlength);

src/java/org/apache/hadoop/hdfs/server/datanode/FinalizedReplica.java

@@ -27,7 +27,7 @@ import org.apache.hadoop.hdfs.server.datanode.FSDataset.FSVolume;
  * This class describes a replica that has been finalized.
  */
 class FinalizedReplica extends ReplicaInfo {
-  private boolean detached;      // copy-on-write done for block
+  private boolean unlinked;      // copy-on-write done for block
 
   /**
    * Constructor
@@ -58,13 +58,13 @@ class FinalizedReplica extends ReplicaInfo {
   }
   
   @Override // ReplicaInfo
-  boolean isDetached() {
-    return detached;
+  boolean isUnlinked() {
+    return unlinked;
   }
 
   @Override  // ReplicaInfo
-  void setDetached() {
-    detached = true;
+  void setUnlinked() {
+    unlinked = true;
   }
   
   @Override
@@ -90,6 +90,6 @@ class FinalizedReplica extends ReplicaInfo {
   @Override
   public String toString() {
     return super.toString()
-        + "\n  detached=" + detached;
+        + "\n  unlinked=" + unlinked;
   }
 }

src/java/org/apache/hadoop/hdfs/server/datanode/ReplicaInfo.java

@@ -129,30 +129,30 @@ abstract public class ReplicaInfo extends Block implements Replica {
   }
 
   /**
-   * check if this replica has already detached.
-   * @return true if the replica has already detached or no need to detach; 
-   *         false otherwise
+   * check if this replica has already been unlinked.
+   * @return true if the replica has already been unlinked 
+   *         or no need to be detached; false otherwise
    */
-  boolean isDetached() {
-    return true;                // no need to be detached
+  boolean isUnlinked() {
+    return true;                // no need to be unlinked
   }
 
   /**
-   * set that this replica is detached
+   * set that this replica is unlinked
    */
-  void setDetached() {
-    // no need to be detached
+  void setUnlinked() {
+    // no need to be unlinked
   }
   
    /**
    * Copy specified file into a temporary file. Then rename the
    * temporary file to the original name. This will cause any
    * hardlinks to the original file to be removed. The temporary
-   * files are created in the detachDir. The temporary files will
+   * files are created in the same directory. The temporary files will
    * be recovered (especially on Windows) on datanode restart.
    */
-  private void detachFile(File file, Block b) throws IOException {
-    File tmpFile = getVolume().createDetachFile(b, file.getName());
+  private void unlinkFile(File file, Block b) throws IOException {
+    File tmpFile = FSDataset.createTmpFile(b, FSDataset.getUnlinkTmpFile(file));
     try {
       FileInputStream in = new FileInputStream(file);
       try {
@@ -189,8 +189,8 @@ abstract public class ReplicaInfo extends Block implements Replica {
    *         false if it is already detached or no need to be detached
    * @throws IOException if there is any copy error
    */
-  boolean detachBlock(int numLinks) throws IOException {
-    if (isDetached()) {
+  boolean unlinkBlock(int numLinks) throws IOException {
+    if (isUnlinked()) {
       return false;
     }
     File file = getBlockFile();
@@ -204,12 +204,12 @@ abstract public class ReplicaInfo extends Block implements Replica {
 
     if (HardLink.getLinkCount(file) > numLinks) {
       DataNode.LOG.info("CopyOnWrite for block " + this);
-      detachFile(file, this);
+      unlinkFile(file, this);
     }
     if (HardLink.getLinkCount(meta) > numLinks) {
-      detachFile(meta, this);
+      unlinkFile(meta, this);
     }
-    setDetached();
+    setUnlinked();
     return true;
   }
 

src/java/org/apache/hadoop/hdfs/server/datanode/ReplicaUnderRecovery.java

@@ -89,13 +89,13 @@ class ReplicaUnderRecovery extends ReplicaInfo {
   }
 
   @Override //ReplicaInfo
-  boolean isDetached() {
-    return original.isDetached();
+  boolean isUnlinked() {
+    return original.isUnlinked();
   }
 
   @Override //ReplicaInfo
-  void setDetached() {
-    original.setDetached();
+  void setUnlinked() {
+    original.setUnlinked();
   }
   
   @Override //ReplicaInfo

src/java/org/apache/hadoop/hdfs/server/datanode/ReplicaWaitingToBeRecovered.java

@@ -33,7 +33,7 @@ import org.apache.hadoop.hdfs.server.datanode.FSDataset.FSVolume;
  * lease recovery.
  */
 class ReplicaWaitingToBeRecovered extends ReplicaInfo {
-  private boolean detached;      // copy-on-write done for block
+  private boolean unlinked;      // copy-on-write done for block
 
   /**
    * Constructor
@@ -64,13 +64,13 @@ class ReplicaWaitingToBeRecovered extends ReplicaInfo {
   }
   
   @Override //ReplicaInfo
-  boolean isDetached() {
-    return detached;
+  boolean isUnlinked() {
+    return unlinked;
   }
 
   @Override //ReplicaInfo
-  void setDetached() {
-    detached = true;
+  void setUnlinked() {
+    unlinked = true;
   }
   
   @Override //ReplicaInfo
@@ -96,6 +96,6 @@ class ReplicaWaitingToBeRecovered extends ReplicaInfo {
   @Override
   public String toString() {
     return super.toString()
-        + "\n  detached=" + detached;
+        + "\n  unlinked=" + unlinked;
   }
 }

src/test/hdfs/org/apache/hadoop/hdfs/TestFileAppend.java

@@ -148,7 +148,7 @@ public class TestFileAppend extends TestCase {
         Block b = blocks.get(i).getBlock();
         System.out.println("testCopyOnWrite detaching block " + b);
         assertTrue("Detaching block " + b + " should have returned true",
-            dataset.detachBlock(b, 1));
+            dataset.unlinkBlock(b, 1));
       }
 
       // Since the blocks were already detached earlier, these calls should
@@ -158,7 +158,7 @@ public class TestFileAppend extends TestCase {
         Block b = blocks.get(i).getBlock();
         System.out.println("testCopyOnWrite detaching block " + b);
         assertTrue("Detaching block " + b + " should have returned false",
-            !dataset.detachBlock(b, 1));
+            !dataset.unlinkBlock(b, 1));
       }
 
     } finally {

src/test/hdfs/org/apache/hadoop/hdfs/server/datanode/TestDatanodeRestart.java

@@ -18,8 +18,12 @@
 package org.apache.hadoop.hdfs.server.datanode;
 
 import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
 import java.io.IOException;
 import java.io.RandomAccessFile;
+import java.util.Collection;
+import java.util.Iterator;
 import java.util.Random;
 
 import org.apache.hadoop.conf.Configuration;
@@ -63,12 +67,12 @@ public class TestDatanodeRestart {
   }
   
   // test rbw replicas persist across DataNode restarts
-  @Test public void testRbwReplicas() throws IOException {
+  public void testRbwReplicas() throws IOException {
     Configuration conf = new Configuration();
     conf.setLong("dfs.block.size", 1024L);
     conf.setInt("dfs.write.packet.size", 512);
     conf.setBoolean("dfs.support.append", true);
-    MiniDFSCluster cluster = new MiniDFSCluster(conf, 1, true, null);
+    MiniDFSCluster cluster = new MiniDFSCluster(conf, 2, true, null);
     cluster.waitActive();
     try {
       testRbwReplicas(cluster, false);
@@ -81,13 +85,13 @@ public class TestDatanodeRestart {
   private void testRbwReplicas(MiniDFSCluster cluster, boolean isCorrupt) 
   throws IOException {
     FSDataOutputStream out = null;
+    FileSystem fs = cluster.getFileSystem();
+    final Path src = new Path("/test.txt");
     try {
-      FileSystem fs = cluster.getFileSystem();
       final int fileLen = 515;
       // create some rbw replicas on disk
       byte[] writeBuf = new byte[fileLen];
       new Random().nextBytes(writeBuf);
-      final Path src = new Path("/test.txt");
       out = fs.create(src);
       out.write(writeBuf);
       out.sync();
@@ -116,9 +120,85 @@ public class TestDatanodeRestart {
         Assert.assertEquals(fileLen, replica.getNumBytes());
       }
       dn.data.invalidate(new Block[]{replica});
-      fs.delete(src, false);
     } finally {
       IOUtils.closeStream(out);
+      if (fs.exists(src)) {
+        fs.delete(src, false);
+      }
+      fs.close();
     }      
   }
+
+  // test recovering unlinked tmp replicas
+  @Test public void testRecoverReplicas() throws IOException {
+    Configuration conf = new Configuration();
+    conf.setLong("dfs.block.size", 1024L);
+    conf.setInt("dfs.write.packet.size", 512);
+    conf.setBoolean("dfs.support.append", true);
+    MiniDFSCluster cluster = new MiniDFSCluster(conf, 1, true, null);
+    cluster.waitActive();
+    try {
+      FileSystem fs = cluster.getFileSystem();
+      for (int i=0; i<4; i++) {
+        Path fileName = new Path("/test"+i);
+        DFSTestUtil.createFile(fs, fileName, 1, (short)1, 0L);
+        DFSTestUtil.waitReplication(fs, fileName, (short)1);
+      }
+      DataNode dn = cluster.getDataNodes().get(0);
+      Iterator<ReplicaInfo> replicasItor = 
+        ((FSDataset)dn.data).volumeMap.replicas().iterator();
+      ReplicaInfo replica = replicasItor.next();
+      createUnlinkTmpFile(replica, true, true); // rename block file
+      createUnlinkTmpFile(replica, false, true); // rename meta file
+      replica = replicasItor.next();
+      createUnlinkTmpFile(replica, true, false); // copy block file
+      createUnlinkTmpFile(replica, false, false); // copy meta file
+      replica = replicasItor.next();
+      createUnlinkTmpFile(replica, true, true); // rename block file
+      createUnlinkTmpFile(replica, false, false); // copy meta file
+
+      cluster.restartDataNodes();
+      cluster.waitActive();
+      dn = cluster.getDataNodes().get(0);
+
+      // check volumeMap: 4 finalized replica
+      Collection<ReplicaInfo> replicas = 
+        ((FSDataset)(dn.data)).volumeMap.replicas();
+      Assert.assertEquals(4, replicas.size());
+      replicasItor = replicas.iterator();
+      while (replicasItor.hasNext()) {
+        Assert.assertEquals(ReplicaState.FINALIZED, 
+            replicasItor.next().getState());
+      }
+    } finally {
+      cluster.shutdown();
+    }
+  }
+
+  private static void createUnlinkTmpFile(ReplicaInfo replicaInfo, 
+      boolean changeBlockFile, 
+      boolean isRename) throws IOException {
+    File src;
+    if (changeBlockFile) {
+      src = replicaInfo.getBlockFile();
+    } else {
+      src = replicaInfo.getMetaFile();
+    }
+    File dst = FSDataset.getUnlinkTmpFile(src);
+    if (isRename) {
+      src.renameTo(dst);
+    } else {
+      FileInputStream in = new FileInputStream(src);
+      try {
+        FileOutputStream out = new FileOutputStream(dst);
+        try {
+          IOUtils.copyBytes(in, out, 1);
+        } finally {
+          out.close();
+        }
+      } finally {
+        in.close();
+      }
+    }
+  }
 }