HADOOP-3113. An fsync invoked on a HDFS file really really persists data! The datanode

moves blocks in the tmp directory to the real block directory on a datanode-restart.
(dhruba)



git-svn-id: https://svn.apache.org/repos/asf/hadoop/core/trunk@663328 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -74,6 +74,10 @@ Trunk (unreleased changes)
     HADOOP-3459. Change in the output format of dfs -ls to more closely match
     /bin/ls. New format is: perm repl owner group size date name (Mukund Madhugiri via omally)
 
+    HADOOP-3113. An fsync invoked on a HDFS file really really persists data! The datanode
+    moves blocks in the tmp directory to the real block directory on a datanode-restart.
+    (dhruba)
+
   NEW FEATURES
 
     HADOOP-3074. Provides a UrlStreamHandler for DFS and other FS,

src/java/org/apache/hadoop/dfs/DataNode.java

@@ -3066,11 +3066,19 @@ public class DataNode extends Configured
   }
 
   /** {@inheritDoc} */
-  public void updateBlock(Block oldblock, Block newblock) throws IOException {
+  public void updateBlock(Block oldblock, Block newblock, boolean finalize) throws IOException {
     if (LOG.isDebugEnabled()) {
       LOG.debug("oldblock=" + oldblock + ", newblock=" + newblock);
     }
     data.updateBlock(oldblock, newblock);
+    if (finalize) {
+      data.finalizeBlock(newblock);
+      myMetrics.blocksWritten.inc(); 
+      notifyNamenodeReceivedBlock(newblock, EMPTY_DEL_HINT);
+      LOG.info("Received block " + newblock +
+                " of size " + newblock.getNumBytes() +
+                " as part of lease recovery.");
+    }
   }
 
   /** {@inheritDoc} */

src/java/org/apache/hadoop/dfs/FSDataset.java

@@ -299,11 +299,17 @@ class FSDataset implements FSConstants, FSDatasetInterface {
       if (detachDir.exists()) {
         recoverDetachedBlocks(currentDir, detachDir);
       }
-      this.dataDir = new FSDir(currentDir);
+
+      // 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
+      // recovery for these blocks. For example, crc validation.
+      //
       this.tmpDir = new File(parent, "tmp");
       if (tmpDir.exists()) {
-        FileUtil.fullyDelete(tmpDir);
+        recoverDetachedBlocks(currentDir, tmpDir);
       }
+      this.dataDir = new FSDir(currentDir);
       if (!tmpDir.mkdirs()) {
         if (!tmpDir.isDirectory()) {
           throw new IOException("Mkdirs failed to create " + tmpDir.toString());
@@ -651,8 +657,6 @@ class FSDataset implements FSConstants, FSDatasetInterface {
   private HashMap<Block,DatanodeBlockInfo> volumeMap = null;
   static  Random random = new Random();
   
-  long blockWriteTimeout = 3600 * 1000;
-  
   /**
    * An FSDataset has a directory where it loads its data files.
    */
@@ -665,8 +669,6 @@ class FSDataset implements FSConstants, FSDatasetInterface {
     volumes = new FSVolumeSet(volArray);
     volumeMap = new HashMap<Block, DatanodeBlockInfo>();
     volumes.getVolumeMap(volumeMap);
-    blockWriteTimeout = Math.max(
-         conf.getInt("dfs.datanode.block.write.timeout.sec", 3600), 1) * 1000L;
     registerMBean(storage.getStorageID());
   }
 
@@ -910,18 +912,8 @@ class FSDataset implements FSConstants, FSDatasetInterface {
         threads = activeFile.threads;
         
         if (!isRecovery) {
-          // check how old is the temp file - wait 1 hour
-          if ((System.currentTimeMillis() - f.lastModified()) < 
-              blockWriteTimeout) {
-            throw new IOException("Block " + b +
+          throw new IOException("Block " + b +
                                   " has already been started (though not completed), and thus cannot be created.");
-          } else {
-            // stale temp file - remove
-            if (!f.delete()) {
-              throw new IOException("Can't write the block - unable to remove stale temp file " + f);
-            }
-            f = null;
-          }
         } else {
           for (Thread thread:threads) {
             thread.interrupt();
@@ -1016,7 +1008,11 @@ class FSDataset implements FSConstants, FSDatasetInterface {
    * Complete the block write!
    */
   public synchronized void finalizeBlock(Block b) throws IOException {
-    File f = ongoingCreates.get(b).file;
+    ActiveFile activeFile = ongoingCreates.get(b);
+    if (activeFile == null) {
+      throw new IOException("Block " + b + " is already finalized.");
+    }
+    File f = activeFile.file;
     if (f == null || !f.exists()) {
       throw new IOException("No temporary file " + f + " for block " + b);
     }

src/java/org/apache/hadoop/dfs/FSNamesystem.java

@@ -1118,7 +1118,7 @@ class FSNamesystem implements FSConstants, FSNamesystemMBean {
 
       // allocate new block record block locations in INode.
       newBlock = allocateBlock(src, pendingFile);
-      pendingFile.targets = targets;
+      pendingFile.setTargets(targets);
     }
         
     // Create next block
@@ -1638,8 +1638,31 @@ class FSNamesystem implements FSConstants, FSNamesystemMBean {
 
     INodeFileUnderConstruction pendingFile = (INodeFileUnderConstruction) iFile;
 
-    // Initialize lease recovery for pendingFile
+    // Initialize lease recovery for pendingFile. If there are no blocks 
+    // associated with this file, then reap lease immediately. Otherwise 
+    // renew the lease and trigger lease recovery.
+    if (pendingFile.getTargets().length == 0) {
+      if (pendingFile.getBlocks().length == 0) {
+        finalizeINodeFileUnderConstruction(src, pendingFile);
+        NameNode.stateChangeLog.warn("BLOCK*"
+          + " internalReleaseLease: No blocks found, lease removed.");
+        return;
+      }
+      // setup the Inode.targets for the last block from the blocksMap
+      //
+      Block[] blocks = pendingFile.getBlocks();
+      Block last = blocks[blocks.length-1];
+      DatanodeDescriptor[] targets = 
+         new DatanodeDescriptor[blocksMap.numNodes(last)];
+      Iterator<DatanodeDescriptor> it = blocksMap.nodeIterator(last);
+      for (int i = 0; it != null && it.hasNext(); i++) {
+        targets[i] = it.next();
+      }
+      pendingFile.setTargets(targets);
+    }
+    // start lease recovery of the last block for this file.
     pendingFile.assignPrimaryDatanode();
+    leaseManager.renewLease(lease);
   }
 
   private void finalizeINodeFileUnderConstruction(String src,

src/java/org/apache/hadoop/dfs/INode.java

@@ -776,7 +776,7 @@ class INodeFileUnderConstruction extends INodeFile {
   DatanodeDescriptor clientNode = null; // if client is a cluster node too.
 
   private int primaryNodeIndex = -1; //the node working on lease recovery
-  DatanodeDescriptor[] targets = null;   //locations for last block
+  private DatanodeDescriptor[] targets = null;   //locations for last block
   
   INodeFileUnderConstruction() {}
 
@@ -833,6 +833,15 @@ class INodeFileUnderConstruction extends INodeFile {
     return true;
   }
 
+  DatanodeDescriptor[] getTargets() {
+    return targets;
+  }
+
+  void setTargets(DatanodeDescriptor[] targets) {
+    this.targets = targets;
+    this.primaryNodeIndex = -1;
+  }
+
   //
   // converts a INodeFileUnderConstruction into a INodeFile
   //
@@ -870,14 +879,14 @@ class INodeFileUnderConstruction extends INodeFile {
 
   /**
    * Initialize lease recovery for this object
-   * @return the chosen primary datanode
    */
   void assignPrimaryDatanode() {
     //assign the first alive datanode as the primary datanode
+
     if (targets.length == 0) {
       NameNode.stateChangeLog.warn("BLOCK*"
-          + " INodeFileUnderConstruction.initLeaseRecovery:"
-          + " all targets are not alive.");
+        + " INodeFileUnderConstruction.initLeaseRecovery:"
+        + " No blocks found, lease removed.");
     }
 
     int previous = primaryNodeIndex;

src/java/org/apache/hadoop/dfs/InterDatanodeProtocol.java

@@ -30,9 +30,9 @@ interface InterDatanodeProtocol extends VersionedProtocol {
   public static final Log LOG = LogFactory.getLog(InterDatanodeProtocol.class);
 
   /**
-   * 2: change updateGenerationStamp to updataBlock
+   * 3: added a finalize parameter to updateBlock
    */
-  public static final long versionID = 2L;
+  public static final long versionID = 3L;
 
   /** @return the BlockMetaDataInfo of a block;
    *  null if the block is not found 
@@ -42,5 +42,5 @@ interface InterDatanodeProtocol extends VersionedProtocol {
   /**
    * Update the block to the new generation stamp and length.  
    */
-  void updateBlock(Block oldblock, Block newblock) throws IOException;
-}
+  void updateBlock(Block oldblock, Block newblock, boolean finalize) throws IOException;
+}

src/java/org/apache/hadoop/dfs/LeaseManager.java

@@ -357,7 +357,6 @@ class LeaseManager {
                   for(StringBytesWritable s : top.paths) {
                     fsnamesystem.internalReleaseLease(top, s.getString());
                   }
-                  renewLease(top);
                 } else {
                   break;
                 }
@@ -476,7 +475,7 @@ class LeaseManager {
 
     for(BlockRecord r : syncList) {
       try {
-        r.datanode.updateBlock(r.block, newblock);
+        r.datanode.updateBlock(r.block, newblock, closeFile);
         successList.add(r.id);
       } catch (IOException e) {
         InterDatanodeProtocol.LOG.warn("Failed to updateBlock (newblock="

src/test/org/apache/hadoop/dfs/TestInterDatanodeProtocol.java

@@ -91,11 +91,11 @@ public class TestInterDatanodeProtocol extends junit.framework.TestCase {
       //verify updateBlock
       Block newblock = new Block(
           b.getBlockId(), b.getNumBytes()/2, b.getGenerationStamp()+1);
-      idp.updateBlock(b, newblock);
+      idp.updateBlock(b, newblock, false);
       checkMetaInfo(newblock, idp, datanode.blockScanner);
     }
     finally {
       if (cluster != null) {cluster.shutdown();}
     }
   }
-}
+}

src/test/org/apache/hadoop/dfs/TestLeaseRecovery.java

@@ -100,7 +100,7 @@ public class TestLeaseRecovery extends junit.framework.TestCase {
       for(int i = 0; i < REPLICATION_NUM; i++) {
         newblocks[i] = new Block(lastblock.getBlockId(), newblocksizes[i],
             lastblock.getGenerationStamp());
-        idps[i].updateBlock(lastblock, newblocks[i]);
+        idps[i].updateBlock(lastblock, newblocks[i], false);
         checkMetaInfo(newblocks[i], idps[i]);
       }