HDFS-4800. Fix INodeDirectoryWithSnapshot#cleanDeletedINode. Contributed by Jing Zhao

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/HDFS-2802@1479707 13f79535-47bb-0310-9956-ffa450edef68

hadoop-hdfs-project/hadoop-hdfs/CHANGES.HDFS-2802.txt

@@ -338,3 +338,6 @@ Branch-2802 Snapshot (Unreleased)
 
   HDFS-4798. Update computeContentSummary() for the reference nodes in
   snapshots.  (szetszwo)
+
+  HDFS-4800. Fix INodeDirectoryWithSnapshot#cleanDeletedINode.  (Jing Zhao via
+  szetszwo)

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

@@ -482,15 +482,18 @@ public abstract class INodeReference extends INode {
     }
     
     @Override
-    public Quota.Counts cleanSubtree(Snapshot snapshot, Snapshot prior,
+    public Quota.Counts cleanSubtree(final Snapshot snapshot, Snapshot prior,
         final BlocksMapUpdateInfo collectedBlocks,
         final List<INode> removedINodes) throws QuotaExceededException {
       // since WithName node resides in deleted list acting as a snapshot copy,
       // the parameter snapshot must be non-null
       Preconditions.checkArgument(snapshot != null);
-      // if prior is null, or if prior's id is <= dstSnapshotId, we will call
-      // destroyAndCollectBlocks method
-      Preconditions.checkArgument(prior != null);
+      // if prior is null, we need to check snapshot belonging to the previous
+      // WithName instance
+      if (prior == null) {
+        prior = getPriorSnapshot(this);
+      }
+
       Quota.Counts counts = getReferredINode().cleanSubtree(snapshot, prior,
           collectedBlocks, removedINodes);
       INodeReference ref = getReferredINode().getParentReference();
@@ -504,17 +507,26 @@ public abstract class INodeReference extends INode {
     @Override
     public void destroyAndCollectBlocks(BlocksMapUpdateInfo collectedBlocks,
         final List<INode> removedINodes) {
+      Snapshot snapshot = getSelfSnapshot();
       if (removeReference(this) <= 0) {
         getReferredINode().destroyAndCollectBlocks(collectedBlocks,
             removedINodes);
       } else {
         Snapshot prior = getPriorSnapshot(this);
         INode referred = getReferredINode().asReference().getReferredINode();
-        Snapshot snapshot = getSelfSnapshot();
         
         if (snapshot != null) {
-          Preconditions.checkState(prior == null || 
-              snapshot.getId() > prior.getId());
+          if (prior != null && snapshot.getId() <= prior.getId()) {
+            // the snapshot to be deleted has been deleted while traversing 
+            // the src tree of the previous rename operation. This usually 
+            // happens when rename's src and dst are under the same 
+            // snapshottable directory. E.g., the following operation sequence:
+            // 1. create snapshot s1 on /test
+            // 2. rename /test/foo/bar to /test/foo2/bar
+            // 3. create snapshot s2 on /test
+            // 4. delete snapshot s2
+            return;
+          }
           try {
             Quota.Counts counts = referred.cleanSubtree(snapshot, prior,
                 collectedBlocks, removedINodes);
@@ -579,6 +591,14 @@ public abstract class INodeReference extends INode {
         destroyAndCollectBlocks(collectedBlocks, removedINodes);
         return counts;
       } else {
+        // if prior is null, we need to check snapshot belonging to the previous
+        // WithName instance
+        if (prior == null) {
+          prior = getPriorSnapshot(this);
+        }
+        if (snapshot != null && snapshot.equals(prior)) {
+          return Quota.Counts.newInstance();
+        }
         return getReferredINode().cleanSubtree(snapshot, prior,
             collectedBlocks, removedINodes);
       }

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

@@ -752,7 +752,15 @@ public class INodeDirectoryWithSnapshot extends INodeDirectoryWithQuota {
     queue.addLast(inode);
     while (!queue.isEmpty()) {
       INode topNode = queue.pollFirst();
-      if (topNode.isFile() && topNode.asFile() instanceof FileWithSnapshot) {
+      if (topNode instanceof INodeReference.WithName) {
+        INodeReference.WithName wn = (INodeReference.WithName) topNode;
+        if (wn.getLastSnapshotId() >= post.getId()) {
+          wn.cleanSubtree(post, prior, collectedBlocks, removedINodes);
+        }
+        // For DstReference node, since the node is not in the created list of
+        // prior, we should treat it as regular file/dir
+      } else if (topNode.isFile()
+          && topNode.asFile() instanceof FileWithSnapshot) {
         FileWithSnapshot fs = (FileWithSnapshot) topNode.asFile();
         counts.add(fs.getDiffs().deleteSnapshotDiff(post, prior,
             topNode.asFile(), collectedBlocks, removedINodes));
@@ -843,7 +851,7 @@ public class INodeDirectoryWithSnapshot extends INodeDirectoryWithQuota {
       final List<INode> removedINodes) throws QuotaExceededException {
     Preconditions.checkArgument(prior != null);
     if (inode.isReference()) {
-      if (inode instanceof INodeReference.WithName) {
+      if (inode instanceof INodeReference.WithName && snapshot != null) {
         // this inode has been renamed before the deletion of the DstReference
         // subtree
         inode.cleanSubtree(snapshot, prior, collectedBlocks, removedINodes);