Revert "HBASE-19970 Remove unused functions from TableAuthManager."

This reverts commit e6ce789b6f.

hbase-server/src/main/java/org/apache/hadoop/hbase/security/access/AccessControlLists.java

@@ -73,10 +73,6 @@ import org.apache.hadoop.io.Text;
 import com.google.common.collect.ArrayListMultimap;
 import com.google.common.collect.ListMultimap;
 import com.google.common.collect.Lists;
-import org.apache.hadoop.io.Writable;
-import org.apache.hadoop.io.WritableFactories;
-import org.apache.hadoop.io.WritableUtils;
-import org.apache.jasper.tagplugins.jstl.core.Remove;
 
 /**
  * Maintains lists of permission grants to users and groups to allow for
@@ -671,7 +667,8 @@ public class AccessControlLists {
    *
    * Writes a set of permission [user: table permission]
    */
-  public static byte[] writePermissionsAsBytes(ListMultimap<String, TablePermission> perms) {
+  public static byte[] writePermissionsAsBytes(ListMultimap<String, TablePermission> perms,
+      Configuration conf) {
     return ProtobufUtil.prependPBMagic(ProtobufUtil.toUserTablePermissions(perms).toByteArray());
   }
 

hbase-server/src/main/java/org/apache/hadoop/hbase/security/access/AccessController.java

@@ -243,7 +243,7 @@ public class AccessController extends BaseMasterAndRegionObserver
       tables.entrySet()) {
       byte[] entry = t.getKey();
       ListMultimap<String,TablePermission> perms = t.getValue();
-      byte[] serialized = AccessControlLists.writePermissionsAsBytes(perms);
+      byte[] serialized = AccessControlLists.writePermissionsAsBytes(perms, conf);
       getAuthManager().getZKPermissionWatcher().writeToZookeeper(entry, serialized);
     }
     initialized = true;
