HADOOP-3883. Limit namenode to assign at most one generation stamp for a particular block within a short period. (szetszwo)

git-svn-id: https://svn.apache.org/repos/asf/hadoop/core/branches/branch-0.18@712335 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -15,6 +15,9 @@ Release 0.18.3 - Unreleased
     HADOOP-4610. Always calculate mis-replicated blocks when safe-mode is 
     turned off. (shv)
 
+    HADOOP-3883. Limit namenode to assign at most one generation stamp for
+    a particular block within a short period. (szetszwo)
+
 Release 0.18.2 - 2008-11-03
 
   BUG FIXES

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

@@ -3174,8 +3174,17 @@ public class DataNode extends Configured
   /** {@inheritDoc} */
   public Block recoverBlock(Block block, DatanodeInfo[] targets
       ) throws IOException {
-    LOG.info("Client invoking recoverBlock for block " + block);
+    logRecoverBlock("Client", block, targets);
     return LeaseManager.recoverBlock(block, targets, this, namenode, 
                                      getConf(), false);
   }
+
+  static void logRecoverBlock(String who, Block block, DatanodeID[] targets) {
+    StringBuilder msg = new StringBuilder(targets[0].getName());
+    for (int i = 1; i < targets.length; i++) {
+      msg.append(", " + targets[i].getName());
+    }
+    LOG.info(who + " calls recoverBlock(block=" + block
+        + ", targets=[" + msg + "])");
+  }
 }

src/hdfs/org/apache/hadoop/dfs/FSConstants.java

