HADOOP-2905. 'fsck -move' triggers NPE in NameNode. (Lohit Vjayarenu via rangadi)

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

CHANGES.txt

@@ -13,6 +13,9 @@ Trunk (unreleased changes)
 
   BUG FIXES
 
+   HADOOP-2905. 'fsck -move' triggers NPE in NameNode. 
+   (Lohit Vjayarenu via rangadi)
+
 Release 0.17.0 - Unreleased
 
   INCOMPATIBLE CHANGES

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

@@ -689,8 +689,7 @@ class DFSClient implements FSConstants {
   /**
    */
   public boolean mkdirs(String src) throws IOException {
-    checkOpen();
-    return namenode.mkdirs(src, null);
+    return mkdirs(src, null);
   }
 
   /**

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

@@ -185,6 +185,19 @@ public class DFSTestUtil extends TestCase {
       waitReplication(fs, new Path(root, files[idx].getName()), value);
     }
   }
+
+  /** return list of filenames created as part of createFiles */
+  public String[] getFileNames(String topDir) {
+    if (nFiles == 0)
+      return new String[]{};
+    else {
+      String[] fileNames =  new String[nFiles];
+      for (int idx=0; idx < nFiles; idx++) {
+        fileNames[idx] = topDir + "/" + files[idx].getName();
+      }
+      return fileNames;
+    }
+  }
   
   /** wait for the file's replication to be done */
   static void waitReplication(FileSystem fs, Path fileName, 

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

@@ -20,6 +20,8 @@ package org.apache.hadoop.dfs;
 
 import java.io.ByteArrayOutputStream;
 import java.io.PrintStream;
+import java.net.InetSocketAddress;
+import java.io.File;
 
 import junit.framework.TestCase;
 
@@ -33,13 +35,13 @@ import org.apache.hadoop.util.ToolRunner;
  * A JUnit test for doing fsck
  */
 public class TestFsck extends TestCase {
-  static String runFsck(Configuration conf, String path) throws Exception {
+  static String runFsck(Configuration conf, String... path) throws Exception {
     PrintStream oldOut = System.out;
     ByteArrayOutputStream bStream = new ByteArrayOutputStream();
     PrintStream newOut = new PrintStream(bStream, true);
     System.setOut(newOut);
     ((Log4JLogger)PermissionChecker.LOG).getLogger().setLevel(Level.ALL);
-    assertEquals(0, ToolRunner.run(new DFSck(conf), new String[] {path}));
+    assertEquals(0, ToolRunner.run(new DFSck(conf), path));
     ((Log4JLogger)PermissionChecker.LOG).getLogger().setLevel(Level.INFO);
     System.setOut(oldOut);
     return bStream.toString();
@@ -98,4 +100,60 @@ public class TestFsck extends TestCase {
       if (cluster != null) { cluster.shutdown(); }
     }
   }
-}
+
+  public void testFsckMove() throws Exception {
+    DFSTestUtil util = new DFSTestUtil("TestFsck", 5, 3, 8*1024);
+    MiniDFSCluster cluster = null;
+    FileSystem fs = null;
+    try {
+      Configuration conf = new Configuration();
+      conf.set("dfs.blockreport.intervalMsec", Integer.toString(30));
+      cluster = new MiniDFSCluster(conf, 4, true, null);
+      String topDir = "/srcdat";
+      fs = cluster.getFileSystem();
+      cluster.waitActive();
+      util.createFiles(fs, topDir);
+      String outStr = runFsck(conf, "/");
+      assertTrue(outStr.contains("HEALTHY"));
+      
+      // Corrupt a block by deleting it
+      String[] fileNames = util.getFileNames(topDir);
+      DFSClient dfsClient = new DFSClient(new InetSocketAddress("localhost",
+                                          cluster.getNameNodePort()), conf);
+      String block = dfsClient.namenode.
+                      getBlockLocations(fileNames[0], 0, Long.MAX_VALUE).
+                      get(0).getBlock().toString();
+      File baseDir = new File(System.getProperty("test.build.data"),"dfs/data");
+      for (int i=0; i<8; i++) {
+        File blockFile = new File(baseDir, "data" +(i+1)+ "/current/" + block);
+        if(blockFile.exists()) {
+          assertTrue(blockFile.delete());
+        }
+      }
+
+      //Sleep for a while until block reports are sent to discover corruption
+      try {
+        Thread.sleep(5000);
+      } catch (InterruptedException e) {
+      }
+      
+      // We excpect the filesystem to be corrupted
+      outStr = runFsck(conf, "/");
+      assertTrue(outStr.contains("CORRUPT"));
+      
+      // Fix the filesystem by moving corrupted files to lost+found
+      outStr = runFsck(conf, "/", "-move");
+      assertTrue(outStr.contains("CORRUPT"));
+      
+      // Check to make sure we have healthy filesystem
+      outStr = runFsck(conf, "/");
+      assertTrue(outStr.contains("HEALTHY")); 
+      util.cleanup(fs, topDir);
+      if (fs != null) {try{fs.close();} catch(Exception e){}}
+      cluster.shutdown();
+    } finally {
+      if (fs != null) {try{fs.close();} catch(Exception e){}}
+      if (cluster != null) { cluster.shutdown(); }
+    }
+  }
+}