From 6428612f35074e1de88f6338e009d551935294fb Mon Sep 17 00:00:00 2001
From: Tsz-wo Sze <szetszwo@apache.org>
Date: Fri, 31 May 2013 05:13:32 +0000
Subject: [PATCH] svn merge -c 1487698 from trunk for HDFS-4863. The root
 directory should be added to the snapshottable directory list while loading
 fsimage.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/branch-2@1488096 13f79535-47bb-0310-9956-ffa450edef68
---
 hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt     |  3 +++
 .../hdfs/server/namenode/FSImageFormat.java     | 11 +++++++----
 .../namenode/TestFSImageWithSnapshot.java       | 17 +++++++++++++++++
 3 files changed, 27 insertions(+), 4 deletions(-)

diff --git a/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt b/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
index 8b417379bde..85b390f91e4 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
+++ b/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
@@ -777,6 +777,9 @@ Release 2.0.5-beta - UNRELEASED
     HDFS-4857. Snapshot.Root and AbstractINodeDiff#snapshotINode should not be 
     put into INodeMap when loading FSImage. (jing9)
 
+    HDFS-4863. The root directory should be added to the snapshottable 
+    directory list while loading fsimage. (jing9)
+
 Release 2.0.4-alpha - 2013-04-25
 
   INCOMPATIBLE CHANGES
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSImageFormat.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSImageFormat.java
index 6e2509c19dd..22b4c6b49b1 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSImageFormat.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSImageFormat.java
@@ -435,13 +435,16 @@ private void loadDirectoryWithSnapshot(DataInput in)
       if (numSnapshots >= 0) {
         final INodeDirectorySnapshottable snapshottableParent
             = INodeDirectorySnapshottable.valueOf(parent, parent.getLocalName());
-        if (snapshottableParent.getParent() != null) { // not root
-          this.namesystem.getSnapshotManager().addSnapshottable(
-              snapshottableParent);
-        }
         // load snapshots and snapshotQuota
         SnapshotFSImageFormat.loadSnapshotList(snapshottableParent,
             numSnapshots, in, this);
+        if (snapshottableParent.getSnapshotQuota() > 0) {
+          // add the directory to the snapshottable directory list in 
+          // SnapshotManager. Note that we only add root when its snapshot quota
+          // is positive.
+          this.namesystem.getSnapshotManager().addSnapshottable(
+              snapshottableParent);
+        }
       }
 
       // Step 3. Load children nodes under parent
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestFSImageWithSnapshot.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestFSImageWithSnapshot.java
index aece69a4e03..eb6191469d7 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestFSImageWithSnapshot.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestFSImageWithSnapshot.java
@@ -200,6 +200,23 @@ public void testSnapshotOnRoot() throws Exception {
     List<DirectoryDiff> diffList = rootNode.getDiffs().asList();
     assertEquals(1, diffList.size());
     assertEquals("s1", diffList.get(0).getSnapshot().getRoot().getLocalName());
+    
+    // check SnapshotManager's snapshottable directory list
+    assertEquals(1, fsn.getSnapshotManager().getNumSnapshottableDirs());
+    SnapshottableDirectoryStatus[] sdirs = fsn.getSnapshotManager()
+        .getSnapshottableDirListing(null);
+    assertEquals(root, sdirs[0].getFullPath());
+    
+    // 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(REPLICATION).build();
+    cluster.waitActive();
+    fsn = cluster.getNamesystem();
+    hdfs = cluster.getFileSystem();
   }
 
   /**