HDFS-17045. File renamed from a snapshottable dir to a non-snapshottable dir cannot be deleted. (#5738)

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

@@ -831,17 +831,17 @@ public abstract class INodeReference extends INode {
      *
      * @param context to {@link ReclaimContext#getSnapshotIdToBeDeleted()}
      */
-    private boolean shouldDestroy(ReclaimContext context) {
+    private void shouldDestroy(ReclaimContext context) {
       final int snapshotToBeDeleted = context.getSnapshotIdToBeDeleted();
-      if (dstSnapshotId < snapshotToBeDeleted) {
-        return true;
+      if (snapshotToBeDeleted == Snapshot.CURRENT_STATE_ID
+          || snapshotToBeDeleted > dstSnapshotId) {
+        return;
       }
       LOG.warn("Try to destroy a DstReference with dstSnapshotId = {}"
           + " >= snapshotToBeDeleted = {}", dstSnapshotId, snapshotToBeDeleted);
       LOG.warn("    dstRef: {}", toDetailString());
       final INode r = getReferredINode().asReference().getReferredINode();
       LOG.warn("  referred: {}", r.toDetailString());
-      return false;
     }
 
     /**
@@ -857,9 +857,7 @@ public abstract class INodeReference extends INode {
      */
     @Override
     public void destroyAndCollectBlocks(ReclaimContext reclaimContext) {
-      if (!shouldDestroy(reclaimContext)) {
-        return;
-      }
+      shouldDestroy(reclaimContext);
 
       // since we count everything of the subtree for the quota usage of a
       // dst reference node, here we should just simply do a quota computation.

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

@@ -163,6 +163,7 @@ public class TestRenameWithSnapshots {
     
     hdfs.delete(bar, false);
     Assert.assertEquals(1, withCount.getReferenceCount());
+    restartClusterAndCheckImage(true);
   }
   
   private static boolean existsInDiffReport(List<DiffReportEntry> entries,
@@ -196,6 +197,7 @@ public class TestRenameWithSnapshots {
     assertTrue(existsInDiffReport(entries, DiffType.MODIFY, "", null));
     assertTrue(existsInDiffReport(entries, DiffType.CREATE, file2.getName(),
         null));
+    restartClusterAndCheckImage(true);
   }
 
   /**
@@ -218,6 +220,7 @@ public class TestRenameWithSnapshots {
     assertTrue(existsInDiffReport(entries, DiffType.MODIFY, "", null));
     assertTrue(existsInDiffReport(entries, DiffType.RENAME, file1.getName(),
         file2.getName()));
+    restartClusterAndCheckImage(true);
   }
 
   @Test (timeout=60000)