svn merge -c 1354390 from branch-1 for HDFS-3522.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/branch-1.1@1354391 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -256,6 +256,9 @@ Release 1.1.0 - unreleased
     HADOOP-6947.  Kerberos relogin should set refreshKrb5Config to true.
     (Todd Lipcon via ddas)
     
+    HDFS-3522. If a namenode is in safemode, it should throw SafeModeException
+    when getBlockLocations has zero locations.  (Brandon Li via szetszwo)
+
 Release 1.0.3 - 2012.05.07
 
   NEW FEATURES

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

@@ -906,7 +906,8 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean,
    */
   LocatedBlocks getBlockLocations(String clientMachine, String src,
       long offset, long length) throws IOException {
-    LocatedBlocks blocks = getBlockLocations(src, offset, length, true, true);
+    LocatedBlocks blocks = getBlockLocations(src, offset, length, true, true,
+        true);
     if (blocks != null) {
       //sort the blocks
       DatanodeDescriptor client = host2DataNodeMap.getDatanodeByHost(
@@ -924,7 +925,7 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean,
    */
   public LocatedBlocks getBlockLocations(String src, long offset, long length
       ) throws IOException {
-    return getBlockLocations(src, offset, length, false, true);
+    return getBlockLocations(src, offset, length, false, true, true);
   }
 
   /**
@@ -932,7 +933,8 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean,
    * @see ClientProtocol#getBlockLocations(String, long, long)
    */
   public LocatedBlocks getBlockLocations(String src, long offset, long length,
-      boolean doAccessTime, boolean needBlockToken) throws IOException {
+      boolean doAccessTime, boolean needBlockToken, boolean checkSafeMode)
+      throws IOException {
     if (isPermissionEnabled) {
       checkPathAccess(src, FsAction.READ);
     }
@@ -950,6 +952,15 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean,
                     Server.getRemoteIp(),
                     "open", src, null, null);
     }
+    if (checkSafeMode && isInSafeMode()) {
+      for (LocatedBlock b : ret.getLocatedBlocks()) {
+        // if safemode & no block locations yet then throw safemodeException
+        if ((b.getLocations() == null) || (b.getLocations().length == 0)) {
+          throw new SafeModeException("Zero blocklocations for " + src,
+              safeMode);
+        }
+      }
+    }
     return ret;
   }
 

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

@@ -217,7 +217,7 @@ public class NamenodeFsck {
     // Get block locations without updating the file access time 
     // and without block access tokens
     LocatedBlocks blocks = namenode.getNamesystem().getBlockLocations(path, 0,
-        fileLen, false, false);
+        fileLen, false, false, false);
     if (blocks == null) { // the file is deleted
       return;
     }

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

@@ -18,26 +18,32 @@
 
 package org.apache.hadoop.hdfs;
 
+import static org.junit.Assert.*;
+
 import java.io.IOException;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hdfs.protocol.FSConstants.SafeModeAction;
 import org.apache.hadoop.hdfs.server.namenode.NameNode;
-
-import junit.framework.TestCase;
+import org.apache.hadoop.hdfs.server.namenode.SafeModeException;
+import org.apache.hadoop.ipc.RemoteException;
+import org.junit.Assert;
+import org.junit.Test;
 
 /**
  * This test makes sure that if SafeMode is manually entered, NameNode does not
  * come out of safe mode even after the startup safemode conditions are met.
  */
-public class TestSafeMode extends TestCase {
+public class TestSafeMode {
   
   static Log LOG = LogFactory.getLog(TestSafeMode.class);
   
+  @Test
   public void testManualSafeMode() throws IOException {
     MiniDFSCluster cluster = null;
     FileSystem fs = null;
@@ -92,4 +98,84 @@ public class TestSafeMode extends TestCase {
       if(cluster!= null) cluster.shutdown();
     }
   }
+
+  @Test
+  public void testSafeModeWhenZeroBlockLocations() throws IOException {
+    MiniDFSCluster cluster = null;
+    FileSystem fs = null;
+    try {
+      Configuration conf = new Configuration();
+      // disable safemode extension to make the test run faster.
+      conf.set("dfs.safemode.extension", "1");
+      cluster = new MiniDFSCluster(conf, 1, true, null);
+      cluster.waitActive();
+
+      fs = cluster.getFileSystem();
+      Path file1 = new Path("/tmp/testManualSafeMode/file1");
+      Path file2 = new Path("/tmp/testManualSafeMode/file2");
+
+      LOG.info("Created file1 and file2.");
+
+      // create two files with one block each.
+      DFSTestUtil.createFile(fs, file1, 1000, (short) 1, 0);
+      DFSTestUtil.createFile(fs, file2, 2000, (short) 1, 0);
+      checkGetBlockLocationsWorks(fs, file1);
+
+      NameNode namenode = cluster.getNameNode();
+      namenode.setSafeMode(SafeModeAction.SAFEMODE_ENTER);
+      Assert.assertTrue("should still be in SafeMode", namenode.isInSafeMode());
+      // getBlock locations should still work since block locations exists
+      checkGetBlockLocationsWorks(fs, file1);
+      namenode.setSafeMode(SafeModeAction.SAFEMODE_LEAVE);
+      assertFalse("should not be in SafeMode", namenode.isInSafeMode());
+      Assert.assertFalse("should not be in SafeMode", namenode.isInSafeMode());
+
+      // Now 2nd part of the tests where there aren't block locations
+      cluster.shutdownDataNodes();
+      cluster.shutdownNameNode();
+
+      // now bring up just the NameNode.
+      cluster.restartNameNode();
+      cluster.waitActive();
+
+      LOG.info("Restarted cluster with just the NameNode");
+
+      namenode = cluster.getNameNode();
+
+      Assert.assertTrue("No datanode is started. Should be in SafeMode",
+          namenode.isInSafeMode());
+      FileStatus stat = fs.getFileStatus(file1);
+      try {
+        fs.getFileBlockLocations(stat, 0, 1000);
+        Assert.assertTrue("Should have got safemode exception", false);
+      } catch (SafeModeException e) {
+        // as expected
+      } catch (RemoteException re) {
+        if (!re.getClassName().equals(SafeModeException.class.getName()))
+          Assert.assertTrue("Should have got safemode exception", false);
+      }
+
+      namenode.setSafeMode(SafeModeAction.SAFEMODE_LEAVE);
+      Assert.assertFalse("Should not be in safemode", namenode.isInSafeMode());
+      checkGetBlockLocationsWorks(fs, file1);
+
+    } finally {
+      if (fs != null)
+        fs.close();
+      if (cluster != null)
+        cluster.shutdown();
+    }
+  }
+
+  void checkGetBlockLocationsWorks(FileSystem fs, Path fileName)
+      throws IOException {
+    FileStatus stat = fs.getFileStatus(fileName);
+    try {
+      fs.getFileBlockLocations(stat, 0, 1000);
+    } catch (SafeModeException e) {
+      Assert.assertTrue("Should have not got safemode exception", false);
+    } catch (RemoteException re) {
+      Assert.assertTrue("Should have not got safemode exception", false);
+    }
+  }
 }