HDFS-9470. Encryption zone on root not loaded from fsimage after NN restart. Xiao Chen via wang.

(cherry picked from commit 9b8e50b424)
(cherry picked from commit ce1111ceea)

 Conflicts:
	hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirectory.java

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

@@ -1172,6 +1172,9 @@ Release 2.6.3 - UNRELEASED
     HDFS-9434. Recommission a datanode with 500k blocks may pause NN for 30
     seconds for printing info log messags.  (szetszwo)
 
+    HDFS-9470. Encryption zone on root not loaded from fsimage after NN
+    restart. (Xiao Chen via wang)
+
 Release 2.6.2 - 2015-10-28
 
   INCOMPATIBLE CHANGES

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

@@ -1163,30 +1163,44 @@ public class FSDirectory implements Closeable {
       inodeMap.put(inode);
       if (!inode.isSymlink()) {
         final XAttrFeature xaf = inode.getXAttrFeature();
-        if (xaf != null) {
+        addEncryptionZone((INodeWithAdditionalFields) inode, xaf);
-          final List<XAttr> xattrs = xaf.getXAttrs();
+      }
-          for (XAttr xattr : xattrs) {
+    }
-            final String xaName = XAttrHelper.getPrefixName(xattr);
+  }
-            if (CRYPTO_XATTR_ENCRYPTION_ZONE.equals(xaName)) {
+
-              try {
+  private void addEncryptionZone(INodeWithAdditionalFields inode,
-                final HdfsProtos.ZoneEncryptionInfoProto ezProto =
+      XAttrFeature xaf) {
-                    HdfsProtos.ZoneEncryptionInfoProto.parseFrom(
+    if (xaf == null) {
-                        xattr.getValue());
+      return;
-                ezManager.unprotectedAddEncryptionZone(inode.getId(),
+    }
-                    PBHelper.convert(ezProto.getSuite()),
+    final List<XAttr> xattrs = xaf.getXAttrs();
-                    PBHelper.convert(ezProto.getCryptoProtocolVersion()),
+    for (XAttr xattr : xattrs) {
-                    ezProto.getKeyName());
+      final String xaName = XAttrHelper.getPrefixName(xattr);
-              } catch (InvalidProtocolBufferException e) {
+      if (CRYPTO_XATTR_ENCRYPTION_ZONE.equals(xaName)) {
-                NameNode.LOG.warn("Error parsing protocol buffer of " +
+        try {
-                    "EZ XAttr " + xattr.getName());
+          final HdfsProtos.ZoneEncryptionInfoProto ezProto =
-              }
+              HdfsProtos.ZoneEncryptionInfoProto.parseFrom(
-            }
+                  xattr.getValue());
-          }
+          ezManager.unprotectedAddEncryptionZone(inode.getId(),
+              PBHelper.convert(ezProto.getSuite()),
+              PBHelper.convert(ezProto.getCryptoProtocolVersion()),
+              ezProto.getKeyName());
+        } catch (InvalidProtocolBufferException e) {
+          NameNode.LOG.warn("Error parsing protocol buffer of " +
+              "EZ XAttr " + xattr.getName() + " dir:" + inode.getFullPathName());
         }
       }
     }
   }
   
+  /**
+   * This is to handle encryption zone for rootDir when loading from
+   * fsimage, and should only be called during NN restart.
+   */
+  public final void addRootDirToEncryptionZone(XAttrFeature xaf) {
+    addEncryptionZone(rootDir, xaf);
+  }
+
   /**
    * This method is always called with writeLock of FSDirectory held.
    */

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

@@ -389,6 +389,7 @@ public final class FSImageFormatPBINode {
       if (f != null) {
         dir.rootDir.addXAttrFeature(f);
       }
+      dir.addRootDirToEncryptionZone(f);
     }
   }
 

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestEncryptionZones.java

@@ -374,6 +374,44 @@ public class TestEncryptionZones {
     assertZonePresent(null, nonpersistZone.toString());
   }
 
+  @Test(timeout = 60000)
+  public void testBasicOperationsRootDir() throws Exception {
+    int numZones = 0;
+    final Path rootDir = new Path("/");
+    final Path zone1 = new Path(rootDir, "zone1");
+
+    /* Normal creation of an EZ on rootDir */
+    dfsAdmin.createEncryptionZone(rootDir, TEST_KEY);
+    assertNumZones(++numZones);
+    assertZonePresent(null, rootDir.toString());
+
+    /* create EZ on child of rootDir which is already an EZ should fail */
+    fsWrapper.mkdir(zone1, FsPermission.getDirDefault(), true);
+    try {
+      dfsAdmin.createEncryptionZone(zone1, TEST_KEY);
+      fail("EZ over an EZ");
+    } catch (IOException e) {
+      assertExceptionContains("already in an encryption zone", e);
+    }
+
+    // Verify rootDir ez is present after restarting the NameNode
+    // and saving/loading from fsimage.
+    fs.setSafeMode(SafeModeAction.SAFEMODE_ENTER);
+    fs.saveNamespace();
+    fs.setSafeMode(SafeModeAction.SAFEMODE_LEAVE);
+    cluster.restartNameNode(true);
+    assertNumZones(numZones);
+    assertZonePresent(null, rootDir.toString());
+
+    /* create EZ on child of rootDir which is already an EZ should fail */
+    try {
+      dfsAdmin.createEncryptionZone(zone1, TEST_KEY);
+      fail("EZ over an EZ");
+    } catch (IOException e) {
+      assertExceptionContains("already in an encryption zone", e);
+    }
+  }
+
   /**
    * Test listing encryption zones as a non super user.
    */