HBASE-16773 AccessController should access local region if possible

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

@@ -22,10 +22,8 @@ import java.io.ByteArrayInputStream;
 import java.io.DataInput;
 import java.io.DataInputStream;
 import java.io.IOException;
-import java.io.InputStream;
 import java.util.ArrayList;
 import java.util.Arrays;
-import java.util.Collection;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
@@ -41,6 +39,7 @@ import org.apache.hadoop.hbase.Cell;
 import org.apache.hadoop.hbase.CellUtil;
 import org.apache.hadoop.hbase.HColumnDescriptor;
 import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.HRegionInfo;
 import org.apache.hadoop.hbase.HTableDescriptor;
 import org.apache.hadoop.hbase.NamespaceDescriptor;
 import org.apache.hadoop.hbase.TableName;
@@ -64,8 +63,6 @@ import org.apache.hadoop.hbase.filter.RegexStringComparator;
 import org.apache.hadoop.hbase.master.MasterServices;
 import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
 import org.apache.hadoop.hbase.protobuf.generated.AccessControlProtos;
-import org.apache.hadoop.hbase.protobuf.generated.AccessControlProtos.AccessControlService;
-import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos;
 import org.apache.hadoop.hbase.regionserver.BloomType;
 import org.apache.hadoop.hbase.regionserver.InternalScanner;
 import org.apache.hadoop.hbase.regionserver.Region;
@@ -77,11 +74,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 com.google.protobuf.ByteString;
-import com.google.protobuf.CodedInputStream;
-import com.google.protobuf.Message;
-import com.google.protobuf.RpcController;
-import com.google.protobuf.ServiceException;
 
 /**
  * Maintains lists of permission grants to users and groups to allow for
@@ -153,9 +145,10 @@ public class AccessControlLists {
    * Stores a new user permission grant in the access control lists table.
    * @param conf the configuration
    * @param userPerm the details of the permission to be granted
+   * @param t acl table instance. It is closed upon method return.
    * @throws IOException in the case of an error accessing the metadata table
    */
-  static void addUserPermission(Configuration conf, UserPermission userPerm)
+  static void addUserPermission(Configuration conf, UserPermission userPerm, Table t)
       throws IOException {
     Permission.Action[] actions = userPerm.getActions();
     byte[] rowKey = userPermissionRowKey(userPerm);
@@ -179,11 +172,10 @@ public class AccessControlLists {
           Bytes.toString(key)+": "+Bytes.toStringBinary(value)
           );
     }
-    // TODO: Pass in a Connection rather than create one each time.
+    try {
-    try (Connection connection = ConnectionFactory.createConnection(conf)) {
+      t.put(p);
-      try (Table table = connection.getTable(ACL_TABLE_NAME)) {
+    } finally {
-        table.put(p);
+      t.close();
-      }
     }
   }
 
