HDFS-12614. FSPermissionChecker#getINodeAttrs() throws NPE when INodeAttributesProvider configured.

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

@@ -275,8 +275,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

@@ -313,31 +313,59 @@ public class TestINodeAttributeProvider {
     testBypassProviderHelper(users, HDFS_PERMISSION, true);
   }
 
-  @Test
+  private void verifyFileStatus(UserGroupInformation ugi) throws IOException {
-  public void testCustomProvider() throws Exception {
     FileSystem fs = FileSystem.get(miniDFS.getConfiguration(0));
-    fs.mkdirs(new Path("/user/xxx"));
+
-    FileStatus status = fs.getFileStatus(new Path("/user/xxx"));
+    FileStatus status = fs.getFileStatus(new Path("/"));
-    Assert.assertEquals(System.getProperty("user.name"), status.getOwner());
+    LOG.info("Path '/' is owned by: "
+        + status.getOwner() + ":" + status.getGroup());
+
+    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");
+    Path authzDir = new Path("/user/authz");
-    status = fs.getFileStatus(p);
+    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)
+    Assert.assertEquals(AclEntryType.GROUP,
-            .getType());
+        aclStatus.getEntries().get(0).getType());
-    Assert.assertEquals("xxx", aclStatus.getEntries().get(0)
+    Assert.assertEquals("xxx",
-            .getName());
+        aclStatus.getEntries().get(0).getName());
-    Assert.assertEquals(FsAction.ALL, aclStatus.getEntries().get(0)
+    Assert.assertEquals(FsAction.ALL,
-            .getPermission());
+        aclStatus.getEntries().get(0).getPermission());
-    Map<String, byte[]> xAttrs = fs.getXAttrs(p);
+    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((PrivilegedExceptionAction<Object>) () -> {
+        verifyFileStatus(user);
+        return null;
+      });
+    }
+  }
 }