HDFS-3061. Cached directory size in INodeDirectory can get permanently out of sync with computed size, causing quota issues; port of HDFS-1487. Contributed by Kihwal Lee.

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

CHANGES.txt

@@ -68,6 +68,10 @@ Release 1.0.3 - unreleased
     HADOOP-8352. Regenerate configure scripts for the c++ compilation. 
     (omalley)
 
+    HDFS-3061. Cached directory size in INodeDirectory can get permanently
+    out of sync with computed size, causing quota issues; port of HDFS-1487.
+    (Kihwal Lee via mattf)
+
 Release 1.0.2 - 2012.03.24
 
   NEW FEATURES

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

@@ -343,7 +343,12 @@ class FSDirectory implements FSConstants, Closeable {
       NameNode.stateChangeLog.debug("DIR* FSDirectory.addFile: "
                                     +path+" with "+block
                                     +" block is added to the file system");
+      // update space consumed
+      INode[] pathINodes = getExistingPathINodes(path);
+      updateCount(pathINodes, pathINodes.length-1, 0,
+          -fileNode.getPreferredBlockSize()*fileNode.getReplication(), true);
     }
+
     return true;
   }
 

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

@@ -23,11 +23,15 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.*;
+import org.apache.hadoop.hdfs.protocol.FSConstants;
 import org.apache.hadoop.hdfs.protocol.LocatedBlock;
 import org.apache.hadoop.hdfs.protocol.LocatedBlocks;
+import org.apache.hadoop.hdfs.protocol.QuotaExceededException;
 import org.apache.hadoop.hdfs.server.namenode.NameNode;
 import org.apache.hadoop.util.StringUtils;
 
+import static org.junit.Assert.*;
+
 public class TestAbandonBlock extends junit.framework.TestCase {
   public static final Log LOG = LogFactory.getLog(TestAbandonBlock.class);
   
@@ -68,4 +72,37 @@ public class TestAbandonBlock extends junit.framework.TestCase {
       try{cluster.shutdown();} catch(Exception e) {}
     }
   }
+
+  /** Make sure that the quota is decremented correctly when a block is abandoned */
+  public void testQuotaUpdatedWhenBlockAbandoned() throws IOException {
+    MiniDFSCluster cluster = new MiniDFSCluster(CONF, 2, true, null);
+    FileSystem fs = cluster.getFileSystem();
+    DistributedFileSystem dfs = (DistributedFileSystem)fs;
+
+    try {
+      // Setting diskspace quota to 3MB
+      dfs.setQuota(new Path("/"), FSConstants.QUOTA_DONT_SET, 3 * 1024 * 1024);
+
+      // Start writing a file with 2 replicas to ensure each datanode has one.
+      // Block Size is 1MB.
+      String src = FILE_NAME_PREFIX + "test_quota1";
+      FSDataOutputStream fout = fs.create(new Path(src), true, 4096, (short)2, 1024 * 1024);
+      for (int i = 0; i < 1024; i++) {
+        fout.writeByte(123);
+      }
+
+      // Shutdown one datanode, causing the block abandonment.
+      cluster.getDataNodes().get(0).shutdown();
+
+      // Close the file, new block will be allocated with 2MB pending size.
+      try {
+        fout.close();
+      } catch (QuotaExceededException e) {
+        fail("Unexpected quota exception when closing fout");
+      }
+    } finally {
+      try{fs.close();} catch(Exception e) {}
+      try{cluster.shutdown();} catch(Exception e) {}
+    }
+  }
 }