HADOOP-4530. In fsck, HttpServletResponse sendError fails with IllegalStateException. Contributed by Hairong Kuang.

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

CHANGES.txt

@@ -88,6 +88,9 @@ Trunk (unreleased changes)
 
     HADOOP-4587. Fix a typo in Mapper javadoc.  (Koji Noguchi via szetszwo)
 
+    HADOOP-4530. In fsck, HttpServletResponse sendError fails with
+    IllegalStateException. (hairong)
+
 Release 0.19.0 - Unreleased
 
   INCOMPATIBLE CHANGES

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

@@ -19,7 +19,6 @@ package org.apache.hadoop.hdfs.server.namenode;
 
 import java.util.*;
 import java.io.*;
-import org.apache.hadoop.util.StringUtils;
 import org.apache.hadoop.conf.*;
 import org.apache.commons.logging.*;
 import javax.servlet.ServletContext;
@@ -32,26 +31,15 @@ import javax.servlet.http.HttpServletResponse;
  * This class is used in Namesystem's jetty to do fsck on namenode.
  */
 public class FsckServlet extends HttpServlet {
-
-  private static final Log LOG = LogFactory.getLog(FSNamesystem.class.getName());
-
   @SuppressWarnings("unchecked")
   public void doGet(HttpServletRequest request,
                     HttpServletResponse response
                     ) throws ServletException, IOException {
     Map<String,String[]> pmap = request.getParameterMap();
-    try {
-      ServletContext context = getServletContext();
-      NameNode nn = (NameNode) context.getAttribute("name.node");
-      Configuration conf = (Configuration) context.getAttribute("name.conf");
-      NamenodeFsck fscker = new NamenodeFsck(conf, nn, pmap, response);
-      fscker.fsck();
-    } catch (IOException ie) {
-      StringUtils.stringifyException(ie);
-      LOG.warn(ie);
-      String errMsg = "Fsck on path " + pmap.get("path") + " failed.";
-      response.sendError(HttpServletResponse.SC_GONE, errMsg);
-      throw ie;
-    }
+    ServletContext context = getServletContext();
+    NameNode nn = (NameNode) context.getAttribute("name.node");
+    Configuration conf = (Configuration) context.getAttribute("name.conf");
+    NamenodeFsck fscker = new NamenodeFsck(conf, nn, pmap, response);
+    fscker.fsck();
   }
 }

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

@@ -68,6 +68,12 @@ import org.apache.hadoop.fs.permission.PermissionStatus;
 public class NamenodeFsck {
   public static final Log LOG = LogFactory.getLog(NameNode.class.getName());
   
+  // return string marking fsck status
+  public static final String CORRUPT_STATUS = "is CORRUPT";
+  public static final String HEALTHY_STATUS = "is HEALTHY";
+  public static final String NONEXISTENT_STATUS = "does not exist";
+  public static final String FAILURE_STATUS = "FAILED";
+  
   /** Don't attempt any fixing . */
   public static final int FIXING_NONE = 0;
   /** Move corrupted files to /lost+found . */
@@ -139,13 +145,18 @@ public class NamenodeFsck {
         // of file system and return appropriate code. Changing the output string
         // might break testcases. 
         if (res.isHealthy()) {
-          out.println("\n\nThe filesystem under path '" + path + "' is HEALTHY");
+          out.print("\n\nThe filesystem under path '" + path + "' " + HEALTHY_STATUS);
         }  else {
-          out.println("\n\nThe filesystem under path '" + path + "' is CORRUPT");
+          out.print("\n\nThe filesystem under path '" + path + "' " + CORRUPT_STATUS);
         }
       } else {
-        out.println("\n\nPath '" + path + "' does not exist.");
+        out.print("\n\nPath '" + path + "' " + NONEXISTENT_STATUS);
       }
+    } catch (Exception e) {
+      String errMsg = "Fsck on path '" + path + "' " + FAILURE_STATUS;
+      LOG.warn(errMsg, e);
+      out.println(e.getMessage());
+      out.print("\n\n"+errMsg);
     } finally {
       out.close();
     }

src/hdfs/org/apache/hadoop/hdfs/tools/DFSck.java

@@ -27,6 +27,7 @@ import java.net.URLEncoder;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.conf.Configured;
+import org.apache.hadoop.hdfs.server.namenode.NamenodeFsck;
 import org.apache.hadoop.net.NetUtils;
 import org.apache.hadoop.util.Tool;
 import org.apache.hadoop.util.ToolRunner;
