HDFS-761. Fix failure to process rename operation from edits log due to quota verification. Contributed by Suresh Srinivas.

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

CHANGES.txt

@@ -57,6 +57,9 @@ Release 0.20.2 - Unreleased
     MAPREDUCE-1163. Remove unused, hard-coded paths from libhdfs. (Allen
     Wittenauer via cdouglas)
 
+    HDFS-761. Fix failure to process rename operation from edits log due to 
+    quota verification. (suresh)
+
 Release 0.20.1 - 2009-09-01
 
   INCOMPATIBLE CHANGES

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

@@ -1008,6 +1008,10 @@ class FSDirectory implements FSConstants, Closeable {
    */
   private void verifyQuota(INode[] inodes, int pos, long nsDelta, long dsDelta,
       INode commonAncestor) throws QuotaExceededException {
+    if (!ready) {
+      // Do not check quota if edits log is still being processed
+      return;
+    }
     if (pos>inodes.length) {
       pos = inodes.length;
     }
@@ -1041,6 +1045,10 @@ class FSDirectory implements FSConstants, Closeable {
    */
   private void verifyQuotaForRename(INode[] srcInodes, INode[]dstInodes)
       throws QuotaExceededException {
+    if (!ready) {
+      // Do not check quota if edits log is still being processed
+      return;
+    }
     INode srcInode = srcInodes[srcInodes.length - 1];
     INode commonAncestor = null;
     for(int i =0;srcInodes[i] == dstInodes[i]; i++) {

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

@@ -27,7 +27,8 @@ import org.apache.hadoop.hdfs.protocol.FSConstants;
 import org.apache.hadoop.hdfs.protocol.QuotaExceededException;
 
 public class TestDFSRename extends junit.framework.TestCase {
-  MiniDFSCluster cluster = null;
+  static Configuration CONF = new Configuration();
+  static MiniDFSCluster cluster = null;
   static int countLease(MiniDFSCluster cluster) {
     return cluster.getNameNode().namesystem.leaseManager.countLease();
   }
@@ -36,8 +37,16 @@ public class TestDFSRename extends junit.framework.TestCase {
 
   @Override
   protected void setUp() throws Exception {
-    Configuration conf = new Configuration();
-    cluster = new MiniDFSCluster(conf, 2, true, null);
+    cluster = new MiniDFSCluster(CONF, 2, true, null);
+  }
+  
+  private void restartCluster() throws IOException {
+    if (cluster != null) {
+      cluster.shutdown();
+      cluster = null;
+    }
+    cluster = new MiniDFSCluster(CONF, 1, false, null);
+    cluster.waitClusterUp();
   }
   
   @Override
@@ -156,6 +165,32 @@ public class TestDFSRename extends junit.framework.TestCase {
     rename(dst1, src1, false, true);
   }
   
+  /**
+   * Perform operations such as setting quota, deletion of files, rename and
+   * ensure system can apply edits log during startup.
+   */
+  public void testEditsLog() throws Exception {
+    DistributedFileSystem fs = (DistributedFileSystem) cluster.getFileSystem();
+    Path src1 = new Path(dir, "testEditsLog/srcdir/src1");
+    Path dst1 = new Path(dir, "testEditsLog/dstdir/dst1");
+    createFile(fs, src1);
+    fs.mkdirs(dst1.getParent());
+    createFile(fs, dst1);
+    
+    // Set quota so that dst1 parent cannot allow under it new files/directories 
+    fs.setQuota(dst1.getParent(), 2, FSConstants.QUOTA_DONT_SET);
+    // Free up quota for a subsequent rename
+    fs.delete(dst1, true);
+    rename(src1, dst1, true, false);
+    
+    // Restart the cluster and ensure the above operations can be
+    // loaded from the edits log
+    restartCluster();
+    fs = (DistributedFileSystem)cluster.getFileSystem();
+    assertFalse(fs.exists(src1));   // ensure src1 is already renamed
+    assertTrue(fs.exists(dst1));    // ensure rename dst exists
+  }
+  
   private void rename(Path src, Path dst, boolean renameSucceeds,
       boolean quotaException) throws Exception {
     DistributedFileSystem fs = (DistributedFileSystem) cluster.getFileSystem();