@@ -275,7 +275,7 @@ public class AccessController extends BaseMasterAndRegionObserver
         try (Table t = regionEnv.getTable(AccessControlLists.ACL_TABLE_NAME)) {
           ListMultimap<String,TablePermission> perms =
               AccessControlLists.getPermissions(conf, entry, t);
-          byte[] serialized = AccessControlLists.writePermissionsAsBytes(perms);
+          byte[] serialized = AccessControlLists.writePermissionsAsBytes(perms, conf);
           zkw.writeToZookeeper(entry, serialized);
         }
       } catch (IOException ex) {

hbase-server/src/main/java/org/apache/hadoop/hbase/security/access/TableAuthManager.java

@@ -659,6 +659,81 @@ public class TableAuthManager implements Closeable {
     tableCache.remove(table);
   }
 
+  /**
+   * Overwrites the existing permission set for a given user for a table, and
+   * triggers an update for zookeeper synchronization.
+   * @param username
+   * @param table
+   * @param perms
+   */
+  public void setTableUserPermissions(String username, TableName table,
+      List<TablePermission> perms) {
+    PermissionCache<TablePermission> tablePerms = getTablePermissions(table);
+    tablePerms.replaceUser(username, perms);
+    writeTableToZooKeeper(table, tablePerms);
+  }
+
+  /**
+   * Overwrites the existing permission set for a group and triggers an update
+   * for zookeeper synchronization.
+   * @param group
+   * @param table
+   * @param perms
+   */
+  public void setTableGroupPermissions(String group, TableName table,
+      List<TablePermission> perms) {
+    PermissionCache<TablePermission> tablePerms = getTablePermissions(table);
+    tablePerms.replaceGroup(group, perms);
+    writeTableToZooKeeper(table, tablePerms);
+  }
+
+  /**
+   * Overwrites the existing permission set for a given user for a table, and
+   * triggers an update for zookeeper synchronization.
+   * @param username
+   * @param namespace
+   * @param perms
+   */
+  public void setNamespaceUserPermissions(String username, String namespace,
+      List<TablePermission> perms) {
+    PermissionCache<TablePermission> tablePerms = getNamespacePermissions(namespace);
+    tablePerms.replaceUser(username, perms);
+    writeNamespaceToZooKeeper(namespace, tablePerms);
+  }
+
+  /**
+   * Overwrites the existing permission set for a group and triggers an update
+   * for zookeeper synchronization.
+   * @param group
+   * @param namespace
+   * @param perms
+   */
+  public void setNamespaceGroupPermissions(String group, String namespace,
+      List<TablePermission> perms) {
+    PermissionCache<TablePermission> tablePerms = getNamespacePermissions(namespace);
+    tablePerms.replaceGroup(group, perms);
+    writeNamespaceToZooKeeper(namespace, tablePerms);
+  }
+
+  public void writeTableToZooKeeper(TableName table,
+      PermissionCache<TablePermission> tablePerms) {
+    byte[] serialized = new byte[0];
+    if (tablePerms != null) {
+      serialized = AccessControlLists.writePermissionsAsBytes(tablePerms.getAllPermissions(), conf);
+    }
+    zkperms.writeToZookeeper(table.getName(), serialized);
+  }
+
+  public void writeNamespaceToZooKeeper(String namespace,
+      PermissionCache<TablePermission> tablePerms) {
+    byte[] serialized = new byte[0];
+    if (tablePerms != null) {
+      serialized = AccessControlLists.writePermissionsAsBytes(tablePerms.getAllPermissions(), conf);
+    }
+    zkperms.writeToZookeeper(Bytes.toBytes(AccessControlLists.toNamespaceEntry(namespace)),
+        serialized);
+  }
+
   public long getMTime() {
     return mtime.get();
   }

hbase-server/src/test/java/org/apache/hadoop/hbase/security/access/TestTablePermissions.java

@@ -323,7 +323,7 @@ public class TestTablePermissions {
   public void testSerialization() throws Exception {
     Configuration conf = UTIL.getConfiguration();
     ListMultimap<String,TablePermission> permissions = createPermissions();
-    byte[] permsData = AccessControlLists.writePermissionsAsBytes(permissions);
+    byte[] permsData = AccessControlLists.writePermissionsAsBytes(permissions, conf);
 
     ListMultimap<String, TablePermission> copy =
         AccessControlLists.readPermissions(permsData, conf);

hbase-server/src/test/java/org/apache/hadoop/hbase/security/access/TestZKPermissionsWatcher.java

@@ -21,10 +21,7 @@ package org.apache.hadoop.hbase.security.access;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 
-import com.google.common.collect.ArrayListMultimap;
-import com.google.common.collect.ListMultimap;
 import java.util.ArrayList;
-import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicBoolean;
 
@@ -89,26 +86,6 @@ public class TestZKPermissionsWatcher {
     UTIL.shutdownMiniCluster();
   }
 
-  private void setTableACL(
-      User user, TableAuthManager srcAuthManager, TableAuthManager destAuthManager,
-      TablePermission.Action... actions) throws Exception{
-    // update ACL: george RW
-    ListMultimap<String, TablePermission> perms = ArrayListMultimap.create();
-    perms.replaceValues(user.getShortName(),
-        Collections.singletonList(new TablePermission(TEST_TABLE, null, actions)));
-    byte[] serialized = AccessControlLists.writePermissionsAsBytes(perms);
-    final long mtime = destAuthManager.getMTime();
-    srcAuthManager.getZKPermissionWatcher().writeToZookeeper(TEST_TABLE.getName(), serialized);
-    // Wait for the update to propagate
-    UTIL.waitFor(10000, 100, new Predicate<Exception>() {
-      @Override
-      public boolean evaluate() throws Exception {
-        return destAuthManager.getMTime() > mtime;
-      }
-    });
-    Thread.sleep(1000);
-  }
-
   @Test
   public void testPermissionsWatcher() throws Exception {
     Configuration conf = UTIL.getConfiguration();
@@ -133,9 +110,20 @@ public class TestZKPermissionsWatcher {
     assertFalse(AUTH_B.authorizeUser(hubert, TEST_TABLE, null,
       TablePermission.Action.WRITE));
 
-    // update ACL: george, RW
+    // update ACL: george RW
-    setTableACL(george, AUTH_A, AUTH_B,
+    List<TablePermission> acl = new ArrayList<TablePermission>();
-        TablePermission.Action.READ, TablePermission.Action.WRITE);
+    acl.add(new TablePermission(TEST_TABLE, null, TablePermission.Action.READ,
+      TablePermission.Action.WRITE));
+    final long mtimeB = AUTH_B.getMTime();
+    AUTH_A.setTableUserPermissions(george.getShortName(), TEST_TABLE, acl);
+    // Wait for the update to propagate
+    UTIL.waitFor(10000, 100, new Predicate<Exception>() {
+      @Override
+      public boolean evaluate() throws Exception {
+        return AUTH_B.getMTime() > mtimeB;
+      }
+    });
+    Thread.sleep(1000);
 
     // check it
     assertTrue(AUTH_A.authorizeUser(george, TEST_TABLE, null,
@@ -155,8 +143,19 @@ public class TestZKPermissionsWatcher {
     assertFalse(AUTH_B.authorizeUser(hubert, TEST_TABLE, null,
       TablePermission.Action.WRITE));
 
-    // update ACL: hubert, Read
+    // update ACL: hubert R
-    setTableACL(hubert, AUTH_B, AUTH_A, TablePermission.Action.READ);
+    acl = new ArrayList<TablePermission>();
+    acl.add(new TablePermission(TEST_TABLE, null, TablePermission.Action.READ));
+    final long mtimeA = AUTH_A.getMTime();
+    AUTH_B.setTableUserPermissions("hubert", TEST_TABLE, acl);
+    // Wait for the update to propagate
+    UTIL.waitFor(10000, 100, new Predicate<Exception>() {
+      @Override
+      public boolean evaluate() throws Exception {
+        return AUTH_A.getMTime() > mtimeA;
+      }
+    });
+    Thread.sleep(1000);
 
     // check it
     assertTrue(AUTH_A.authorizeUser(george, TEST_TABLE, null,