@@ -123,19 +124,23 @@ public class DFSck extends Configured implements Tool {
     BufferedReader input = new BufferedReader(new InputStreamReader(
                                               stream, "UTF-8"));
     String line = null;
-    int errCode = 0;
-    // errCode returned indicating the status of Filesystem (HEALTHY/CORRUPT)
-    // depends on the format of the string. Changing the script might break
-    // fsck related testcases. For now, we scan for "is CORRUPT" as it unique.
+    String lastLine = null;
+    int errCode = -1;
     try {
       while ((line = input.readLine()) != null) {
         System.out.println(line);
-        if (line.contains("is CORRUPT"))
-          errCode = 1;
+        lastLine = line;
       }
     } finally {
       input.close();
     }
+    if (lastLine.endsWith(NamenodeFsck.HEALTHY_STATUS)) {
+      errCode = 0;
+    } else if (lastLine.endsWith(NamenodeFsck.CORRUPT_STATUS)) {
+      errCode = 1;
+    } else if (lastLine.endsWith(NamenodeFsck.NONEXISTENT_STATUS)) {
+      errCode = 0;
+    }
     return errCode;
   }
 

src/test/org/apache/hadoop/hdfs/server/namenode/TestFsck.java

@@ -26,7 +26,6 @@ import java.io.RandomAccessFile;
 import java.lang.Exception;
 import java.io.IOException;
 import java.nio.channels.FileChannel;
-import java.nio.ByteBuffer;
 import java.util.Random;
 
 import junit.framework.TestCase;
