add snapshot ut

Author: huangzhaobo99

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

@@ -673,7 +673,7 @@ private static class RenameOperation {
     private INode oldDstChild;
 
     RenameOperation(FSDirectory fsd, INodesInPath srcIIP, INodesInPath dstIIP,
-        Triple<Boolean, Optional<QuotaCounts>, Optional<QuotaCounts>> quotaPair) {
+        Triple<Boolean, Optional<QuotaCounts>, Optional<QuotaCounts>> countTriple) {
       this.fsd = fsd;
       this.srcIIP = srcIIP;
       this.dstIIP = dstIIP;
@@ -723,10 +723,10 @@ private static class RenameOperation {
         withCount = null;
       }
       // Set quota for src and dst, ignore src is in Snapshot or is Reference
-      this.updateQuota = quotaPair.getLeft() || isSrcInSnapshot || srcChildIsReference;
+      this.updateQuota = countTriple.getLeft() || withCount != null;
       this.srcSubTreeCount = withCount == null ?
-          quotaPair.getMiddle() : Optional.empty();
-      this.dstSubTreeCount = quotaPair.getRight();
+          countTriple.getMiddle() : Optional.empty();
+      this.dstSubTreeCount = countTriple.getRight();
     }
 
     boolean isSameStoragePolicy() {

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

@@ -219,7 +219,6 @@ public void testRenameWithoutValidFeature() throws Exception {
     long expectedCount =
         dfs.getQuotaUsage(new Path("/")).getFileAndDirectoryCount() - originDstDirUsage;
     ContentSummary contentSummary3 = dfs.getContentSummary(testParentDir2);
-
     // Src and dst must be same
     // dstDir2=/testRename/testDir2/dst-dir
     // dstDir3=/testRename/testDir3/dst-dir
@@ -303,10 +302,12 @@ public void testRenameUndoWithoutValidFeature() throws Exception {
     Mockito.doReturn(false).when(mockDir4).addChild(Mockito.eq(srcInode3), anyBoolean(), anyInt());
     rootDir.replaceChild(dir4, mockDir4, fsDirectory.getINodeMap());
     mockDir4.setParent(rootDir);
+
     // srcDir=/testRenameUndo/testDir3/src-dir
     // dstDir=/testRenameUndo/testDir4/src-dir dstDir exist
     assertThrows(RemoteException.class,
         () -> dfs.rename(srcDir3, dstDir4, Options.Rename.OVERWRITE));
+
     ContentSummary rootContentSummary4 = dfs.getContentSummary(new Path("/"));
     QuotaUsage rootQuotaUsage4 = dfs.getQuotaUsage(new Path("/"));
     ContentSummary contentSummary4 = dfs.getContentSummary(testParentDir3);
@@ -316,4 +317,47 @@ public void testRenameUndoWithoutValidFeature() throws Exception {
     assertEquals(rootContentSummary3.getFileAndDirectoryCount(),
         rootQuotaUsage4.getFileAndDirectoryCount());
   }
+
+  @Test
+  public void testRenameFileInSnapshotDirWithoutValidFeature() throws Exception {
+    final int fileLen = 1024;
+    final short replication = 3;
+    final Path root = new Path("/testRenameSrcWithSnapshot");
+    assertTrue(dfs.mkdirs(root));
+
+    Path testParentDir1 = new Path(root, "testDir1");
+    assertTrue(dfs.mkdirs(testParentDir1));
+    for (int i = 0; i < 2; i++) {
+      Path file = new Path(testParentDir1, "file" + i);
+      DFSTestUtil.createFile(dfs, file, fileLen, replication, 0); 
+    }
+    dfs.allowSnapshot(testParentDir1);
+    dfs.createSnapshot(testParentDir1, "snapshot1");
+
+    Path testParentDir2 = new Path(root, "testDir2");
+    assertTrue(dfs.mkdirs(testParentDir2));
+    Path testParentDir3 = new Path(root, "testDir3");
+    assertTrue(dfs.mkdirs(testParentDir3));
+
+    // Test rename1
+    ContentSummary contentSummary1 = dfs.getContentSummary(new Path("/"));
+    QuotaUsage quotaUsage1 = dfs.getQuotaUsage(new Path("/"));
+    assertEquals(contentSummary1.getSpaceConsumed(), quotaUsage1.getSpaceConsumed());
+    assertEquals(contentSummary1.getFileAndDirectoryCount(),
+        quotaUsage1.getFileAndDirectoryCount());
+
+    // The snapshot of file1 not be cleaned up.
+    assertTrue(dfs.rename(new Path(testParentDir1, "file0"), testParentDir2));
+
+    ContentSummary contentSummary2 = dfs.getContentSummary(new Path("/"));
+    QuotaUsage quotaUsage2 = dfs.getQuotaUsage(new Path("/"));
+    assertEquals(quotaUsage1.getFileAndDirectoryCount() + 1,
+        quotaUsage2.getFileAndDirectoryCount());
+    assertEquals(quotaUsage1.getSpaceConsumed() + fileLen * replication,
+        quotaUsage2.getSpaceConsumed());
+    // Root directory's actual capacity must match quota usage.
+    assertEquals(contentSummary2.getFileAndDirectoryCount(),
+        quotaUsage2.getFileAndDirectoryCount());
+    assertEquals(contentSummary2.getSpaceConsumed(), quotaUsage2.getSpaceConsumed());
+  }
 }