HDFS-14492. Snapshot memory leak. Contributed by Wei-Chiu Chuang. (#1370)

* HDFS-14492. Snapshot memory leak. Contributed by Wei-Chiu Chuang.

Change-Id: I9e5e450c07ad70aa1905973896c4f627042dbd37

* Fix checkstyle

Change-Id: I16d4bd4f03a971e1ed36cf57d89dc42357ef8fbf
(cherry picked from commit 6ef6594c7e)
(cherry picked from commit 570ffa1cd6)

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

@@ -838,6 +838,12 @@ public class INodeDirectory extends INodeWithAdditionalFields
     // there is snapshot data
     if (sf != null) {
       sf.cleanDirectory(reclaimContext, this, snapshotId, priorSnapshotId);
+      // If the inode has empty diff list and sf is not a
+      // DirectorySnapshottableFeature, remove the feature to save heap.
+      if (sf.getDiffs().isEmpty() &&
+          !(sf instanceof DirectorySnapshottableFeature)) {
+        this.removeFeature(sf);
+      }
     } else {
       // there is no snapshot data
       if (priorSnapshotId == Snapshot.NO_SNAPSHOT_ID &&

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

@@ -744,6 +744,9 @@ public class INodeFile extends INodeWithAdditionalFields
       sf.cleanFile(reclaimContext, this, snapshot, priorSnapshotId,
           getStoragePolicyID());
       updateRemovedUnderConstructionFiles(reclaimContext);
+      if (sf.getDiffs().isEmpty()) {
+        this.removeFeature(sf);
+      }
     } else {
       if (snapshot == CURRENT_STATE_ID) {
         if (priorSnapshotId == NO_SNAPSHOT_ID) {

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

@@ -46,6 +46,10 @@ abstract class AbstractINodeDiffList<N extends INode,
         DiffList.unmodifiableList(diffs) : DiffList.emptyList();
   }
 
+  public boolean isEmpty() {
+    return diffs == null || diffs.isEmpty();
+  }
+  
   /** Clear the list. */
   public void clear() {
     diffs = null;

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

@@ -2141,24 +2141,11 @@ public class TestRenameWithSnapshots {
     INodeDirectory dir2Node = fsdir.getINode4Write(dir2.toString())
         .asDirectory();
     assertTrue("the diff list of " + dir2
-        + " should be empty after deleting s0", dir2Node.getDiffs().asList()
-        .isEmpty());
+        + " should be empty after deleting s0", !dir2Node.isWithSnapshot());
     
     assertTrue(hdfs.exists(newfoo));
     INode fooRefNode = fsdir.getINode4Write(newfoo.toString());
     assertTrue(fooRefNode instanceof INodeReference.DstReference);
-    INodeDirectory fooNode = fooRefNode.asDirectory();
-    // fooNode should be still INodeDirectory (With Snapshot) since we call
-    // recordModification before the rename
-    assertTrue(fooNode.isWithSnapshot());
-    assertTrue(fooNode.getDiffs().asList().isEmpty());
-    INodeDirectory barNode = fooNode.getChildrenList(Snapshot.CURRENT_STATE_ID)
-        .get(0).asDirectory();
-    // bar should also be INodeDirectory (With Snapshot), and both of its diff 
-    // list and children list are empty 
-    assertTrue(barNode.getDiffs().asList().isEmpty());
-    assertTrue(barNode.getChildrenList(Snapshot.CURRENT_STATE_ID).isEmpty());
-    
     restartClusterAndCheckImage(true);
   }
   

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

@@ -527,7 +527,7 @@ public class TestSnapshotDeletion {
     assertEquals(snapshot1.getId(), diffList.getLast().getSnapshotId());
     diffList = fsdir.getINode(metaChangeDir.toString()).asDirectory()
         .getDiffs();
-    assertEquals(0, diffList.asList().size());
+    assertEquals(null, diffList);
     
     // check 2. noChangeDir and noChangeFile are still there
     final INodeDirectory noChangeDirNode =