HDFS-636. SafeMode counts complete blocks only. Contributed by Konstantin Shvachko.

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

CHANGES.txt

@@ -8,6 +8,8 @@ Append branch (unreleased changes)
 
     HDFS-576. Block report includes under-construction replicas. (shv)
 
+    HDFS-636. SafeMode counts complete blocks only. (shv)
+
   NEW FEATURES
 
     HDFS-536. Support hflush at DFSClient. (hairong)

src/java/org/apache/hadoop/hdfs/server/namenode/BlockManager.java

@@ -1102,15 +1102,18 @@ public class BlockManager {
     int numCurrentReplica = numLiveReplicas
       + pendingReplications.getNumReplicas(storedBlock);
 
+    if(storedBlock.getBlockUCState() == BlockUCState.COMMITTED &&
+        numLiveReplicas >= minReplication)
+      storedBlock = completeBlock(fileINode, storedBlock);
+
     // check whether safe replication is reached for the block
-    namesystem.incrementSafeBlockCount(numCurrentReplica);
+    // only complete blocks are counted towards that
+    if(!storedBlock.isUnderConstruction())
+      namesystem.incrementSafeBlockCount(numCurrentReplica);
 
     // if file is under construction, then check whether the block
     // can be completed
     if (fileINode.isUnderConstruction()) {
-      if(storedBlock.getBlockUCState() == BlockUCState.COMMITTED &&
-          numLiveReplicas >= minReplication)
-        storedBlock = completeBlock(fileINode, storedBlock);
       return storedBlock;
     }
 

src/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java

@@ -3371,7 +3371,7 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean, FSClusterSt
   void setBlockTotal() {
     if (safeMode == null)
       return;
-    safeMode.setBlockTotal((int)getBlocksTotal());
+    safeMode.setBlockTotal((int)getCompleteBlocksTotal());
   }
 
   /**
@@ -3381,6 +3381,33 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean, FSClusterSt
     return blockManager.getTotalBlocks();
   }
 
+  /**
+   * Get the total number of COMPLETE blocks in the system.
+   * For safe mode only complete blocks are counted.
+   */
+  long getCompleteBlocksTotal() {
+    // Calculate number of blocks under construction
+    long numUCBlocks = 0;
+    for (Lease lease : leaseManager.getSortedLeases()) {
+      for(String path : lease.getPaths()) {
+        INode node = dir.getFileINode(path);
+        assert node != null : "Found a lease for nonexisting file.";
+        assert node.isUnderConstruction() :
+          "Found a lease for file that is not under construction.";
+        INodeFileUnderConstruction cons = (INodeFileUnderConstruction) node;
+        BlockInfo[] blocks = cons.getBlocks();
+        if(blocks == null)
+          continue;
+        for(BlockInfo b : blocks) {
+          if(b.isUnderConstruction())
+            numUCBlocks++;
+        }
+      }
+    }
+    LOG.info("Number of blocks under construction: " + numUCBlocks);
+    return getBlocksTotal() - numUCBlocks;
+  }
+
   /**
    * Enter safe mode manually.
    * @throws IOException

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

@@ -30,8 +30,6 @@ import org.apache.hadoop.hdfs.server.namenode.NameNode;
 import org.apache.log4j.Level;
 
 public class TestRenameWhileOpen extends junit.framework.TestCase {
-  private static final long LEASE_PERIOD = 500L;
-
   {
     ((Log4JLogger)NameNode.stateChangeLog).getLogger().setLevel(Level.ALL);
     ((Log4JLogger)LeaseManager.LOG).getLogger().setLevel(Level.ALL);
@@ -54,7 +52,7 @@ public class TestRenameWhileOpen extends junit.framework.TestCase {
     conf.setInt("ipc.client.connection.maxidletime", MAX_IDLE_TIME);
     conf.setInt("heartbeat.recheck.interval", 1000);
     conf.setInt("dfs.heartbeat.interval", 1);
-    conf.setFloat("dfs.safemode.threshold.pct", 0.5f);
+    conf.setInt("dfs.safemode.threshold.pct", 1);
     conf.setBoolean("dfs.support.append", true);
 
     // create cluster
@@ -106,7 +104,7 @@ public class TestRenameWhileOpen extends junit.framework.TestCase {
       try {Thread.sleep(2*MAX_IDLE_TIME);} catch (InterruptedException e) {}
       cluster = new MiniDFSCluster(nnport, conf, 1, false, true, 
                                    null, null, null);
-      waitLeaseRecovery(cluster);
+      cluster.waitActive();
 
       // restart cluster yet again. This triggers the code to read in
       // persistent leases from fsimage.
@@ -138,7 +136,7 @@ public class TestRenameWhileOpen extends junit.framework.TestCase {
     conf.setInt("ipc.client.connection.maxidletime", MAX_IDLE_TIME);
     conf.setInt("heartbeat.recheck.interval", 1000);
     conf.setInt("dfs.heartbeat.interval", 1);
-    conf.setFloat("dfs.safemode.threshold.pct", 0.5f);
+    conf.setInt("dfs.safemode.threshold.pct", 1);
     conf.setBoolean("dfs.support.append", true);
     System.out.println("Test 2************************************");
 
@@ -178,7 +176,7 @@ public class TestRenameWhileOpen extends junit.framework.TestCase {
       try {Thread.sleep(2*MAX_IDLE_TIME);} catch (InterruptedException e) {}
       cluster = new MiniDFSCluster(nnport, conf, 1, false, true, 
                                    null, null, null);
-      waitLeaseRecovery(cluster);
+      cluster.waitActive();
 
       // restart cluster yet again. This triggers the code to read in
       // persistent leases from fsimage.
@@ -211,7 +209,7 @@ public class TestRenameWhileOpen extends junit.framework.TestCase {
     conf.setInt("ipc.client.connection.maxidletime", MAX_IDLE_TIME);
     conf.setInt("heartbeat.recheck.interval", 1000);
     conf.setInt("dfs.heartbeat.interval", 1);
-    conf.setFloat("dfs.safemode.threshold.pct", 0.5f);
+    conf.setInt("dfs.safemode.threshold.pct", 1);
     conf.setBoolean("dfs.support.append", true);
     System.out.println("Test 3************************************");
 
@@ -243,7 +241,7 @@ public class TestRenameWhileOpen extends junit.framework.TestCase {
       try {Thread.sleep(2*MAX_IDLE_TIME);} catch (InterruptedException e) {}
       cluster = new MiniDFSCluster(nnport, conf, 1, false, true, 
                                    null, null, null);
-      waitLeaseRecovery(cluster);
+      cluster.waitActive();
 
       // restart cluster yet again. This triggers the code to read in
       // persistent leases from fsimage.
@@ -274,7 +272,7 @@ public class TestRenameWhileOpen extends junit.framework.TestCase {
     conf.setInt("ipc.client.connection.maxidletime", MAX_IDLE_TIME);
     conf.setInt("heartbeat.recheck.interval", 1000);
     conf.setInt("dfs.heartbeat.interval", 1);
-    conf.setFloat("dfs.safemode.threshold.pct", 0.5f);
+    conf.setInt("dfs.safemode.threshold.pct", 1);
     conf.setBoolean("dfs.support.append", true);
     System.out.println("Test 4************************************");
 
@@ -305,7 +303,7 @@ public class TestRenameWhileOpen extends junit.framework.TestCase {
       try {Thread.sleep(2*MAX_IDLE_TIME);} catch (InterruptedException e) {}
       cluster = new MiniDFSCluster(nnport, conf, 1, false, true, 
                                    null, null, null);
-      waitLeaseRecovery(cluster);
+      cluster.waitActive();
 
       // restart cluster yet again. This triggers the code to read in
       // persistent leases from fsimage.
@@ -325,13 +323,4 @@ public class TestRenameWhileOpen extends junit.framework.TestCase {
       cluster.shutdown();
     }
   }
-
-  void waitLeaseRecovery(MiniDFSCluster cluster) {
-    cluster.setLeasePeriod(LEASE_PERIOD, LEASE_PERIOD);
-    // wait for the lease to expire
-    try {
-      Thread.sleep(5 * LEASE_PERIOD);
-    } catch (InterruptedException e) {
-    }
-  }
 }