HBASE-18323 Remove multiple ACLs for the same user in kerberos

Signed-off-by: Josh Elser <elserj@apache.org>

hbase-client/src/main/java/org/apache/hadoop/hbase/zookeeper/ZKUtil.java

@@ -58,6 +58,7 @@ import org.apache.hadoop.hbase.zookeeper.ZKUtil.ZKUtilOp.CreateAndFailSilent;
 import org.apache.hadoop.hbase.zookeeper.ZKUtil.ZKUtilOp.DeleteNodeFailSilent;
 import org.apache.hadoop.hbase.zookeeper.ZKUtil.ZKUtilOp.SetData;
 import org.apache.hadoop.security.SecurityUtil;
+import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.security.authentication.util.KerberosUtil;
 import org.apache.zookeeper.AsyncCallback;
 import org.apache.zookeeper.CreateMode;
@@ -907,6 +908,12 @@ public class ZKUtil {
       ArrayList<ACL> acls = new ArrayList<>();
       // add permission to hbase supper user
       String[] superUsers = zkw.getConfiguration().getStrings(Superusers.SUPERUSER_CONF_KEY);
+      String hbaseUser = null;
+      try {
+        hbaseUser = UserGroupInformation.getCurrentUser().getShortUserName();
+      } catch (IOException e) {
+        LOG.debug("Could not acquire current User.", e);
+      }
       if (superUsers != null) {
         List<String> groups = new ArrayList<>();
         for (String user : superUsers) {
@@ -914,9 +921,11 @@ public class ZKUtil {
             // TODO: Set node ACL for groups when ZK supports this feature
             groups.add(user);
           } else {
+            if(!user.equals(hbaseUser)) {
               acls.add(new ACL(Perms.ALL, new Id("sasl", user)));
             }
           }
+        }
         if (!groups.isEmpty()) {
           LOG.warn("Znode ACL setting for group " + groups
               + " is skipped, ZooKeeper doesn't support this feature presently.");

hbase-client/src/test/java/org/apache/hadoop/hbase/zookeeper/TestZKUtil.java

@@ -26,6 +26,7 @@ import org.apache.hadoop.hbase.HBaseConfiguration;
 import org.apache.hadoop.hbase.ZooKeeperConnectionException;
 import org.apache.hadoop.hbase.security.Superusers;
 import org.apache.hadoop.hbase.testclassification.SmallTests;
+import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.zookeeper.ZooDefs.Ids;
 import org.apache.zookeeper.ZooDefs.Perms;
 import org.apache.zookeeper.data.ACL;
@@ -77,4 +78,19 @@ public class TestZKUtil {
     Assert.assertTrue(aclList.contains(new ACL(Perms.ALL, new Id("sasl", "user2"))));
     Assert.assertTrue(aclList.contains(new ACL(Perms.ALL, new Id("sasl", "user3"))));
   }
+
+  @Test
+  public void testCreateACLWithSameUser() throws ZooKeeperConnectionException, IOException {
+    Configuration conf = HBaseConfiguration.create();
+    conf.set(Superusers.SUPERUSER_CONF_KEY, "user4,@group1,user5,user6");
+    UserGroupInformation.setLoginUser(UserGroupInformation.createRemoteUser("user4"));
+    String node = "/hbase/testCreateACL";
+    ZooKeeperWatcher watcher = new ZooKeeperWatcher(conf, node, null, false);
+    List<ACL> aclList = ZKUtil.createACL(watcher, node, true);
+    Assert.assertEquals(aclList.size(), 3); // 3, since service user the same as one of superuser
+    Assert.assertFalse(aclList.contains(new ACL(Perms.ALL, new Id("sasl", "@group1"))));
+    Assert.assertTrue(aclList.contains(new ACL(Perms.ALL, new Id("auth", ""))));
+    Assert.assertTrue(aclList.contains(new ACL(Perms.ALL, new Id("sasl", "user5"))));
+    Assert.assertTrue(aclList.contains(new ACL(Perms.ALL, new Id("sasl", "user6"))));
+  }
 }