@@ -123,6 +123,8 @@ public interface FSConstants {
   public static long BLOCKREPORT_INITIAL_DELAY = 0;
   public static final long LEASE_SOFTLIMIT_PERIOD = 60 * 1000;
   public static final long LEASE_HARDLIMIT_PERIOD = 60 * LEASE_SOFTLIMIT_PERIOD;
+  public static final long LEASE_RECOVER_PERIOD = 10 * 1000; //in ms
+
   public static int READ_TIMEOUT = 60 * 1000;
   public static int WRITE_TIMEOUT = 8 * 60 * 1000;  
   public static int WRITE_TIMEOUT_EXTENSION = 5 * 1000; //for write pipeline

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

@@ -1783,10 +1783,7 @@ class FSNamesystem implements FSConstants, FSNamesystemMBean {
     if (!closeFile) {
       dir.persistBlocks(src, pendingFile);
       getEditLog().logSync();
-      LOG.info("commitBlockSynchronization(lastblock=" + lastblock
-          + ", newgenerationstamp=" + newgenerationstamp
-          + ", newlength=" + newlength
-          + ", newtargets=" + Arrays.asList(newtargets) + ") successful");
+      LOG.info("commitBlockSynchronization(" + lastblock + ") successful");
       return;
     }
     
@@ -4367,14 +4364,20 @@ class FSNamesystem implements FSConstants, FSNamesystemMBean {
    * Increments, logs and then returns the stamp
    */
   synchronized long nextGenerationStampForBlock(Block block) throws IOException {
-    String msg = "Block " + block + " is already commited.";
     BlockInfo storedBlock = blocksMap.getStoredBlock(block);
     if (storedBlock == null) {
+      String msg = block + " is already commited, storedBlock == null.";
       LOG.info(msg);
       throw new IOException(msg);
     }
-    INode fileINode = storedBlock.getINode();
+    INodeFile fileINode = storedBlock.getINode();
     if (!fileINode.isUnderConstruction()) {
+      String msg = block + " is already commited, !fileINode.isUnderConstruction().";
+      LOG.info(msg);
+      throw new IOException(msg);
+    }
+    if (!((INodeFileUnderConstruction)fileINode).setLastRecoveryTime(now())) {
+      String msg = block + " is beening recovered, ignoring this request.";
       LOG.info(msg);
       throw new IOException(msg);
     }

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

@@ -929,6 +929,7 @@ class INodeFileUnderConstruction extends INodeFile {
 
   private int primaryNodeIndex = -1; //the node working on lease recovery
   private DatanodeDescriptor[] targets = null;   //locations for last block
+  private long lastRecoveryTime = 0;
   
   INodeFileUnderConstruction() {}
 
@@ -1029,7 +1030,7 @@ class INodeFileUnderConstruction extends INodeFile {
     targets = null;
   }
 
-  void setLastBlock(BlockInfo newblock, DatanodeDescriptor[] newtargets
+  synchronized void setLastBlock(BlockInfo newblock, DatanodeDescriptor[] newtargets
       ) throws IOException {
     if (blocks == null) {
       throw new IOException("Trying to update non-existant block (newblock="
@@ -1037,6 +1038,7 @@ class INodeFileUnderConstruction extends INodeFile {
     }
     blocks[blocks.length - 1] = newblock;
     setTargets(newtargets);
+    lastRecoveryTime = 0;
   }
 
   /**
@@ -1063,4 +1065,16 @@ class INodeFileUnderConstruction extends INodeFile {
       }
     }
   }
+
+  /**
+   * Update lastRecoveryTime if expired.
+   * @return true if lastRecoveryTimeis updated. 
+   */
+  synchronized boolean setLastRecoveryTime(long now) {
+    boolean expired = now - lastRecoveryTime > NameNode.LEASE_RECOVER_PERIOD;
+    if (expired) {
+      lastRecoveryTime = now;
+    }
+    return expired;
+  }
 }

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

@@ -409,9 +409,10 @@ class LeaseManager {
       DataNode primary, DatanodeProtocol namenode, Configuration conf) {
     for(int i = 0; i < blocks.length; i++) {
       try {
+        DataNode.logRecoverBlock("NameNode", blocks[i], targets[i]);
         recoverBlock(blocks[i], targets[i], primary, namenode, conf, true);
       } catch (IOException e) {
-        LOG.warn("recoverBlocks, i=" + i, e);
+        LOG.warn("recoverBlocks FAILED, blocks[" + i + "]=" + blocks[i], e);
       }
     }
   }
@@ -436,10 +437,6 @@ class LeaseManager {
       ongoingRecovery.put(block, block);
     }
     try {
-      if (LOG.isDebugEnabled()) {
-        LOG.debug("block=" + block
-            + ", datanodeids=" + Arrays.asList(datanodeids));
-      }
       List<BlockRecord> syncList = new ArrayList<BlockRecord>();
       long minlength = Long.MAX_VALUE;
       int errorCount = 0;
@@ -514,6 +511,12 @@ class LeaseManager {
           successList.toArray(new DatanodeID[successList.size()]));
       return newblock; // success
     }
-    return null; // failed
-  }
+
+    //failed
+    StringBuilder b = new StringBuilder();
+    for(BlockRecord r : syncList) {
+      b.append("\n  " + r.id);
+    }
+    throw new IOException("Cannot recover " + block + ", none of these "
+        + syncList.size() + " datanodes success {" + b + "\n}");  }
 }

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

@@ -18,12 +18,22 @@
 package org.apache.hadoop.dfs;
 
 import java.io.IOException;
-import java.util.*;
+import java.util.Random;
 
+import org.apache.commons.logging.impl.Log4JLogger;
 import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.*;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.Path;
+import org.apache.log4j.Level;
 
 public class TestLeaseRecovery2 extends junit.framework.TestCase {
+  {
+    ((Log4JLogger)DataNode.LOG).getLogger().setLevel(Level.ALL);
+    ((Log4JLogger)LeaseManager.LOG).getLogger().setLevel(Level.ALL);
+    ((Log4JLogger)FSNamesystem.LOG).getLogger().setLevel(Level.ALL);
+  }
+
   static final int BLOCK_SIZE = 64;
   static final int FILE_SIZE = 1024;
   static final short REPLICATION_NUM = (short)3;