@@ -77,7 +76,7 @@ public class TestFsck extends TestCase {
       util.createFiles(fs, "/srcdat");
       util.waitReplication(fs, "/srcdat", (short)3);
       String outStr = runFsck(conf, 0, true, "/");
-      assertTrue(-1 != outStr.indexOf("HEALTHY"));
+      assertTrue(outStr.contains(NamenodeFsck.HEALTHY_STATUS));
       System.out.println(outStr);
       if (fs != null) {try{fs.close();} catch(Exception e){}}
       cluster.shutdown();
@@ -86,7 +85,7 @@ public class TestFsck extends TestCase {
       cluster = new MiniDFSCluster(conf, 0, false, null);
       outStr = runFsck(conf, 1, true, "/");
       // expect the result is corrupt
-      assertTrue(outStr.contains("CORRUPT"));
+      assertTrue(outStr.contains(NamenodeFsck.CORRUPT_STATUS));
       System.out.println(outStr);
       
       // bring up data nodes & cleanup cluster
@@ -112,7 +111,7 @@ public class TestFsck extends TestCase {
       util.createFiles(fs, "/srcdat");
       util.waitReplication(fs, "/srcdat", (short)3);
       String outStr = runFsck(conf, 0, true, "/non-existent");
-      assertEquals(-1, outStr.indexOf("HEALTHY"));
+      assertEquals(-1, outStr.indexOf(NamenodeFsck.HEALTHY_STATUS));
       System.out.println(outStr);
       util.cleanup(fs, "/srcdat");
     } finally {
@@ -135,7 +134,7 @@ public class TestFsck extends TestCase {
       util.createFiles(fs, topDir);
       util.waitReplication(fs, topDir, (short)3);
       String outStr = runFsck(conf, 0, true, "/");
-      assertTrue(outStr.contains("HEALTHY"));
+      assertTrue(outStr.contains(NamenodeFsck.HEALTHY_STATUS));
       
       // Corrupt a block by deleting it
       String[] fileNames = util.getFileNames(topDir);
@@ -154,7 +153,7 @@ public class TestFsck extends TestCase {
 
       // We excpect the filesystem to be corrupted
       outStr = runFsck(conf, 1, false, "/");
-      while (!outStr.contains("CORRUPT")) {
+      while (!outStr.contains(NamenodeFsck.CORRUPT_STATUS)) {
         try {
           Thread.sleep(100);
         } catch (InterruptedException ignore) {
@@ -164,11 +163,11 @@ public class TestFsck extends TestCase {
       
       // Fix the filesystem by moving corrupted files to lost+found
       outStr = runFsck(conf, 1, true, "/", "-move");
-      assertTrue(outStr.contains("CORRUPT"));
+      assertTrue(outStr.contains(NamenodeFsck.CORRUPT_STATUS));
       
       // Check to make sure we have healthy filesystem
       outStr = runFsck(conf, 0, true, "/");
-      assertTrue(outStr.contains("HEALTHY")); 
+      assertTrue(outStr.contains(NamenodeFsck.HEALTHY_STATUS)); 
       util.cleanup(fs, topDir);
       if (fs != null) {try{fs.close();} catch(Exception e){}}
       cluster.shutdown();
@@ -192,7 +191,7 @@ public class TestFsck extends TestCase {
       util.createFiles(fs, topDir);
       util.waitReplication(fs, topDir, (short)3);
       String outStr = runFsck(conf, 0, true, "/");
-      assertTrue(outStr.contains("HEALTHY"));
+      assertTrue(outStr.contains(NamenodeFsck.HEALTHY_STATUS));
       // Open a file for writing and do not close for now
       Path openFile = new Path(topDir + "/openFile");
       FSDataOutputStream out = fs.create(openFile);
@@ -204,7 +203,7 @@ public class TestFsck extends TestCase {
       // We expect the filesystem to be HEALTHY and show one open file
       outStr = runFsck(conf, 0, true, topDir);
       System.out.println(outStr);
-      assertTrue(outStr.contains("HEALTHY"));
+      assertTrue(outStr.contains(NamenodeFsck.HEALTHY_STATUS));
       assertFalse(outStr.contains("OPENFORWRITE")); 
       // Use -openforwrite option to list open files
       outStr = runFsck(conf, 0, true, topDir, "-openforwrite");
@@ -216,7 +215,7 @@ public class TestFsck extends TestCase {
       // Now, fsck should show HEALTHY fs and should not show any open files
       outStr = runFsck(conf, 0, true, topDir);
       System.out.println(outStr);
-      assertTrue(outStr.contains("HEALTHY"));
+      assertTrue(outStr.contains(NamenodeFsck.HEALTHY_STATUS));
       assertFalse(outStr.contains("OPENFORWRITE"));
       util.cleanup(fs, topDir);
       if (fs != null) {try{fs.close();} catch(Exception e){}}
@@ -237,7 +236,9 @@ public class TestFsck extends TestCase {
     Random random = new Random();
     String outStr = null;
 
-    MiniDFSCluster cluster = new MiniDFSCluster(conf, 3, true, null);
+    MiniDFSCluster cluster = null;
+    try {
+    cluster = new MiniDFSCluster(conf, 3, true, null);
     cluster.waitActive();
     fs = cluster.getFileSystem();
     Path file1 = new Path("/testCorruptBlock");
@@ -249,7 +250,7 @@ public class TestFsck extends TestCase {
     // Make sure filesystem is in healthy state
     outStr = runFsck(conf, 0, true, "/");
     System.out.println(outStr);
-    assertTrue(outStr.contains("HEALTHY"));
+    assertTrue(outStr.contains(NamenodeFsck.HEALTHY_STATUS));
     
     // corrupt replicas 
     File baseDir = new File(System.getProperty("test.build.data"), "dfs/data");
@@ -293,9 +294,45 @@ public class TestFsck extends TestCase {
     // Check if fsck reports the same
     outStr = runFsck(conf, 1, true, "/");
     System.out.println(outStr);
-    assertTrue(outStr.contains("CORRUPT"));
+    assertTrue(outStr.contains(NamenodeFsck.CORRUPT_STATUS));
     assertTrue(outStr.contains("testCorruptBlock"));
-
-    cluster.shutdown();
+    } finally {
+      if (cluster != null) {cluster.shutdown();}
+    }
+  }
+  
+  /** Test if fsck can return -1 in case of failure
+   * 
+   * @throws Exception
+   */
+  public void testFsckError() throws Exception {
+    MiniDFSCluster cluster = null;
+    try {
+      // bring up a one-node cluster
+      Configuration conf = new Configuration();
+      cluster = new MiniDFSCluster(conf, 1, true, null);
+      String fileName = "/test.txt";
+      Path filePath = new Path(fileName);
+      FileSystem fs = cluster.getFileSystem();
+      
+      // create a one-block file
+      DFSTestUtil.createFile(fs, filePath, 1L, (short)1, 1L);
+      DFSTestUtil.waitReplication(fs, filePath, (short)1);
+      
+      // intentionally corrupt NN data structure
+      INodeFile node = (INodeFile)cluster.getNameNode().namesystem.dir.rootDir.getNode(fileName);
+      assertEquals(node.blocks.length, 1);
+      node.blocks[0].setNumBytes(-1L);  // set the block length to be negative
+      
+      // run fsck and expect a failure with -1 as the error code
+      String outStr = runFsck(conf, -1, true, fileName);
+      System.out.println(outStr);
+      assertTrue(outStr.contains(NamenodeFsck.FAILURE_STATUS));
+      
+      // clean up file system
+      fs.delete(filePath, true);
+    } finally {
+      if (cluster != null) {cluster.shutdown();}
+    }
   }
 }