HDFS-14499. Misleading REM_QUOTA value with snapshot and trash feature enabled for a directory. Contributed by Shashikant Banerjee.

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/INodeReference.java

@@ -500,14 +500,15 @@ public abstract class INodeReference extends INode {
     
     @Override
     public final ContentSummaryComputationContext computeContentSummary(
-        int snapshotId, ContentSummaryComputationContext summary) {
-      final int s = snapshotId < lastSnapshotId ? snapshotId : lastSnapshotId;
-      // only count storagespace for WithName
-      final QuotaCounts q = computeQuotaUsage(
-          summary.getBlockStoragePolicySuite(), getStoragePolicyID(), false, s);
-      summary.getCounts().addContent(Content.DISKSPACE, q.getStorageSpace());
-      summary.getCounts().addTypeSpaces(q.getTypeSpaces());
-      return summary;
+        int snapshotId, ContentSummaryComputationContext summary)
+        throws AccessControlException {
+      Preconditions.checkState(snapshotId == Snapshot.CURRENT_STATE_ID
+          || this.lastSnapshotId >= snapshotId);
+      final INode referred =
+          this.getReferredINode().asReference().getReferredINode();
+      int id = snapshotId != Snapshot.CURRENT_STATE_ID ? snapshotId :
+          this.lastSnapshotId;
+      return referred.computeContentSummary(id, summary);
     }
 
     @Override

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/snapshot/TestGetContentSummaryWithSnapshot.java

@@ -90,18 +90,22 @@ public class TestGetContentSummaryWithSnapshot {
     final Path foo = new Path("/foo");
     final Path bar = new Path(foo, "bar");
     final Path baz = new Path(bar, "baz");
+    final Path qux = new Path(bar, "qux");
+    final Path temp = new Path("/temp");
 
     dfs.mkdirs(bar);
+    dfs.mkdirs(temp);
     dfs.allowSnapshot(foo);
     dfs.createSnapshot(foo, "s1");
 
     DFSTestUtil.createFile(dfs, baz, 10, REPLICATION, 0L);
+    DFSTestUtil.createFile(dfs, qux, 10, REPLICATION, 0L);
 
     ContentSummary summary = cluster.getNameNodeRpc().getContentSummary(
         bar.toString());
     Assert.assertEquals(1, summary.getDirectoryCount());
-    Assert.assertEquals(1, summary.getFileCount());
-    Assert.assertEquals(10, summary.getLength());
+    Assert.assertEquals(2, summary.getFileCount());
+    Assert.assertEquals(20, summary.getLength());
 
     final Path barS1 = SnapshotTestHelper.getSnapshotPath(foo, "s1", "bar");
     summary = cluster.getNameNodeRpc().getContentSummary(barS1.toString());
@@ -112,8 +116,8 @@ public class TestGetContentSummaryWithSnapshot {
     // also check /foo and /foo/.snapshot/s1
     summary = cluster.getNameNodeRpc().getContentSummary(foo.toString());
     Assert.assertEquals(2, summary.getDirectoryCount());
-    Assert.assertEquals(1, summary.getFileCount());
-    Assert.assertEquals(10, summary.getLength());
+    Assert.assertEquals(2, summary.getFileCount());
+    Assert.assertEquals(20, summary.getLength());
 
     final Path fooS1 = SnapshotTestHelper.getSnapshotRoot(foo, "s1");
     summary = cluster.getNameNodeRpc().getContentSummary(fooS1.toString());
@@ -127,14 +131,14 @@ public class TestGetContentSummaryWithSnapshot {
     summary = cluster.getNameNodeRpc().getContentSummary(
         bar.toString());
     Assert.assertEquals(1, summary.getDirectoryCount());
-    Assert.assertEquals(1, summary.getFileCount());
-    Assert.assertEquals(20, summary.getLength());
+    Assert.assertEquals(2, summary.getFileCount());
+    Assert.assertEquals(30, summary.getLength());
 
     final Path fooS2 = SnapshotTestHelper.getSnapshotRoot(foo, "s2");
     summary = cluster.getNameNodeRpc().getContentSummary(fooS2.toString());
     Assert.assertEquals(2, summary.getDirectoryCount());
-    Assert.assertEquals(1, summary.getFileCount());
-    Assert.assertEquals(10, summary.getLength());
+    Assert.assertEquals(2, summary.getFileCount());
+    Assert.assertEquals(20, summary.getLength());
 
     cluster.getNameNodeRpc().delete(baz.toString(), false);
 
@@ -143,11 +147,24 @@ public class TestGetContentSummaryWithSnapshot {
     Assert.assertEquals(0, summary.getSnapshotDirectoryCount());
     Assert.assertEquals(1, summary.getSnapshotFileCount());
     Assert.assertEquals(20, summary.getSnapshotLength());
+    Assert.assertEquals(2, summary.getDirectoryCount());
+    Assert.assertEquals(2, summary.getFileCount());
+    Assert.assertEquals(30, summary.getLength());
 
     final Path bazS1 = SnapshotTestHelper.getSnapshotPath(foo, "s1", "bar/baz");
     try {
       cluster.getNameNodeRpc().getContentSummary(bazS1.toString());
       Assert.fail("should get FileNotFoundException");
     } catch (FileNotFoundException ignored) {}
+    cluster.getNameNodeRpc().rename(qux.toString(), "/temp/qux");
+    summary = cluster.getNameNodeRpc().getContentSummary(
+        foo.toString());
+    Assert.assertEquals(0, summary.getSnapshotDirectoryCount());
+    Assert.assertEquals(2, summary.getSnapshotFileCount());
+    Assert.assertEquals(30, summary.getSnapshotLength());
+    Assert.assertEquals(2, summary.getDirectoryCount());
+    Assert.assertEquals(2, summary.getFileCount());
+    Assert.assertEquals(30, summary.getLength());
+
   }
 }