HDFS-14504. Rename with Snapshots does not honor quota limit.

(cherry picked from commit 83fd15b412)

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

@@ -1199,7 +1199,8 @@ static void verifyQuota(INodesInPath iip, int pos, QuotaCounts deltas,
 
     // check existing components in the path
     for(int i = (pos > iip.length() ? iip.length(): pos) - 1; i >= 0; i--) {
-      if (commonAncestor == iip.getINode(i)) {
+      if (commonAncestor == iip.getINode(i)
+          && !commonAncestor.isInLatestSnapshot(iip.getLatestSnapshotId())) {
         // Stop checking for quota when common ancestor is reached
         return;
       }

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

@@ -33,6 +33,7 @@
 import org.apache.hadoop.hdfs.DistributedFileSystem;
 import org.apache.hadoop.hdfs.MiniDFSCluster;
 import org.apache.hadoop.hdfs.protocol.HdfsConstants;
+import org.apache.hadoop.hdfs.protocol.NSQuotaExceededException;
 import org.apache.hadoop.hdfs.protocol.SnapshotException;
 import org.apache.hadoop.hdfs.server.namenode.FSDirectory;
 import org.apache.hadoop.hdfs.server.namenode.FSNamesystem;
@@ -41,6 +42,7 @@
 import org.apache.hadoop.hdfs.util.ReadOnlyList;
 import org.apache.hadoop.ipc.RemoteException;
 import org.apache.hadoop.test.GenericTestUtils;
+import org.apache.hadoop.test.LambdaTestUtils;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Rule;
@@ -280,4 +282,150 @@ public void testRenameSnapshotCommandWithIllegalArguments() throws Exception {
       System.setErr(oldErr);
     }
   }
+
+  // Test rename of a snapshotted by setting quota in same directory.
+  @Test
+  public void testQuotaAndRenameWithSnapshot() throws Exception {
+    String dirr = "/dir2";
+    Path dir2 = new Path(dirr);
+    Path fil1 = new Path(dir2, "file1");
+    hdfs.mkdirs(dir2);
+    hdfs.setQuota(dir2, 3, 0);
+    hdfs.allowSnapshot(dir2);
+    hdfs.create(fil1);
+    hdfs.createSnapshot(dir2, "snap1");
+    Path file2 = new Path(dir2, "file2");
+    hdfs.rename(fil1, file2);
+    hdfs.getFileStatus(dir2);
+    Path filex = new Path(dir2, "filex");
+    // create a file after exceeding namespace quota
+    LambdaTestUtils.intercept(NSQuotaExceededException.class,
+        "The NameSpace quota (directories and files) of "
+            + "directory /dir2 is exceeded",
+        () -> hdfs.create(filex));
+    hdfs.createSnapshot(dir2, "snap2");
+    // rename a file after exceeding namespace quota
+    Path file3 = new Path(dir2, "file3");
+    LambdaTestUtils
+        .intercept(NSQuotaExceededException.class,
+            "The NameSpace quota (directories and files) of"
+                + " directory /dir2 is exceeded",
+            () -> hdfs.rename(file2, file3));
+  }
+
+  // Test Rename across directories within snapshot with quota set on source
+  // directory.
+  @Test
+  public void testRenameAcrossDirWithinSnapshot() throws Exception {
+    // snapshottable directory
+    String dirr = "/dir";
+    Path rootDir = new Path(dirr);
+    hdfs.mkdirs(rootDir);
+    hdfs.allowSnapshot(rootDir);
+
+    // set quota for source directory under snapshottable root directory
+    Path dir2 = new Path(rootDir, "dir2");
+    Path fil1 = new Path(dir2, "file1");
+    hdfs.mkdirs(dir2);
+    hdfs.setQuota(dir2, 3, 0);
+    hdfs.create(fil1);
+    Path file2 = new Path(dir2, "file2");
+    hdfs.rename(fil1, file2);
+    Path fil3 = new Path(dir2, "file3");
+    hdfs.create(fil3);
+
+    // destination directory under snapshottable root directory
+    Path dir1 = new Path(rootDir, "dir1");
+    Path dir1fil1 = new Path(dir1, "file1");
+    hdfs.mkdirs(dir1);
+    hdfs.create(dir1fil1);
+    Path dir1fil2 = new Path(dir1, "file2");
+    hdfs.rename(dir1fil1, dir1fil2);
+
+    hdfs.createSnapshot(rootDir, "snap1");
+    Path filex = new Path(dir2, "filex");
+    // create a file after exceeding namespace quota
+    LambdaTestUtils.intercept(NSQuotaExceededException.class,
+        "The NameSpace quota (directories and files) of "
+            + "directory /dir/dir2 is exceeded",
+        () -> hdfs.create(filex));
+
+    // Rename across directories within snapshot with quota set on source
+    // directory
+    assertTrue(hdfs.rename(fil3, dir1));
+  }
+
+  // Test rename within the same directory within a snapshottable root with
+  // quota set.
+  @Test
+  public void testRenameInSameDirWithSnapshotableRoot() throws Exception {
+
+    // snapshottable root directory
+    String rootStr = "/dir";
+    Path rootDir = new Path(rootStr);
+    hdfs.mkdirs(rootDir);
+    hdfs.setQuota(rootDir, 3, 0);
+    hdfs.allowSnapshot(rootDir);
+
+    // rename to be performed directory
+    String dirr = "dir2";
+    Path dir2 = new Path(rootDir, dirr);
+    Path fil1 = new Path(dir2, "file1");
+    hdfs.mkdirs(dir2);
+    hdfs.create(fil1);
+    hdfs.createSnapshot(rootDir, "snap1");
+    Path file2 = new Path(dir2, "file2");
+    // rename a file after exceeding namespace quota
+    LambdaTestUtils
+        .intercept(NSQuotaExceededException.class,
+            "The NameSpace quota (directories and files) of"
+                + " directory /dir is exceeded",
+            () -> hdfs.rename(fil1, file2));
+
+  }
+
+  // Test rename from a directory under snapshottable root to a directory with
+  // quota set to a directory not under under any snapshottable root.
+  @Test
+  public void testRenameAcrossDirWithSnapshotableSrc() throws Exception {
+    // snapshottable directory
+    String dirr = "/dir";
+    Path rootDir = new Path(dirr);
+    hdfs.mkdirs(rootDir);
+    hdfs.allowSnapshot(rootDir);
+
+    // set quota for source directory
+    Path dir2 = new Path(rootDir, "dir2");
+    Path fil1 = new Path(dir2, "file1");
+    hdfs.mkdirs(dir2);
+    hdfs.setQuota(dir2, 3, 0);
+    hdfs.create(fil1);
+    Path file2 = new Path(dir2, "file2");
+    hdfs.rename(fil1, file2);
+    Path fil3 = new Path(dir2, "file3");
+    hdfs.create(fil3);
+
+    hdfs.createSnapshot(rootDir, "snap1");
+
+    // destination directory not under any snapshot
+    String dirr1 = "/dir1";
+    Path dir1 = new Path(dirr1);
+    Path dir1fil1 = new Path(dir1, "file1");
+    hdfs.mkdirs(dir1);
+    hdfs.create(dir1fil1);
+    Path dir1fil2 = new Path(dir1, "file2");
+    hdfs.rename(dir1fil1, dir1fil2);
+
+    Path filex = new Path(dir2, "filex");
+    // create a file after exceeding namespace quota on source
+    LambdaTestUtils.intercept(NSQuotaExceededException.class,
+        "The NameSpace quota (directories and files) of "
+            + "directory /dir/dir2 is exceeded",
+        () -> hdfs.create(filex));
+
+    // Rename across directories source dir under snapshot with quota set and
+    // destination directory not under any snapshot
+    assertTrue(hdfs.rename(fil3, dir1));
+  }
+
 }