HDFS-15470. Added more unit tests to validate rename behaviour across snapshots. Contributed by Shashikant Banerjee.

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

@@ -649,4 +649,182 @@ public class TestFSImageWithSnapshot {
     fsn = cluster.getNamesystem();
     hdfs = cluster.getFileSystem();
   }
+
+
+  @Test (timeout=60000)
+  public void testFSImageWithRename1() throws Exception {
+    final Path dir1 = new Path("/dir1");
+    final Path dir2 = new Path("/dir2");
+    hdfs.mkdirs(dir1);
+    hdfs.mkdirs(dir2);
+    Path dira = new Path(dir1, "dira");
+    Path dirx = new Path(dir1, "dirx");
+    Path dirb = new Path(dirx, "dirb");
+    hdfs.mkdirs(dira);
+    hdfs.mkdirs(dirx);
+    hdfs.allowSnapshot(dir1);
+    hdfs.createSnapshot(dir1, "s0");
+    hdfs.mkdirs(dirb);
+    hdfs.createSnapshot(dir1, "s1");
+    Path rennamePath = new Path(dira, "dirb");
+    // mv /dir1/dirx/dirb to /dir1/dira/dirb
+    hdfs.rename(dirb, rennamePath);
+    hdfs.createSnapshot(dir1, "s2");
+    Path diry = new Path("/dir1/dira/dirb/diry");
+    hdfs.mkdirs(diry);
+    hdfs.createSnapshot(dir1, "s3");
+    Path file1 = new Path("/dir1/dira/dirb/diry/file1");
+    DFSTestUtil.createFile(hdfs, file1, BLOCKSIZE, (short) 1, seed);
+    hdfs.createSnapshot(dir1, "s4");
+    hdfs.delete(new Path("/dir1/dira/dirb"), true);
+    hdfs.deleteSnapshot(dir1, "s1");
+    hdfs.deleteSnapshot(dir1, "s3");
+    // file1 should exist in the last snapshot
+    assertTrue(hdfs.exists(
+        new Path("/dir1/.snapshot/s4/dira/dirb/diry/file1")));
+
+    // save namespace and restart cluster
+    hdfs.setSafeMode(SafeModeAction.SAFEMODE_ENTER);
+    hdfs.saveNamespace();
+    hdfs.setSafeMode(SafeModeAction.SAFEMODE_LEAVE);
+
+    cluster.shutdown();
+    cluster = new MiniDFSCluster.Builder(conf).format(false)
+            .numDataNodes(NUM_DATANODES).build();
+    cluster.waitActive();
+    fsn = cluster.getNamesystem();
+    hdfs = cluster.getFileSystem();
+  }
+
+  @Test (timeout=60000)
+  public void testFSImageWithRename2() throws Exception {
+    final Path dir1 = new Path("/dir1");
+    final Path dir2 = new Path("/dir2");
+    hdfs.mkdirs(dir1);
+    hdfs.mkdirs(dir2);
+    Path dira = new Path(dir1, "dira");
+    Path dirx = new Path(dir1, "dirx");
+    Path dirb = new Path(dirx, "dirb");
+    hdfs.mkdirs(dira);
+    hdfs.mkdirs(dirx);
+    hdfs.allowSnapshot(dir1);
+    hdfs.createSnapshot(dir1, "s0");
+    hdfs.mkdirs(dirb);
+    hdfs.createSnapshot(dir1, "s1");
+    Path rennamePath = new Path(dira, "dirb");
+    // mv /dir1/dirx/dirb to /dir1/dira/dirb
+    hdfs.rename(dirb, rennamePath);
+    hdfs.createSnapshot(dir1, "s2");
+    Path file1 = new Path("/dir1/dira/dirb/file1");
+    DFSTestUtil.createFile(hdfs,
+            new Path(
+                    "/dir1/dira/dirb/file1"), BLOCKSIZE, (short) 1, seed);
+    hdfs.createSnapshot(dir1, "s3");
+    hdfs.deleteSnapshot(dir1, "s1");
+    hdfs.deleteSnapshot(dir1, "s3");
+    assertTrue(hdfs.exists(file1));
+
+    // save namespace and restart cluster
+    hdfs.setSafeMode(SafeModeAction.SAFEMODE_ENTER);
+    hdfs.saveNamespace();
+    hdfs.setSafeMode(SafeModeAction.SAFEMODE_LEAVE);
+
+    cluster.shutdown();
+    cluster = new MiniDFSCluster.Builder(conf).format(false)
+            .numDataNodes(NUM_DATANODES).build();
+    cluster.waitActive();
+    fsn = cluster.getNamesystem();
+    hdfs = cluster.getFileSystem();
+  }
+
+  @Test(timeout = 60000)
+  public void testFSImageWithRename3() throws Exception {
+    final Path dir1 = new Path("/dir1");
+    final Path dir2 = new Path("/dir2");
+    hdfs.mkdirs(dir1);
+    hdfs.mkdirs(dir2);
+    Path dira = new Path(dir1, "dira");
+    Path dirx = new Path(dir1, "dirx");
+    Path dirb = new Path(dirx, "dirb");
+    hdfs.mkdirs(dira);
+    hdfs.mkdirs(dirx);
+    hdfs.allowSnapshot(dir1);
+    hdfs.createSnapshot(dir1, "s0");
+    hdfs.mkdirs(dirb);
+    hdfs.createSnapshot(dir1, "s1");
+    Path rennamePath = new Path(dira, "dirb");
+    // mv /dir1/dirx/dirb to /dir1/dira/dirb
+    hdfs.rename(dirb, rennamePath);
+    hdfs.createSnapshot(dir1, "s2");
+    Path diry = new Path("/dir1/dira/dirb/diry");
+    hdfs.mkdirs(diry);
+    hdfs.createSnapshot(dir1, "s3");
+    Path file1 = new Path("/dir1/dira/dirb/diry/file1");
+    DFSTestUtil.createFile(hdfs, file1, BLOCKSIZE, (short) 1, seed);
+    hdfs.createSnapshot(dir1, "s4");
+    hdfs.delete(new Path("/dir1/dira/dirb"), true);
+    hdfs.deleteSnapshot(dir1, "s1");
+    hdfs.deleteSnapshot(dir1, "s3");
+    // file1 should exist in the last snapshot
+    assertTrue(hdfs.exists(new Path(
+        "/dir1/.snapshot/s4/dira/dirb/diry/file1")));
+
+    // save namespace and restart cluster
+    hdfs.setSafeMode(SafeModeAction.SAFEMODE_ENTER);
+    hdfs.saveNamespace();
+    hdfs.setSafeMode(SafeModeAction.SAFEMODE_LEAVE);
+
+    cluster.shutdown();
+    cluster = new MiniDFSCluster.Builder(conf).format(false)
+            .numDataNodes(NUM_DATANODES).build();
+    cluster.waitActive();
+    fsn = cluster.getNamesystem();
+    hdfs = cluster.getFileSystem();
+  }
+
+  @Test (timeout=60000)
+  public void testFSImageWithRename4() throws Exception {
+    final Path dir1 = new Path("/dir1");
+    final Path dir2 = new Path("/dir2");
+    hdfs.mkdirs(dir1);
+    hdfs.mkdirs(dir2);
+    Path dira = new Path(dir1, "dira");
+    Path dirx = new Path(dir1, "dirx");
+    Path dirb = new Path(dirx, "dirb");
+    hdfs.mkdirs(dira);
+    hdfs.mkdirs(dirx);
+    hdfs.allowSnapshot(dir1);
+    hdfs.createSnapshot(dir1, "s0");
+    hdfs.mkdirs(dirb);
+    hdfs.createSnapshot(dir1, "s1");
+    Path renamePath = new Path(dira, "dirb");
+    // mv /dir1/dirx/dirb to /dir1/dira/dirb
+    hdfs.rename(dirb, renamePath);
+    hdfs.createSnapshot(dir1, "s2");
+    Path diry = new Path("/dir1/dira/dirb/diry");
+    hdfs.mkdirs(diry);
+    hdfs.createSnapshot(dir1, "s3");
+    Path file1 = new Path("/dir1/dira/dirb/diry/file1");
+    DFSTestUtil.createFile(hdfs, file1, BLOCKSIZE, (short) 1, seed);
+    hdfs.createSnapshot(dir1, "s4");
+    hdfs.delete(new Path("/dir1/dira/dirb/diry/file1"), false);
+    hdfs.deleteSnapshot(dir1, "s1");
+    hdfs.deleteSnapshot(dir1, "s3");
+    // file1 should exist in the last snapshot
+    assertTrue(hdfs.exists(
+        new Path("/dir1/.snapshot/s4/dira/dirb/diry/file1")));
+
+    // save namespace and restart cluster
+    hdfs.setSafeMode(SafeModeAction.SAFEMODE_ENTER);
+    hdfs.saveNamespace();
+    hdfs.setSafeMode(SafeModeAction.SAFEMODE_LEAVE);
+
+    cluster.shutdown();
+    cluster = new MiniDFSCluster.Builder(conf).format(false)
+            .numDataNodes(NUM_DATANODES).build();
+    cluster.waitActive();
+    fsn = cluster.getNamesystem();
+    hdfs = cluster.getFileSystem();
+  }
+
 }