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

This reverts commit b19531f04b.

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

@@ -644,7 +644,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(AccessControlUtil.toUserTablePermissions(perms).toByteArray());
   }
 

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

@@ -246,7 +246,7 @@ public class AccessController implements MasterCoprocessor, RegionCoprocessor,
       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;
@@ -283,7 +283,7 @@ public class AccessController implements MasterCoprocessor, RegionCoprocessor,
         currentEntry = entry;
         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

@@ -656,6 +656,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

@@ -293,7 +293,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/TestZKPermissionWatcher.java

@@ -21,7 +21,6 @@ import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 
 import java.util.ArrayList;
-import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicBoolean;
 import org.apache.hadoop.conf.Configuration;
@@ -34,8 +33,6 @@ import org.apache.hadoop.hbase.security.User;
 import org.apache.hadoop.hbase.testclassification.LargeTests;
 import org.apache.hadoop.hbase.testclassification.SecurityTests;
 import org.apache.hadoop.hbase.zookeeper.ZKWatcher;
-import org.apache.hbase.thirdparty.com.google.common.collect.ArrayListMultimap;
-import org.apache.hbase.thirdparty.com.google.common.collect.ListMultimap;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
 import org.junit.ClassRule;
@@ -95,26 +92,6 @@ public class TestZKPermissionWatcher {
     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();
@@ -139,9 +116,20 @@ public class TestZKPermissionWatcher {
     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<>(1);
-        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,
@@ -161,8 +149,19 @@ public class TestZKPermissionWatcher {
     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<>(1);
+    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,