HDFS-12614. FSPermissionChecker#getINodeAttrs() throws NPE when INodeAttributesProvider configured. Contributed by Manoj Govindassamy.

(cherry picked from commit 0aa52d4085f8c77dbfd9d913011c347882573ad9)

Conflicts:
	hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestINodeAttributeProvider.java

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

@@ -239,8 +239,16 @@ class FSPermissionChecker implements AccessControlEnforcer {
     INodeAttributes inodeAttrs = inode.getSnapshotINode(snapshotId);
     if (getAttributesProvider() != null) {
       String[] elements = new String[pathIdx + 1];
-      for (int i = 0; i < elements.length; i++) {
-        elements[i] = DFSUtil.bytes2String(pathByNameArr[i]);
+      /**
+       * {@link INode#getPathComponents(String)} returns a null component
+       * for the root only path "/". Assign an empty string if so.
+       */
+      if (pathByNameArr.length == 1 && pathByNameArr[0] == null) {
+        elements[0] = "";
+      } else {
+        for (int i = 0; i < elements.length; i++) {
+          elements[i] = DFSUtil.bytes2String(pathByNameArr[i]);
+        }
       }
       inodeAttrs = getAttributesProvider().getAttributes(elements, inodeAttrs);
     }

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

@@ -222,31 +222,59 @@ public class TestINodeAttributeProvider {
     });
   }
 
-  @Test
-  public void testCustomProvider() throws Exception {
+  private void verifyFileStatus(UserGroupInformation ugi) throws IOException {
     FileSystem fs = FileSystem.get(miniDFS.getConfiguration(0));
-    fs.mkdirs(new Path("/user/xxx"));
-    FileStatus status = fs.getFileStatus(new Path("/user/xxx"));
-    Assert.assertEquals(System.getProperty("user.name"), status.getOwner());
+
+    FileStatus status = fs.getFileStatus(new Path("/"));
+    Path userDir = new Path("/user/" + ugi.getShortUserName());
+    fs.mkdirs(userDir);
+    status = fs.getFileStatus(userDir);
+    Assert.assertEquals(ugi.getShortUserName(), status.getOwner());
     Assert.assertEquals("supergroup", status.getGroup());
     Assert.assertEquals(new FsPermission((short) 0755), status.getPermission());
-    fs.mkdirs(new Path("/user/authz"));
-    Path p = new Path("/user/authz");
-    status = fs.getFileStatus(p);
+
+    Path authzDir = new Path("/user/authz");
+    fs.mkdirs(authzDir);
+    status = fs.getFileStatus(authzDir);
     Assert.assertEquals("foo", status.getOwner());
     Assert.assertEquals("bar", status.getGroup());
     Assert.assertEquals(new FsPermission((short) 0770), status.getPermission());
-    AclStatus aclStatus = fs.getAclStatus(p);
+
+    AclStatus aclStatus = fs.getAclStatus(authzDir);
     Assert.assertEquals(1, aclStatus.getEntries().size());
-    Assert.assertEquals(AclEntryType.GROUP, aclStatus.getEntries().get(0)
-            .getType());
-    Assert.assertEquals("xxx", aclStatus.getEntries().get(0)
-            .getName());
-    Assert.assertEquals(FsAction.ALL, aclStatus.getEntries().get(0)
-            .getPermission());
-    Map<String, byte[]> xAttrs = fs.getXAttrs(p);
+    Assert.assertEquals(AclEntryType.GROUP,
+        aclStatus.getEntries().get(0).getType());
+    Assert.assertEquals("xxx",
+        aclStatus.getEntries().get(0).getName());
+    Assert.assertEquals(FsAction.ALL,
+        aclStatus.getEntries().get(0).getPermission());
+    Map<String, byte[]> xAttrs = fs.getXAttrs(authzDir);
     Assert.assertTrue(xAttrs.containsKey("user.test"));
     Assert.assertEquals(2, xAttrs.get("user.test").length);
   }
 
+  /**
+   * With the custom provider configured, verify file status attributes.
+   * A superuser can bypass permission check while resolving paths. So,
+   * verify file status for both superuser and non-superuser.
+   */
+  @Test
+  public void testCustomProvider() throws Exception {
+    final UserGroupInformation[] users = new UserGroupInformation[]{
+        UserGroupInformation.createUserForTesting(
+            System.getProperty("user.name"), new String[]{"supergroup"}),
+        UserGroupInformation.createUserForTesting(
+            "normaluser", new String[]{"normalusergroup"}),
+    };
+
+    for (final UserGroupInformation user : users) {
+      user.doAs(new PrivilegedExceptionAction<Void>() {
+        @Override
+        public Void run() throws Exception {
+          verifyFileStatus(user);
+          return null;
+        }
+      });
+    }
+  }
 }