@@ -198,9 +190,10 @@ public class AccessControlLists {
    *
    * @param conf the configuration
    * @param userPerm the details of the permission to be revoked
+   * @param t acl table
    * @throws IOException if there is an error accessing the metadata table
    */
-  static void removeUserPermission(Configuration conf, UserPermission userPerm)
+  static void removeUserPermission(Configuration conf, UserPermission userPerm, Table t)
       throws IOException {
     Delete d = new Delete(userPermissionRowKey(userPerm));
     byte[] key = userPermissionKey(userPerm);
@@ -209,36 +202,34 @@ public class AccessControlLists {
       LOG.debug("Removing permission "+ userPerm.toString());
     }
     d.addColumns(ACL_LIST_FAMILY, key);
-    // TODO: Pass in a Connection rather than create one each time.
+    try {
-    try (Connection connection = ConnectionFactory.createConnection(conf)) {
+      t.delete(d);
-      try (Table table = connection.getTable(ACL_TABLE_NAME)) {
+    } finally {
-        table.delete(d);
+      t.close();
-      }
     }
   }
 
   /**
    * Remove specified table from the _acl_ table.
    */
-  static void removeTablePermissions(Configuration conf, TableName tableName)
+  static void removeTablePermissions(Configuration conf, TableName tableName, Table t)
       throws IOException{
     Delete d = new Delete(tableName.getName());
 
     if (LOG.isDebugEnabled()) {
       LOG.debug("Removing permissions of removed table "+ tableName);
     }
-    // TODO: Pass in a Connection rather than create one each time.
+    try {
-    try (Connection connection = ConnectionFactory.createConnection(conf)) {
+      t.delete(d);
-      try (Table table = connection.getTable(ACL_TABLE_NAME)) {
+    } finally {
-        table.delete(d);
+      t.close();
-      }
     }
   }
 
   /**
    * Remove specified namespace from the acl table.
    */
-  static void removeNamespacePermissions(Configuration conf, String namespace)
+  static void removeNamespacePermissions(Configuration conf, String namespace, Table t)
       throws IOException{
     Delete d = new Delete(Bytes.toBytes(toNamespaceEntry(namespace)));
 
@@ -246,56 +237,57 @@ public class AccessControlLists {
       LOG.debug("Removing permissions of removed namespace "+ namespace);
     }
 
-    try (Connection connection = ConnectionFactory.createConnection(conf)) {
+    try {
-      try (Table table = connection.getTable(ACL_TABLE_NAME)) {
+      t.delete(d);
+    } finally {
+      t.close();
+    }
+  }
+
+  static private void removeTablePermissions(TableName tableName, byte[] column, Table table,
+      boolean closeTable) throws IOException {
+    Scan scan = new Scan();
+    scan.addFamily(ACL_LIST_FAMILY);
+
+    String columnName = Bytes.toString(column);
+    scan.setFilter(new QualifierFilter(CompareOp.EQUAL, new RegexStringComparator(
+        String.format("(%s%s%s)|(%s%s)$",
+            ACL_KEY_DELIMITER, columnName, ACL_KEY_DELIMITER,
+            ACL_KEY_DELIMITER, columnName))));
+
+    Set<byte[]> qualifierSet = new TreeSet<byte[]>(Bytes.BYTES_COMPARATOR);
+    ResultScanner scanner = null;
+    try {
+      scanner = table.getScanner(scan);
+      for (Result res : scanner) {
+        for (byte[] q : res.getFamilyMap(ACL_LIST_FAMILY).navigableKeySet()) {
+          qualifierSet.add(q);
+        }
+      }
+
+      if (qualifierSet.size() > 0) {
+        Delete d = new Delete(tableName.getName());
+        for (byte[] qualifier : qualifierSet) {
+          d.addColumns(ACL_LIST_FAMILY, qualifier);
+        }
         table.delete(d);
       }
+    } finally {
+      if (scanner != null) scanner.close();
+      if (closeTable) table.close();
     }
   }
 
   /**
    * Remove specified table column from the acl table.
    */
-  static void removeTablePermissions(Configuration conf, TableName tableName, byte[] column)
+  static void removeTablePermissions(Configuration conf, TableName tableName, byte[] column,
-      throws IOException{
+      Table t) throws IOException {
-
     if (LOG.isDebugEnabled()) {
       LOG.debug("Removing permissions of removed column " + Bytes.toString(column) +
           " from table "+ tableName);
     }
-    // TODO: Pass in a Connection rather than create one each time.
+    removeTablePermissions(tableName, column, t, true);
-    try (Connection connection = ConnectionFactory.createConnection(conf)) {
-      try (Table table = connection.getTable(ACL_TABLE_NAME)) {
-        Scan scan = new Scan();
-        scan.addFamily(ACL_LIST_FAMILY);
-
-        String columnName = Bytes.toString(column);
-        scan.setFilter(new QualifierFilter(CompareOp.EQUAL, new RegexStringComparator(
-            String.format("(%s%s%s)|(%s%s)$",
-                ACL_KEY_DELIMITER, columnName, ACL_KEY_DELIMITER,
-                ACL_KEY_DELIMITER, columnName))));
-
-        Set<byte[]> qualifierSet = new TreeSet<byte[]>(Bytes.BYTES_COMPARATOR);
-        ResultScanner scanner = table.getScanner(scan);
-        try {
-          for (Result res : scanner) {
-            for (byte[] q : res.getFamilyMap(ACL_LIST_FAMILY).navigableKeySet()) {
-              qualifierSet.add(q);
-            }
-          }
-        } finally {
-          scanner.close();
-        }
-
-        if (qualifierSet.size() > 0) {
-          Delete d = new Delete(tableName.getName());
-          for (byte[] qualifier : qualifierSet) {
-            d.addColumns(ACL_LIST_FAMILY, qualifier);
-          }
-          table.delete(d);
-        }
-      }
-    }
   }
 
   static byte[] userPermissionRowKey(UserPermission userPerm) {
@@ -442,12 +434,12 @@ public class AccessControlLists {
 
   static ListMultimap<String, TablePermission> getTablePermissions(Configuration conf,
       TableName tableName) throws IOException {
-    return getPermissions(conf, tableName != null ? tableName.getName() : null);
+    return getPermissions(conf, tableName != null ? tableName.getName() : null, null);
   }
 
   static ListMultimap<String, TablePermission> getNamespacePermissions(Configuration conf,
       String namespace) throws IOException {
-    return getPermissions(conf, Bytes.toBytes(toNamespaceEntry(namespace)));
+    return getPermissions(conf, Bytes.toBytes(toNamespaceEntry(namespace)), null);
   }
 
   /**
@@ -460,24 +452,28 @@ public class AccessControlLists {
    * </p>
    */
   static ListMultimap<String, TablePermission> getPermissions(Configuration conf,
-      byte[] entryName) throws IOException {
+      byte[] entryName, Table t) throws IOException {
     if (entryName == null) entryName = ACL_GLOBAL_NAME;
 
     // for normal user tables, we just read the table row from _acl_
     ListMultimap<String, TablePermission> perms = ArrayListMultimap.create();
-    // TODO: Pass in a Connection rather than create one each time.
+    Get get = new Get(entryName);
-    try (Connection connection = ConnectionFactory.createConnection(conf)) {
+    get.addFamily(ACL_LIST_FAMILY);
-      try (Table table = connection.getTable(ACL_TABLE_NAME)) {
+    Result row = null;
-        Get get = new Get(entryName);
+    if (t == null) {
-        get.addFamily(ACL_LIST_FAMILY);
+      try (Connection connection = ConnectionFactory.createConnection(conf)) {
-        Result row = table.get(get);
+        try (Table table = connection.getTable(ACL_TABLE_NAME)) {
-        if (!row.isEmpty()) {
+          row = table.get(get);
-          perms = parsePermissions(entryName, row);
-        } else {
-          LOG.info("No permissions found in " + ACL_TABLE_NAME + " for acl entry "
-              + Bytes.toString(entryName));
         }
       }
+    } else {
+      row = t.get(get);
+    }
+    if (!row.isEmpty()) {
+      perms = parsePermissions(entryName, row);
+    } else {
+      LOG.info("No permissions found in " + ACL_TABLE_NAME + " for acl entry "
+          + Bytes.toString(entryName));
     }
 
     return perms;
@@ -501,7 +497,7 @@ public class AccessControlLists {
       Configuration conf, byte[] entryName)
           throws IOException {
     ListMultimap<String,TablePermission> allPerms = getPermissions(
-        conf, entryName);
+        conf, entryName, null);
 
     List<UserPermission> perms = new ArrayList<UserPermission>();
 

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

@@ -67,6 +67,7 @@ import org.apache.hadoop.hbase.client.Put;
 import org.apache.hadoop.hbase.client.Query;
 import org.apache.hadoop.hbase.client.Result;
 import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.client.Table;
 import org.apache.hadoop.hbase.coprocessor.BaseMasterAndRegionObserver;
 import org.apache.hadoop.hbase.coprocessor.BulkLoadObserver;
 import org.apache.hadoop.hbase.coprocessor.CoprocessorException;
@@ -271,10 +272,12 @@ public class AccessController extends BaseMasterAndRegionObserver
     Configuration conf = regionEnv.getConfiguration();
     for (byte[] entry: entries) {
       try {
-        ListMultimap<String,TablePermission> perms =
+        try (Table t = regionEnv.getTable(AccessControlLists.ACL_TABLE_NAME)) {
-          AccessControlLists.getPermissions(conf, entry);
+          ListMultimap<String,TablePermission> perms =
-        byte[] serialized = AccessControlLists.writePermissionsAsBytes(perms, conf);
+              AccessControlLists.getPermissions(conf, entry, t);
-        zkw.writeToZookeeper(entry, serialized);
+          byte[] serialized = AccessControlLists.writePermissionsAsBytes(perms, conf);
+          zkw.writeToZookeeper(entry, serialized);
+        }
       } catch (IOException ex) {
         LOG.error("Failed updating permissions mirror for '" + Bytes.toString(entry) + "'",
             ex);
@@ -1033,7 +1036,7 @@ public class AccessController extends BaseMasterAndRegionObserver
           @Override
           public Void run() throws Exception {
             AccessControlLists.addUserPermission(c.getEnvironment().getConfiguration(),
-                userperm);
+                userperm, c.getEnvironment().getTable(AccessControlLists.ACL_TABLE_NAME));
             return null;
           }
         });
@@ -1055,7 +1058,8 @@ public class AccessController extends BaseMasterAndRegionObserver
     User.runAsLoginUser(new PrivilegedExceptionAction<Void>() {
       @Override
       public Void run() throws Exception {
-        AccessControlLists.removeTablePermissions(conf, tableName);
+        AccessControlLists.removeTablePermissions(conf, tableName,
+            c.getEnvironment().getTable(AccessControlLists.ACL_TABLE_NAME));
         return null;
       }
     });
@@ -1091,7 +1095,8 @@ public class AccessController extends BaseMasterAndRegionObserver
         List<UserPermission> perms = tableAcls.get(tableName);
         if (perms != null) {
           for (UserPermission perm : perms) {
-            AccessControlLists.addUserPermission(conf, perm);
+            AccessControlLists.addUserPermission(conf, perm,
+                ctx.getEnvironment().getTable(AccessControlLists.ACL_TABLE_NAME));
           }
         }
         tableAcls.remove(tableName);
@@ -1119,7 +1124,8 @@ public class AccessController extends BaseMasterAndRegionObserver
       public Void run() throws Exception {
         UserPermission userperm = new UserPermission(Bytes.toBytes(owner),
             htd.getTableName(), null, Action.values());
-        AccessControlLists.addUserPermission(conf, userperm);
+        AccessControlLists.addUserPermission(conf, userperm,
+            c.getEnvironment().getTable(AccessControlLists.ACL_TABLE_NAME));
         return null;
       }
     });
@@ -1155,7 +1161,8 @@ public class AccessController extends BaseMasterAndRegionObserver
     User.runAsLoginUser(new PrivilegedExceptionAction<Void>() {
       @Override
       public Void run() throws Exception {
-        AccessControlLists.removeTablePermissions(conf, tableName, columnFamily);
+        AccessControlLists.removeTablePermissions(conf, tableName, columnFamily,
+            ctx.getEnvironment().getTable(AccessControlLists.ACL_TABLE_NAME));
         return null;
       }
     });
@@ -1383,7 +1390,8 @@ public class AccessController extends BaseMasterAndRegionObserver
     User.runAsLoginUser(new PrivilegedExceptionAction<Void>() {
       @Override
       public Void run() throws Exception {
-        AccessControlLists.removeNamespacePermissions(conf, namespace);
+        AccessControlLists.removeNamespacePermissions(conf, namespace,
+            ctx.getEnvironment().getTable(AccessControlLists.ACL_TABLE_NAME));
         return null;
       }
     });
@@ -2232,7 +2240,8 @@ public class AccessController extends BaseMasterAndRegionObserver
         User.runAsLoginUser(new PrivilegedExceptionAction<Void>() {
           @Override
           public Void run() throws Exception {
-            AccessControlLists.addUserPermission(regionEnv.getConfiguration(), perm);
+            AccessControlLists.addUserPermission(regionEnv.getConfiguration(), perm,
+                regionEnv.getTable(AccessControlLists.ACL_TABLE_NAME));
             return null;
           }
         });
@@ -2284,7 +2293,8 @@ public class AccessController extends BaseMasterAndRegionObserver
         User.runAsLoginUser(new PrivilegedExceptionAction<Void>() {
           @Override
           public Void run() throws Exception {
-            AccessControlLists.removeUserPermission(regionEnv.getConfiguration(), perm);
+            AccessControlLists.removeUserPermission(regionEnv.getConfiguration(), perm,
+                regionEnv.getTable(AccessControlLists.ACL_TABLE_NAME));
             return null;
           }
         });

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

@@ -39,6 +39,8 @@ import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.Abortable;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.Admin;
+import org.apache.hadoop.hbase.client.Connection;
+import org.apache.hadoop.hbase.client.ConnectionFactory;
 import org.apache.hadoop.hbase.client.Table;
 import org.apache.hadoop.hbase.exceptions.DeserializationException;
 import org.apache.hadoop.hbase.HBaseTestingUtility;
@@ -114,9 +116,12 @@ public class TestTablePermissions {
   @After
   public void tearDown() throws Exception {
     Configuration conf = UTIL.getConfiguration();
-    AccessControlLists.removeTablePermissions(conf, TEST_TABLE);
+    try (Connection connection = ConnectionFactory.createConnection(conf);
-    AccessControlLists.removeTablePermissions(conf, TEST_TABLE2);
+        Table table = connection.getTable(AccessControlLists.ACL_TABLE_NAME)) {
-    AccessControlLists.removeTablePermissions(conf, AccessControlLists.ACL_TABLE_NAME);
+      AccessControlLists.removeTablePermissions(conf, TEST_TABLE, table);
+      AccessControlLists.removeTablePermissions(conf, TEST_TABLE2, table);
+      AccessControlLists.removeTablePermissions(conf, AccessControlLists.ACL_TABLE_NAME, table);
+    }
   }
 
   /**
@@ -171,18 +176,20 @@ public class TestTablePermissions {
   @Test
   public void testBasicWrite() throws Exception {
     Configuration conf = UTIL.getConfiguration();
-    // add some permissions
+    try (Connection connection = ConnectionFactory.createConnection(conf);
-    AccessControlLists.addUserPermission(conf,
+        Table table = connection.getTable(AccessControlLists.ACL_TABLE_NAME)) {
-            new UserPermission(Bytes.toBytes("george"), TEST_TABLE, null, (byte[])null,
+      // add some permissions
-            UserPermission.Action.READ, UserPermission.Action.WRITE));
+      AccessControlLists.addUserPermission(conf,
-    AccessControlLists.addUserPermission(conf,
+          new UserPermission(Bytes.toBytes("george"), TEST_TABLE, null, (byte[])null,
-        new UserPermission(Bytes.toBytes("hubert"), TEST_TABLE, null, (byte[])null,
+              UserPermission.Action.READ, UserPermission.Action.WRITE), table);
-            UserPermission.Action.READ));
+      AccessControlLists.addUserPermission(conf,
-    AccessControlLists.addUserPermission(conf,
+          new UserPermission(Bytes.toBytes("hubert"), TEST_TABLE, null, (byte[])null,
-        new UserPermission(Bytes.toBytes("humphrey"),
+              UserPermission.Action.READ), table);
-            TEST_TABLE, TEST_FAMILY, TEST_QUALIFIER,
+      AccessControlLists.addUserPermission(conf,
-            UserPermission.Action.READ));
+          new UserPermission(Bytes.toBytes("humphrey"),
-
+              TEST_TABLE, TEST_FAMILY, TEST_QUALIFIER,
+              UserPermission.Action.READ), table);
+    }
     // retrieve the same
     ListMultimap<String,TablePermission> perms =
         AccessControlLists.getTablePermissions(conf, TEST_TABLE);
@@ -235,10 +242,12 @@ public class TestTablePermissions {
     assertFalse(actions.contains(TablePermission.Action.WRITE));
 
     // table 2 permissions
-    AccessControlLists.addUserPermission(conf,
+    try (Connection connection = ConnectionFactory.createConnection(conf);
-        new UserPermission(Bytes.toBytes("hubert"), TEST_TABLE2, null, (byte[])null,
+        Table table = connection.getTable(AccessControlLists.ACL_TABLE_NAME)) {
-            TablePermission.Action.READ, TablePermission.Action.WRITE));
+      AccessControlLists.addUserPermission(conf,
-
+          new UserPermission(Bytes.toBytes("hubert"), TEST_TABLE2, null, (byte[])null,
+              TablePermission.Action.READ, TablePermission.Action.WRITE), table);
+    }
     // check full load
     Map<byte[], ListMultimap<String,TablePermission>> allPerms =
         AccessControlLists.loadAll(conf);
@@ -267,22 +276,24 @@ public class TestTablePermissions {
   @Test
   public void testPersistence() throws Exception {
     Configuration conf = UTIL.getConfiguration();
-    AccessControlLists.addUserPermission(conf,
+    try (Connection connection = ConnectionFactory.createConnection(conf);
-        new UserPermission(Bytes.toBytes("albert"), TEST_TABLE, null,
+        Table table = connection.getTable(AccessControlLists.ACL_TABLE_NAME)) {
-                           (byte[])null, TablePermission.Action.READ));
+      AccessControlLists.addUserPermission(conf,
-    AccessControlLists.addUserPermission(conf,
+          new UserPermission(Bytes.toBytes("albert"), TEST_TABLE, null,
-        new UserPermission(Bytes.toBytes("betty"), TEST_TABLE, null,
+              (byte[])null, TablePermission.Action.READ), table);
-                           (byte[])null, TablePermission.Action.READ,
+      AccessControlLists.addUserPermission(conf,
-                           TablePermission.Action.WRITE));
+          new UserPermission(Bytes.toBytes("betty"), TEST_TABLE, null,
-    AccessControlLists.addUserPermission(conf,
+              (byte[])null, TablePermission.Action.READ,
-        new UserPermission(Bytes.toBytes("clark"),
+              TablePermission.Action.WRITE), table);
-                           TEST_TABLE, TEST_FAMILY,
+      AccessControlLists.addUserPermission(conf,
-                           TablePermission.Action.READ));
+          new UserPermission(Bytes.toBytes("clark"),
-    AccessControlLists.addUserPermission(conf,
+              TEST_TABLE, TEST_FAMILY,
-        new UserPermission(Bytes.toBytes("dwight"),
+              TablePermission.Action.READ), table);
-                           TEST_TABLE, TEST_FAMILY, TEST_QUALIFIER,
+      AccessControlLists.addUserPermission(conf,
-                           TablePermission.Action.WRITE));
+          new UserPermission(Bytes.toBytes("dwight"),
-
+              TEST_TABLE, TEST_FAMILY, TEST_QUALIFIER,
+              TablePermission.Action.WRITE), table);
+    }
     // verify permissions survive changes in table metadata
     ListMultimap<String,TablePermission> preperms =
         AccessControlLists.getTablePermissions(conf, TEST_TABLE);
@@ -395,16 +406,18 @@ public class TestTablePermissions {
     Configuration conf = UTIL.getConfiguration();
 
     // add some permissions
-    AccessControlLists.addUserPermission(conf,
+    try (Connection connection = ConnectionFactory.createConnection(conf);
-        new UserPermission(Bytes.toBytes("user1"),
+        Table table = connection.getTable(AccessControlLists.ACL_TABLE_NAME)) {
-            Permission.Action.READ, Permission.Action.WRITE));
+      AccessControlLists.addUserPermission(conf,
-    AccessControlLists.addUserPermission(conf,
+          new UserPermission(Bytes.toBytes("user1"),
-        new UserPermission(Bytes.toBytes("user2"),
+              Permission.Action.READ, Permission.Action.WRITE), table);
-            Permission.Action.CREATE));
+      AccessControlLists.addUserPermission(conf,
-    AccessControlLists.addUserPermission(conf,
+          new UserPermission(Bytes.toBytes("user2"),
-        new UserPermission(Bytes.toBytes("user3"),
+              Permission.Action.CREATE), table);
-            Permission.Action.ADMIN, Permission.Action.READ, Permission.Action.CREATE));
+      AccessControlLists.addUserPermission(conf,
-
+          new UserPermission(Bytes.toBytes("user3"),
+              Permission.Action.ADMIN, Permission.Action.READ, Permission.Action.CREATE), table);
+    }
     ListMultimap<String,TablePermission> perms = AccessControlLists.getTablePermissions(conf, null);
     List<TablePermission> user1Perms = perms.get("user1");
     assertEquals("Should have 1 permission for user1", 1, user1Perms.size());
@@ -437,12 +450,15 @@ public class TestTablePermissions {
     // currently running user is the system user and should have global admin perms
     User currentUser = User.getCurrent();
     assertTrue(authManager.authorize(currentUser, Permission.Action.ADMIN));
-    for (int i=1; i<=50; i++) {
+    try (Connection connection = ConnectionFactory.createConnection(conf);
-      AccessControlLists.addUserPermission(conf, new UserPermission(Bytes.toBytes("testauth"+i),
+        Table table = connection.getTable(AccessControlLists.ACL_TABLE_NAME)) {
-          Permission.Action.ADMIN, Permission.Action.READ, Permission.Action.WRITE));
+      for (int i=1; i<=50; i++) {
-      // make sure the system user still shows as authorized
+        AccessControlLists.addUserPermission(conf, new UserPermission(Bytes.toBytes("testauth"+i),
-      assertTrue("Failed current user auth check on iter "+i,
+            Permission.Action.ADMIN, Permission.Action.READ, Permission.Action.WRITE), table);
-          authManager.authorize(currentUser, Permission.Action.ADMIN));
+        // make sure the system user still shows as authorized
+        assertTrue("Failed current user auth check on iter "+i,
+            authManager.authorize(currentUser, Permission.Action.ADMIN));
+      }
     }
   }
 }