HBASE-20357 AccessControlClient API Enhancement

Signed-off-by: tedyu <yuzhihong@gmail.com>

hbase-client/src/main/java/org/apache/hadoop/hbase/security/access/AccessControlClient.java

@@ -22,6 +22,7 @@ import java.util.ArrayList;
 import java.util.List;
 import java.util.regex.Pattern;
 
+import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.hbase.HConstants;
 import org.apache.hadoop.hbase.HTableDescriptor;
 import org.apache.hadoop.hbase.MasterNotRunningException;
@@ -255,14 +256,27 @@ public class AccessControlClient {
    * along with the list of superusers would be returned. Else, no rows get returned.
    * @param connection The Connection instance to use
    * @param tableRegex The regular expression string to match against
-   * @return - returns an array of UserPermissions
+   * @return List of UserPermissions
    * @throws Throwable
    */
   public static List<UserPermission> getUserPermissions(Connection connection, String tableRegex)
       throws Throwable {
-    /** TODO: Pass an rpcController
+    return getUserPermissions(connection, tableRegex, HConstants.EMPTY_STRING);
-    HBaseRpcController controller
+  }
-      = ((ClusterConnection) connection).getRpcControllerFactory().newController();
+
+  /**
+   * List all the userPermissions matching the given table pattern and user name.
+   * @param connection Connection
+   * @param tableRegex The regular expression string to match against
+   * @param userName User name, if empty then all user permissions will be retrieved.
+   * @return List of UserPermissions
+   * @throws Throwable on failure
+   */
+  public static List<UserPermission> getUserPermissions(Connection connection, String tableRegex,
+      String userName) throws Throwable {
+    /**
+     * TODO: Pass an rpcController HBaseRpcController controller = ((ClusterConnection)
+     * connection).getRpcControllerFactory().newController();
      */
     List<UserPermission> permList = new ArrayList<>();
     try (Table table = connection.getTable(ACL_TABLE_NAME)) {
@@ -272,25 +286,167 @@ public class AccessControlClient {
             AccessControlProtos.AccessControlService.newBlockingStub(service);
         HTableDescriptor[] htds = null;
         if (tableRegex == null || tableRegex.isEmpty()) {
-          permList = AccessControlUtil.getUserPermissions(null, protocol);
+          permList = AccessControlUtil.getUserPermissions(null, protocol, userName);
         } else if (tableRegex.charAt(0) == '@') { // Namespaces
           String namespaceRegex = tableRegex.substring(1);
-          for (NamespaceDescriptor nsds : admin.listNamespaceDescriptors()) {  // Read out all namespaces
+          for (NamespaceDescriptor nsds : admin.listNamespaceDescriptors()) { // Read out all
+                                                                              // namespaces
             String namespace = nsds.getName();
             if (namespace.matches(namespaceRegex)) { // Match the given namespace regex?
               permList.addAll(AccessControlUtil.getUserPermissions(null, protocol,
-                Bytes.toBytes(namespace)));
+                Bytes.toBytes(namespace), userName));
             }
           }
         } else { // Tables
           htds = admin.listTables(Pattern.compile(tableRegex), true);
-          for (HTableDescriptor hd : htds) {
+          for (HTableDescriptor htd : htds) {
-            permList.addAll(AccessControlUtil.getUserPermissions(null, protocol,
+            permList.addAll(AccessControlUtil.getUserPermissions(null, protocol, htd.getTableName(),
-              hd.getTableName()));
+              null, null, userName));
           }
         }
       }
     }
     return permList;
   }
+
+  /**
+   * List all the userPermissions matching the given table pattern and column family.
+   * @param connection Connection
+   * @param tableRegex The regular expression string to match against. It shouldn't be null, empty
+   *          or a namespace regular expression.
+   * @param columnFamily Column family
+   * @return List of UserPermissions
+   * @throws Throwable on failure
+   */
+  public static List<UserPermission> getUserPermissions(Connection connection, String tableRegex,
+      byte[] columnFamily) throws Throwable {
+    return getUserPermissions(connection, tableRegex, columnFamily, null, HConstants.EMPTY_STRING);
+  }
+
+  /**
+   * List all the userPermissions matching the given table pattern, column family and user name.
+   * @param connection Connection
+   * @param tableRegex The regular expression string to match against. It shouldn't be null, empty
+   *          or a namespace regular expression.
+   * @param columnFamily Column family
+   * @param userName User name, if empty then all user permissions will be retrieved.
+   * @return List of UserPermissions
+   * @throws Throwable on failure
+   */
+  public static List<UserPermission> getUserPermissions(Connection connection, String tableRegex,
+      byte[] columnFamily, String userName) throws Throwable {
+    return getUserPermissions(connection, tableRegex, columnFamily, null, userName);
+  }
+
+  /**
+   * List all the userPermissions matching the given table pattern, column family and column
+   * qualifier.
+   * @param connection Connection
+   * @param tableRegex The regular expression string to match against. It shouldn't be null, empty
+   *          or a namespace regular expression.
+   * @param columnFamily Column family
+   * @param columnQualifier Column qualifier
+   * @return List of UserPermissions
+   * @throws Throwable on failure
+   */
+  public static List<UserPermission> getUserPermissions(Connection connection, String tableRegex,
+      byte[] columnFamily, byte[] columnQualifier) throws Throwable {
+    return getUserPermissions(connection, tableRegex, columnFamily, columnQualifier,
+      HConstants.EMPTY_STRING);
+  }
+
+  /**
+   * List all the userPermissions matching the given table pattern, column family and column
+   * qualifier.
+   * @param connection Connection
+   * @param tableRegex The regular expression string to match against. It shouldn't be null, empty
+   *          or a namespace regular expression.
+   * @param columnFamily Column family
+   * @param columnQualifier Column qualifier
+   * @param userName User name, if empty then all user permissions will be retrieved.
+   * @return List of UserPermissions
+   * @throws Throwable on failure
+   */
+  public static List<UserPermission> getUserPermissions(Connection connection, String tableRegex,
+      byte[] columnFamily, byte[] columnQualifier, String userName) throws Throwable {
+    if (tableRegex == null || tableRegex.isEmpty() || tableRegex.charAt(0) == '@') {
+      throw new IllegalArgumentException("Table name can't be null or empty or a namespace.");
+    }
+    /**
+     * TODO: Pass an rpcController HBaseRpcController controller = ((ClusterConnection)
+     * connection).getRpcControllerFactory().newController();
+     */
+    List<UserPermission> permList = new ArrayList<UserPermission>();
+    try (Table table = connection.getTable(ACL_TABLE_NAME)) {
+      try (Admin admin = connection.getAdmin()) {
+        CoprocessorRpcChannel service = table.coprocessorService(HConstants.EMPTY_START_ROW);
+        BlockingInterface protocol =
+            AccessControlProtos.AccessControlService.newBlockingStub(service);
+        HTableDescriptor[] htds = admin.listTables(Pattern.compile(tableRegex), true);
+        // Retrieve table permissions
+        for (HTableDescriptor htd : htds) {
+          permList.addAll(AccessControlUtil.getUserPermissions(null, protocol, htd.getTableName(),
+            columnFamily, columnQualifier, userName));
+        }
+      }
+    }
+    return permList;
+  }
+
+  /**
+   * Validates whether specified user has permission to perform actions on the mentioned table,
+   * column family or column qualifier.
+   * @param connection Connection
+   * @param tableName Table name, it shouldn't be null or empty.
+   * @param columnFamily The column family. Optional argument, can be empty. If empty then
+   *          validation will happen at table level.
+   * @param columnQualifier The column qualifier. Optional argument, can be empty. If empty then
+   *          validation will happen at table and column family level. columnQualifier will not be
+   *          considered if columnFamily is passed as null or empty.
+   * @param userName User name, it shouldn't be null or empty.
+   * @param actions Actions
+   * @return true if access allowed to the specified user, otherwise false.
+   * @throws Throwable on failure
+   */
+  public static boolean hasPermission(Connection connection, String tableName, String columnFamily,
+      String columnQualifier, String userName, Permission.Action... actions) throws Throwable {
+    return hasPermission(connection, tableName, Bytes.toBytes(columnFamily),
+      Bytes.toBytes(columnQualifier), userName, actions);
+  }
+
+  /**
+   * Validates whether specified user has permission to perform actions on the mentioned table,
+   * column family or column qualifier.
+   * @param connection Connection
+   * @param tableName Table name, it shouldn't be null or empty.
+   * @param columnFamily The column family. Optional argument, can be empty. If empty then
+   *          validation will happen at table level.
+   * @param columnQualifier The column qualifier. Optional argument, can be empty. If empty then
+   *          validation will happen at table and column family level. columnQualifier will not be
+   *          considered if columnFamily is passed as null or empty.
+   * @param userName User name, it shouldn't be null or empty.
+   * @param actions Actions
+   * @return true if access allowed to the specified user, otherwise false.
+   * @throws Throwable on failure
+   */
+  public static boolean hasPermission(Connection connection, String tableName, byte[] columnFamily,
+      byte[] columnQualifier, String userName, Permission.Action... actions) throws Throwable {
+    if (StringUtils.isEmpty(tableName) || StringUtils.isEmpty(userName)) {
+      throw new IllegalArgumentException("Table and user name can't be null or empty.");
+    }
+    boolean hasPermission = false;
+    /**
+     * todo: pass an rpccontroller hbaserpccontroller controller = ((clusterconnection)
+     * connection).getrpccontrollerfactory().newcontroller();
+     */
+    try (Table table = connection.getTable(ACL_TABLE_NAME)) {
+      CoprocessorRpcChannel service = table.coprocessorService(HConstants.EMPTY_START_ROW);
+      BlockingInterface protocol =
+          AccessControlProtos.AccessControlService.newBlockingStub(service);
+      // Check whether user has permission
+      hasPermission = AccessControlUtil.hasPermission(null, protocol, TableName.valueOf(tableName),
+        columnFamily, columnQualifier, userName, actions);
+    }
+    return hasPermission;
+  }
 }

hbase-client/src/main/java/org/apache/hadoop/hbase/security/access/AccessControlUtil.java

@@ -22,6 +22,8 @@ import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.hbase.HConstants;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.yetus.audience.InterfaceAudience;
 import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
@@ -29,7 +31,7 @@ import org.apache.hadoop.hbase.protobuf.generated.AccessControlProtos;
 import org.apache.hadoop.hbase.protobuf.generated.AccessControlProtos.AccessControlService;
 import org.apache.hadoop.hbase.protobuf.generated.AccessControlProtos.GetUserPermissionsResponse;
 import org.apache.hadoop.hbase.util.ByteStringer;
-
+import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hbase.thirdparty.com.google.common.collect.ArrayListMultimap;
 import org.apache.hbase.thirdparty.com.google.common.collect.ListMultimap;
 import org.apache.hbase.thirdparty.com.google.common.collect.Lists;
@@ -255,9 +257,32 @@ public class AccessControlUtil {
     }
   }
 
+  /**
+   * Converts a TablePermission proto to a client TablePermission object.
+   * @param proto the protobuf TablePermission
+   * @return the converted TablePermission
+   */
+  public static TablePermission toTablePermission(AccessControlProtos.TablePermission proto) {
+    List<Permission.Action> actions = toPermissionActions(proto.getActionList());
+    TableName table = null;
+    byte[] qualifier = null;
+    byte[] family = null;
+    if (!proto.hasTableName()) {
+      throw new IllegalStateException("TableName cannot be empty");
+    }
+    table = ProtobufUtil.toTableName(proto.getTableName());
+    if (proto.hasFamily()) {
+      family = proto.getFamily().toByteArray();
+    }
+    if (proto.hasQualifier()) {
+      qualifier = proto.getQualifier().toByteArray();
+    }
+    return new TablePermission(table, family, qualifier,
+        actions.toArray(new Permission.Action[actions.size()]));
+  }
+
   /**
    * Converts a Permission proto to a client TablePermission object.
-   *
    * @param proto the protobuf Permission
    * @return the converted TablePermission
    */
@@ -539,6 +564,7 @@ public class AccessControlUtil {
    * <p>
    * It's also called by the shell, in case you want to find references.
    *
+   * @param controller RpcController
    * @param protocol the AccessControlService protocol proxy
    * @param namespace the short name of the user to grant permissions
    * @param actions the permissions to be granted
@@ -562,10 +588,11 @@ public class AccessControlUtil {
    * <p>
    * It's also called by the shell, in case you want to find references.
    *
+   * @param controller RpcController
    * @param protocol the AccessControlService protocol proxy
    * @param userShortName the short name of the user to revoke permissions
    * @param actions the permissions to be revoked
-   * @throws ServiceException
+   * @throws ServiceException on failure
    */
   public static void revoke(RpcController controller,
       AccessControlService.BlockingInterface protocol, String userShortName,
@@ -586,13 +613,14 @@ public class AccessControlUtil {
    * <p>
    * It's also called by the shell, in case you want to find references.
    *
+   * @param controller RpcController
    * @param protocol the AccessControlService protocol proxy
    * @param userShortName the short name of the user to revoke permissions
    * @param tableName optional table name
    * @param f optional column family
    * @param q optional qualifier
    * @param actions the permissions to be revoked
-   * @throws ServiceException
+   * @throws ServiceException on failure
    */
   public static void revoke(RpcController controller,
       AccessControlService.BlockingInterface protocol, String userShortName, TableName tableName,
@@ -612,11 +640,12 @@ public class AccessControlUtil {
    * <p>
    * It's also called by the shell, in case you want to find references.
    *
+   * @param controller RpcController
    * @param protocol the AccessControlService protocol proxy
    * @param userShortName the short name of the user to revoke permissions
    * @param namespace optional table name
    * @param actions the permissions to be revoked
-   * @throws ServiceException
+   * @throws ServiceException on failure
    */
   public static void revoke(RpcController controller,
       AccessControlService.BlockingInterface protocol, String userShortName, String namespace,
@@ -636,14 +665,31 @@ public class AccessControlUtil {
    * <p>
    * It's also called by the shell, in case you want to find references.
    *
+   * @param controller RpcController
    * @param protocol the AccessControlService protocol proxy
-   * @throws ServiceException
+   * @throws ServiceException on failure
    */
   public static List<UserPermission> getUserPermissions(RpcController controller,
       AccessControlService.BlockingInterface protocol) throws ServiceException {
+    return getUserPermissions(controller, protocol, HConstants.EMPTY_STRING);
+  }
+
+  /**
+   * A utility used to get user's global permissions based on the specified user name.
+   * @param controller RpcController
+   * @param protocol the AccessControlService protocol proxy
+   * @param userName User name, if empty then all user permissions will be retrieved.
+   * @throws ServiceException
+   */
+  public static List<UserPermission> getUserPermissions(RpcController controller,
+      AccessControlService.BlockingInterface protocol, String userName) throws ServiceException {
     AccessControlProtos.GetUserPermissionsRequest.Builder builder =
         AccessControlProtos.GetUserPermissionsRequest.newBuilder();
     builder.setType(AccessControlProtos.Permission.Type.Global);
+    if (!StringUtils.isEmpty(userName)) {
+      builder.setUserName(ByteString.copyFromUtf8(userName));
+    }
+
     AccessControlProtos.GetUserPermissionsRequest request = builder.build();
     AccessControlProtos.GetUserPermissionsResponse response =
         protocol.getUserPermissions(controller, request);
@@ -659,6 +705,7 @@ public class AccessControlUtil {
    * <p>
    * It's also called by the shell, in case you want to find references.
    *
+   * @param controller RpcController
    * @param protocol the AccessControlService protocol proxy
    * @param t optional table name
    * @throws ServiceException
@@ -666,11 +713,38 @@ public class AccessControlUtil {
   public static List<UserPermission> getUserPermissions(RpcController controller,
       AccessControlService.BlockingInterface protocol,
       TableName t) throws ServiceException {
+    return getUserPermissions(controller, protocol, t, null, null, HConstants.EMPTY_STRING);
+  }
+
+  /**
+   * A utility used to get user table permissions based on the column family, column qualifier and
+   * user name.
+   * @param controller RpcController
+   * @param protocol the AccessControlService protocol proxy
+   * @param t optional table name
+   * @param columnFamily Column family
+   * @param columnQualifier Column qualifier
+   * @param userName User name, if empty then all user permissions will be retrieved.
+   * @throws ServiceException
+   */
+  public static List<UserPermission> getUserPermissions(RpcController controller,
+      AccessControlService.BlockingInterface protocol, TableName t, byte[] columnFamily,
+      byte[] columnQualifier, String userName) throws ServiceException {
     AccessControlProtos.GetUserPermissionsRequest.Builder builder =
         AccessControlProtos.GetUserPermissionsRequest.newBuilder();
     if (t != null) {
       builder.setTableName(ProtobufUtil.toProtoTableName(t));
     }
+    if (Bytes.len(columnFamily) > 0) {
+      builder.setColumnFamily(ByteString.copyFrom(columnFamily));
+    }
+    if (Bytes.len(columnQualifier) > 0) {
+      builder.setColumnQualifier(ByteString.copyFrom(columnQualifier));
+    }
+    if (!StringUtils.isEmpty(userName)) {
+      builder.setUserName(ByteString.copyFromUtf8(userName));
+    }
+
     builder.setType(AccessControlProtos.Permission.Type.Table);
     AccessControlProtos.GetUserPermissionsRequest request = builder.build();
     AccessControlProtos.GetUserPermissionsResponse response =
@@ -687,6 +761,7 @@ public class AccessControlUtil {
    * <p>
    * It's also called by the shell, in case you want to find references.
    *
+   * @param controller RpcController
    * @param protocol the AccessControlService protocol proxy
    * @param namespace name of the namespace
    * @throws ServiceException
@@ -694,11 +769,28 @@ public class AccessControlUtil {
   public static List<UserPermission> getUserPermissions(RpcController controller,
       AccessControlService.BlockingInterface protocol,
       byte[] namespace) throws ServiceException {
+    return getUserPermissions(controller, protocol, namespace, HConstants.EMPTY_STRING);
+  }
+
+  /**
+   * A utility used to get permissions for selected namespace based on the specified user name.
+   * @param controller RpcController
+   * @param protocol the AccessControlService protocol proxy
+   * @param namespace name of the namespace
+   * @param userName User name, if empty then all user permissions will be retrieved.
+   * @throws ServiceException
+   */
+  public static List<UserPermission> getUserPermissions(RpcController controller,
+      AccessControlService.BlockingInterface protocol, byte[] namespace, String userName)
+      throws ServiceException {
     AccessControlProtos.GetUserPermissionsRequest.Builder builder =
         AccessControlProtos.GetUserPermissionsRequest.newBuilder();
     if (namespace != null) {
       builder.setNamespaceName(ByteStringer.wrap(namespace));
     }
+    if (!StringUtils.isEmpty(userName)) {
+      builder.setUserName(ByteString.copyFromUtf8(userName));
+    }
     builder.setType(AccessControlProtos.Permission.Type.Namespace);
     AccessControlProtos.GetUserPermissionsRequest request = builder.build();
     AccessControlProtos.GetUserPermissionsResponse response =
@@ -710,6 +802,47 @@ public class AccessControlUtil {
     return perms;
   }
 
+  /**
+   * Validates whether specified user has permission to perform actions on the mentioned table,
+   * column family or column qualifier.
+   * @param controller RpcController
+   * @param protocol the AccessControlService protocol proxy
+   * @param tableName Table name, it shouldn't be null or empty.
+   * @param columnFamily The column family. Optional argument, can be empty. If empty then
+   *          validation will happen at table level.
+   * @param columnQualifier The column qualifier. Optional argument, can be empty. If empty then
+   *          validation will happen at table and column family level. columnQualifier will not be
+   *          considered if columnFamily is passed as null or empty.
+   * @param userName User name, it shouldn't be null or empty.
+   * @param actions Actions
+   * @return true if access allowed, otherwise false
+   * @throws ServiceException
+   */
+  public static boolean hasPermission(RpcController controller,
+      AccessControlService.BlockingInterface protocol, TableName tableName, byte[] columnFamily,
+      byte[] columnQualifier, String userName, Permission.Action[] actions)
+      throws ServiceException {
+    AccessControlProtos.TablePermission.Builder tablePermissionBuilder =
+        AccessControlProtos.TablePermission.newBuilder();
+    tablePermissionBuilder
+        .setTableName(org.apache.hadoop.hbase.protobuf.ProtobufUtil.toProtoTableName(tableName));
+    if (Bytes.len(columnFamily) > 0) {
+      tablePermissionBuilder.setFamily(ByteStringer.wrap(columnFamily));
+    }
+    if (Bytes.len(columnQualifier) > 0) {
+      tablePermissionBuilder.setQualifier(ByteString.copyFrom(columnQualifier));
+    }
+    for (Permission.Action a : actions) {
+      tablePermissionBuilder.addAction(toPermissionAction(a));
+    }
+    AccessControlProtos.HasPermissionRequest request = AccessControlProtos.HasPermissionRequest
+        .newBuilder().setTablePermission(tablePermissionBuilder)
+        .setUserName(ByteString.copyFromUtf8(userName)).build();
+    AccessControlProtos.HasPermissionResponse response =
+        protocol.hasPermission(controller, request);
+    return response.getHasPermission();
+  }
+
   /**
    * Convert a protobuf UserTablePermissions to a
    * ListMultimap&lt;String, TablePermission&gt; where key is username.

hbase-client/src/main/java/org/apache/hadoop/hbase/shaded/protobuf/ResponseConverter.java

@@ -31,6 +31,7 @@ import org.apache.hadoop.hbase.client.RegionInfo;
 import org.apache.hadoop.hbase.client.Result;
 import org.apache.hadoop.hbase.client.SingleResponse;
 import org.apache.hadoop.hbase.ipc.ServerRpcController;
+import org.apache.hadoop.hbase.protobuf.generated.AccessControlProtos.HasPermissionResponse;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos;
 import org.apache.hadoop.util.StringUtils;
 import org.apache.yetus.audience.InterfaceAudience;
@@ -228,6 +229,15 @@ public final class ResponseConverter {
     return parameterBuilder.build();
   }
 
+  /**
+   * Builds a protocol buffer HasPermissionResponse
+   */
+  public static HasPermissionResponse buildHasPermissionResponse(boolean hasPermission) {
+    HasPermissionResponse.Builder builder = HasPermissionResponse.newBuilder();
+    builder.setHasPermission(hasPermission);
+    return builder.build();
+  }
+
 // End utilities for Client
 // Start utilities for Admin
 

hbase-common/src/main/java/org/apache/hadoop/hbase/HConstants.java

@@ -530,10 +530,15 @@ public final class HConstants {
   // Other constants
 
   /**
-   * An empty instance.
+   * An empty byte array instance.
    */
   public static final byte [] EMPTY_BYTE_ARRAY = new byte [0];
 
+  /**
+   * An empty string instance.
+   */
+  public static final String EMPTY_STRING = "";
+
   public static final ByteBuffer EMPTY_BYTE_BUFFER = ByteBuffer.wrap(EMPTY_BYTE_ARRAY);
 
   /**

hbase-protocol/src/main/protobuf/AccessControl.proto

@@ -96,6 +96,9 @@ message GetUserPermissionsRequest {
   optional Permission.Type type = 1;
   optional TableName table_name = 2;
   optional bytes namespace_name = 3;
+  optional bytes column_family = 4;
+  optional bytes column_qualifier = 5;
+  optional bytes user_name = 6;
 }
 
 message GetUserPermissionsResponse {
@@ -109,6 +112,15 @@ message CheckPermissionsRequest {
 message CheckPermissionsResponse {
 }
 
+message HasPermissionRequest {
+  required TablePermission table_permission = 1;
+  required bytes user_name = 2;
+}
+
+message HasPermissionResponse {
+  optional bool has_permission = 1;
+}
+
 service AccessControlService {
     rpc Grant(GrantRequest)
       returns (GrantResponse);
@@ -121,4 +133,7 @@ service AccessControlService {
 
     rpc CheckPermissions(CheckPermissionsRequest)
       returns (CheckPermissionsResponse);
+
+    rpc HasPermission(HasPermissionRequest)
+      returns (HasPermissionResponse);
 }

hbase-rsgroup/src/main/java/org/apache/hadoop/hbase/rsgroup/RSGroupAdminEndpoint.java

@@ -543,7 +543,7 @@ public class RSGroupAdminEndpoint implements MasterCoprocessor, MasterObserver {
   }
 
   public void checkPermission(String request) throws IOException {
-    accessChecker.requirePermission(getActiveUser(), request, Action.ADMIN);
+    accessChecker.requirePermission(getActiveUser(), request, null, Action.ADMIN);
   }
 
   /**

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java

@@ -1294,7 +1294,7 @@ public class RSRpcServices implements HBaseRPCErrorHandler,
 
   protected void requirePermission(String request, Permission.Action perm) throws IOException {
     if (accessChecker != null) {
-      accessChecker.requirePermission(RpcServer.getRequestUser().orElse(null), request, perm);
+      accessChecker.requirePermission(RpcServer.getRequestUser().orElse(null), request, null, perm);
     }
   }
 

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

@@ -20,7 +20,11 @@ package org.apache.hadoop.hbase.security.access;
 
 import java.io.IOException;
 import java.net.InetAddress;
+import java.security.PrivilegedAction;
+import java.security.PrivilegedExceptionAction;
+import java.util.ArrayList;
 import java.util.Collection;
+import java.util.List;
 import java.util.Map;
 
 import org.apache.hadoop.conf.Configuration;
@@ -32,18 +36,25 @@ import org.apache.hadoop.hbase.security.AccessDeniedException;
 import org.apache.hadoop.hbase.security.User;
 import org.apache.hadoop.hbase.security.access.Permission.Action;
 import org.apache.hadoop.hbase.zookeeper.ZKWatcher;
+import org.apache.hadoop.security.Groups;
+import org.apache.hadoop.security.HadoopKerberosName;
 import org.apache.yetus.audience.InterfaceAudience;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 @InterfaceAudience.Private
 public final class AccessChecker {
+  private static final Logger LOG = LoggerFactory.getLogger(AccessChecker.class);
   private static final Logger AUDITLOG =
       LoggerFactory.getLogger("SecurityLogger." + AccessChecker.class.getName());
   // TODO: we should move to a design where we don't even instantiate an AccessChecker if
   // authorization is not enabled (like in RSRpcServices), instead of always instantiating one and
   // calling requireXXX() only to do nothing (since authorizationEnabled will be false).
   private TableAuthManager authManager;
+
+  /** Group service to retrieve the user group information */
+  private static Groups groupService;
+
   /**
    * if we are active, usually false, only true if "hbase.security.authorization"
    * has been set to true in site configuration.see HBASE-19483.
@@ -72,6 +83,7 @@ public final class AccessChecker {
       throw new NullPointerException("Error obtaining AccessChecker, zk found null.");
     }
     authorizationEnabled = isAuthorizationSupported(conf);
+    initGroupService(conf);
   }
 
   /**
@@ -88,6 +100,8 @@ public final class AccessChecker {
   /**
    * Authorizes that the current user has any of the given permissions to access the table.
    *
+   * @param user Active user to which authorization checks should be applied
+   * @param request Request type.
    * @param tableName   Table requested
    * @param permissions Actions being requested
    * @throws IOException if obtaining the current user fails
@@ -119,14 +133,16 @@ public final class AccessChecker {
 
   /**
    * Authorizes that the current user has global privileges for the given action.
-   *
+   * @param user Active user to which authorization checks should be applied
+   * @param request Request type
+   * @param filterUser User name to be filtered from permission as requested
    * @param perm The action being requested
    * @throws IOException if obtaining the current user fails
    * @throws AccessDeniedException if authorization is denied
    */
-  public void requirePermission(User user, String request, Action perm)
+  public void requirePermission(User user, String request, String filterUser, Action perm)
       throws IOException {
-    requireGlobalPermission(user, request, perm, null, null);
+    requireGlobalPermission(user, request, perm, null, null, filterUser);
   }
 
   /**
@@ -134,25 +150,29 @@ public final class AccessChecker {
    * audit log message will contain context information for the operation
    * being authorized, based on the given parameters.
    *
+   * @param user Active user to which authorization checks should be applied
+   * @param request Request type
    * @param perm      Action being requested
    * @param tableName Affected table name.
    * @param familyMap Affected column families.
+   * @param filterUser User name to be filtered from permission as requested
    */
   public void requireGlobalPermission(User user, String request,
       Action perm, TableName tableName,
-      Map<byte[], ? extends Collection<byte[]>> familyMap)throws IOException {
+      Map<byte[], ? extends Collection<byte[]>> familyMap, String filterUser) throws IOException {
     if (!authorizationEnabled) {
       return;
     }
     AuthResult result;
     if (authManager.authorize(user, perm)) {
       result = AuthResult.allow(request, "Global check allowed", user, perm, tableName, familyMap);
-      result.getParams().setTableName(tableName).setFamilies(familyMap);
-      logResult(result);
     } else {
       result = AuthResult.deny(request, "Global check failed", user, perm, tableName, familyMap);
+    }
     result.getParams().setTableName(tableName).setFamilies(familyMap);
+    result.getParams().addExtraParam("filterUser", filterUser);
     logResult(result);
+    if (!result.isAllowed()) {
       throw new AccessDeniedException(
           "Insufficient permissions for user '" + (user != null ? user.getShortName() : "null")
               + "' (global, action=" + perm.toString() + ")");
@@ -164,6 +184,8 @@ public final class AccessChecker {
    * audit log message will contain context information for the operation
    * being authorized, based on the given parameters.
    *
+   * @param user Active user to which authorization checks should be applied
+   * @param request Request type
    * @param perm      Action being requested
    * @param namespace The given namespace
    */
@@ -189,12 +211,14 @@ public final class AccessChecker {
 
   /**
    * Checks that the user has the given global or namespace permission.
-   *
+   * @param user Active user to which authorization checks should be applied
-   * @param namespace  The given namespace
+   * @param request Request type
+   * @param namespace Name space as requested
+   * @param filterUser User name to be filtered from permission as requested
    * @param permissions Actions being requested
    */
   public void requireNamespacePermission(User user, String request, String namespace,
-      Action... permissions) throws IOException {
+      String filterUser, Action... permissions) throws IOException {
     if (!authorizationEnabled) {
       return;
     }
@@ -210,6 +234,7 @@ public final class AccessChecker {
         result = AuthResult.deny(request, "Insufficient permissions", user, permission, namespace);
       }
     }
+    result.getParams().addExtraParam("filterUser", filterUser);
     logResult(result);
     if (!result.isAllowed()) {
       throw new AccessDeniedException("Insufficient permissions " + result.toContextString());
@@ -219,7 +244,11 @@ public final class AccessChecker {
   /**
    * Checks that the user has the given global or namespace permission.
    *
+   * @param user Active user to which authorization checks should be applied
+   * @param request Request type
    * @param namespace  The given namespace
+   * @param tableName Table requested
+   * @param familyMap    Column family map requested
    * @param permissions Actions being requested
    */
   public void requireNamespacePermission(User user, String request, String namespace,
@@ -252,14 +281,18 @@ public final class AccessChecker {
    * Authorizes that the current user has any of the given permissions for the
    * given table, column family and column qualifier.
    *
+   * @param user Active user to which authorization checks should be applied
+   * @param request Request type
    * @param tableName Table requested
    * @param family    Column family requested
    * @param qualifier Column qualifier requested
+   * @param filterUser User name to be filtered from permission as requested
+   * @param permissions Actions being requested
    * @throws IOException if obtaining the current user fails
    * @throws AccessDeniedException if user has no authorization
    */
   public void requirePermission(User user, String request, TableName tableName, byte[] family,
-      byte[] qualifier, Action... permissions) throws IOException {
+      byte[] qualifier, String filterUser, Action... permissions) throws IOException {
     if (!authorizationEnabled) {
       return;
     }
@@ -276,6 +309,7 @@ public final class AccessChecker {
           user, permission, tableName, family, qualifier);
       }
     }
+    result.getParams().addExtraParam("filterUser", filterUser);
     logResult(result);
     if (!result.isAllowed()) {
       throw new AccessDeniedException("Insufficient permissions " + result.toContextString());
@@ -286,6 +320,8 @@ public final class AccessChecker {
    * Authorizes that the current user has any of the given permissions for the
    * given table, column family and column qualifier.
    *
+   * @param user Active user to which authorization checks should be applied
+   * @param request Request type
    * @param tableName Table requested
    * @param family    Column family param
    * @param qualifier Column qualifier param
@@ -323,7 +359,7 @@ public final class AccessChecker {
       TableName tableName, RegionInfo[] regionInfos, String reason)
       throws IOException {
     if (namespace != null && !namespace.isEmpty()) {
-      requireNamespacePermission(user, reason, namespace, Action.ADMIN, Action.CREATE);
+      requireNamespacePermission(user, reason, namespace, null, Action.ADMIN, Action.CREATE);
     } else if (tableName != null || (regionInfos != null && regionInfos.length > 0)) {
       // So, either a table or regions op. If latter, check perms ons table.
       TableName tn = tableName != null? tableName: regionInfos[0].getTable();
@@ -340,9 +376,111 @@ public final class AccessChecker {
         "Access {} for user {}; reason: {}; remote address: {}; request: {}; context: {}",
         (result.isAllowed() ? "allowed" : "denied"),
         (result.getUser() != null ? result.getUser().getShortName() : "UNKNOWN"),
-        result.getReason(),
+        result.getReason(), RpcServer.getRemoteAddress().map(InetAddress::toString).orElse(""),
-        RpcServer.getRemoteAddress().map(InetAddress::toString).orElse(""),
         result.getRequest(), result.toContextString());
     }
   }
+
+  /*
+   * Validate the hasPermission operation caller with the filter user. Self check doesn't require
+   * any privilege but for others caller must have ADMIN privilege.
+   */
+  public User validateCallerWithFilterUser(User caller, TablePermission tPerm, String inputUserName)
+      throws IOException {
+    User filterUser = null;
+    if (!caller.getShortName().equals(inputUserName)) {
+      // User should have admin privilege if checking permission for other users
+      requirePermission(caller, "hasPermission", tPerm.getTableName(), tPerm.getFamily(),
+        tPerm.getQualifier(), inputUserName, Action.ADMIN);
+      // Initialize user instance for the input user name
+      List<String> groups = getUserGroups(inputUserName);
+      filterUser = new InputUser(inputUserName, groups.toArray(new String[groups.size()]));
+    } else {
+      // User don't need ADMIN privilege for self check.
+      // Setting action as null in AuthResult to display empty action in audit log
+      AuthResult result = AuthResult.allow("hasPermission", "Self user validation allowed", caller,
+        null, tPerm.getTableName(), tPerm.getFamily(), tPerm.getQualifier());
+      logResult(result);
+      filterUser = caller;
+    }
+    return filterUser;
+  }
+
+  /**
+   * A temporary user class to instantiate User instance based on the name and groups.
+   */
+  public static class InputUser extends User {
+    private String name;
+    private String shortName = null;
+    private String[] groups;
+
+    public InputUser(String name, String[] groups) {
+      this.name = name;
+      this.groups = groups;
+    }
+
+    @Override
+    public String getShortName() {
+      if (this.shortName == null) {
+        try {
+          this.shortName = new HadoopKerberosName(this.name).getShortName();
+        } catch (IOException ioe) {
+          throw new IllegalArgumentException(
+              "Illegal principal name " + this.name + ": " + ioe.toString(), ioe);
+        }
+      }
+      return shortName;
+    }
+
+    @Override
+    public String getName() {
+      return this.name;
+    }
+
+    @Override
+    public String[] getGroupNames() {
+      return this.groups;
+    }
+
+    @Override
+    public <T> T runAs(PrivilegedAction<T> action) {
+      throw new UnsupportedOperationException(
+          "Method not supported, this class has limited implementation");
+    }
+
+    @Override
+    public <T> T runAs(PrivilegedExceptionAction<T> action)
+        throws IOException, InterruptedException {
+      throw new UnsupportedOperationException(
+          "Method not supported, this class has limited implementation");
+    }
+
+    @Override
+    public String toString() {
+      return this.name;
+    }
+  }
+
+  /*
+   * Initialize the group service.
+   */
+  private void initGroupService(Configuration conf) {
+    if (groupService == null) {
+      groupService = Groups.getUserToGroupsMappingService(conf);
+    }
+  }
+
+  /**
+   * Retrieve the groups of the given user.
+   * @param user User name
+   * @return Groups
+   */
+  public static List<String> getUserGroups(String user) {
+    try {
+      return groupService.getGroups(user);
+    } catch (IOException e) {
+      LOG.error("Error occured while retrieving group for " + user, e);
+      return new ArrayList<String>();
+    }
+  }
 }

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

@@ -32,6 +32,7 @@ import java.util.Set;
 import java.util.TreeMap;
 import java.util.TreeSet;
 
+import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.AuthUtil;
 import org.apache.hadoop.hbase.Cell;
@@ -144,7 +145,7 @@ public class AccessControlLists {
 
     Set<Permission.Action> actionSet = new TreeSet<Permission.Action>();
     if(mergeExistingPermissions){
-      List<UserPermission> perms = getUserPermissions(conf, rowKey);
+      List<UserPermission> perms = getUserPermissions(conf, rowKey, null, null, null, false);
       UserPermission currentPerm = null;
       for (UserPermission perm : perms) {
         if (Bytes.equals(perm.getUser(), userPerm.getUser())
@@ -228,7 +229,8 @@ public class AccessControlLists {
       removePermissionRecord(conf, userPerm, t);
     } else {
       // Get all the global user permissions from the acl table
-      List<UserPermission> permsList = getUserPermissions(conf, userPermissionRowKey(userPerm));
+      List<UserPermission> permsList =
+          getUserPermissions(conf, userPermissionRowKey(userPerm), null, null, null, false);
       List<Permission.Action> remainingActions = new ArrayList<>();
       List<Permission.Action> dropActions = Arrays.asList(userPerm.getActions());
       for (UserPermission perm : permsList) {
@@ -431,7 +433,7 @@ public class AccessControlLists {
             entry = CellUtil.cloneRow(kv);
           }
           Pair<String, TablePermission> permissionsOfUserOnTable =
-              parsePermissionRecord(entry, kv);
+              parsePermissionRecord(entry, kv, null, null, false, null);
           if (permissionsOfUserOnTable != null) {
             String username = permissionsOfUserOnTable.getFirst();
             TablePermission permissions = permissionsOfUserOnTable.getSecond();
@@ -474,7 +476,8 @@ public class AccessControlLists {
         scanner = table.getScanner(scan);
         try {
           for (Result row : scanner) {
-            ListMultimap<String,TablePermission> resultPerms = parsePermissions(row.getRow(), row);
+            ListMultimap<String, TablePermission> resultPerms =
+                parsePermissions(row.getRow(), row, null, null, null, false);
             allPerms.put(row.getRow(), resultPerms);
           }
         } finally {
@@ -488,28 +491,27 @@ public class AccessControlLists {
 
   public static ListMultimap<String, TablePermission> getTablePermissions(Configuration conf,
       TableName tableName) throws IOException {
-    return getPermissions(conf, tableName != null ? tableName.getName() : null, null);
+    return getPermissions(conf, tableName != null ? tableName.getName() : null, null, null, null,
+      null, false);
   }
 
   @VisibleForTesting
   public static ListMultimap<String, TablePermission> getNamespacePermissions(Configuration conf,
       String namespace) throws IOException {
-    return getPermissions(conf, Bytes.toBytes(toNamespaceEntry(namespace)), null);
+    return getPermissions(conf, Bytes.toBytes(toNamespaceEntry(namespace)), null, null, null, null,
+      false);
   }
 
   /**
-   * Reads user permission assignments stored in the <code>l:</code> column
+   * Reads user permission assignments stored in the <code>l:</code> column family of the first
-   * family of the first table row in <code>_acl_</code>.
+   * table row in <code>_acl_</code>.
-   *
    * <p>
-   * See {@link AccessControlLists class documentation} for the key structure
+   * See {@link AccessControlLists class documentation} for the key structure used for storage.
-   * used for storage.
    * </p>
    */
-  static ListMultimap<String, TablePermission> getPermissions(Configuration conf,
+  static ListMultimap<String, TablePermission> getPermissions(Configuration conf, byte[] entryName,
-      byte[] entryName, Table t) throws IOException {
+      Table t, byte[] cf, byte[] cq, String user, boolean hasFilterUser) 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();
     Get get = new Get(entryName);
@@ -525,7 +527,7 @@ public class AccessControlLists {
       row = t.get(get);
     }
     if (!row.isEmpty()) {
-      perms = parsePermissions(entryName, row);
+      perms = parsePermissions(entryName, row, cf, cq, user, hasFilterUser);
     } else {
       LOG.info("No permissions found in " + ACL_TABLE_NAME + " for acl entry "
           + Bytes.toString(entryName));
@@ -535,27 +537,43 @@ public class AccessControlLists {
   }
 
   /**
-   * Returns the currently granted permissions for a given table as a list of
+   * Returns the currently granted permissions for a given table as the specified user plus
-   * user plus associated permissions.
+   * associated permissions.
    */
-  static List<UserPermission> getUserTablePermissions(
+  static List<UserPermission> getUserTablePermissions(Configuration conf, TableName tableName,
-      Configuration conf, TableName tableName) throws IOException {
+      byte[] cf, byte[] cq, String userName, boolean hasFilterUser) throws IOException {
-    return getUserPermissions(conf, tableName == null ? null : tableName.getName());
+    return getUserPermissions(conf, tableName == null ? null : tableName.getName(), cf, cq,
+      userName, hasFilterUser);
   }
 
-  static List<UserPermission> getUserNamespacePermissions(
+  /**
-      Configuration conf, String namespace) throws IOException {
+   * Returns the currently granted permissions for a given namespace as the specified user plus
-    return getUserPermissions(conf, Bytes.toBytes(toNamespaceEntry(namespace)));
+   * associated permissions.
+   */
+  static List<UserPermission> getUserNamespacePermissions(Configuration conf, String namespace,
+      String user, boolean hasFilterUser) throws IOException {
+    return getUserPermissions(conf, Bytes.toBytes(toNamespaceEntry(namespace)), null, null, user,
+      hasFilterUser);
   }
 
-  static List<UserPermission> getUserPermissions(
+  /**
-      Configuration conf, byte[] entryName)
+   * Returns the currently granted permissions for a given table/namespace with associated
-          throws IOException {
+   * permissions based on the specified column family, column qualifier and user name.
-    ListMultimap<String,TablePermission> allPerms = getPermissions(
+   * @param conf the configuration
-        conf, entryName, null);
+   * @param entryName Table name or the namespace
+   * @param cf Column family
+   * @param cq Column qualifier
+   * @param user User name to be filtered from permission as requested
+   * @param hasFilterUser true if filter user is provided, otherwise false.
+   * @return List of UserPermissions
+   * @throws IOException on failure
+   */
+  static List<UserPermission> getUserPermissions(Configuration conf, byte[] entryName, byte[] cf,
+      byte[] cq, String user, boolean hasFilterUser) throws IOException {
+    ListMultimap<String, TablePermission> allPerms =
+        getPermissions(conf, entryName, null, cf, cq, user, hasFilterUser);
 
     List<UserPermission> perms = new ArrayList<>();
-
     if (isNamespaceEntry(entryName)) { // Namespace
       for (Map.Entry<String, TablePermission> entry : allPerms.entries()) {
         UserPermission up = new UserPermission(Bytes.toBytes(entry.getKey()),
@@ -570,17 +588,21 @@ public class AccessControlLists {
         perms.add(up);
       }
     }
+
     return perms;
   }
 
-  private static ListMultimap<String, TablePermission> parsePermissions(
+  /**
-      byte[] entryName, Result result) {
+   * Parse and filter permission based on the specified column family, column qualifier and user
+   * name.
+   */
+  private static ListMultimap<String, TablePermission> parsePermissions(byte[] entryName,
+      Result result, byte[] cf, byte[] cq, String user, boolean hasFilterUser) {
     ListMultimap<String, TablePermission> perms = ArrayListMultimap.create();
     if (result != null && result.size() > 0) {
       for (Cell kv : result.rawCells()) {
-
         Pair<String, TablePermission> permissionsOfUserOnTable =
-            parsePermissionRecord(entryName, kv);
+            parsePermissionRecord(entryName, kv, cf, cq, hasFilterUser, user);
 
         if (permissionsOfUserOnTable != null) {
           String username = permissionsOfUserOnTable.getFirst();
@@ -592,11 +614,10 @@ public class AccessControlLists {
     return perms;
   }
 
-  private static Pair<String, TablePermission> parsePermissionRecord(
+  private static Pair<String, TablePermission> parsePermissionRecord(byte[] entryName, Cell kv,
-      byte[] entryName, Cell kv) {
+      byte[] cf, byte[] cq, boolean filterPerms, String filterUser) {
     // return X given a set of permissions encoded in the permissionRecord kv.
     byte[] family = CellUtil.cloneFamily(kv);
-
     if (!Bytes.equals(family, ACL_LIST_FAMILY)) {
       return null;
     }
@@ -613,9 +634,25 @@ public class AccessControlLists {
     // TODO: avoid the string conversion to make this more efficient
     String username = Bytes.toString(key);
 
+    // Retrieve group list for the filterUser if cell key is a group.
+    // Group list is not required when filterUser itself a group
+    List<String> filterUserGroups = null;
+    if (filterPerms) {
+      if (username.charAt(0) == '@' && !StringUtils.isEmpty(filterUser)
+          && filterUser.charAt(0) != '@') {
+        filterUserGroups = AccessChecker.getUserGroups(filterUser);
+      }
+    }
+
     // Handle namespace entry
     if (isNamespaceEntry(entryName)) {
-      return new Pair<>(username, new TablePermission(Bytes.toString(fromNamespaceEntry(entryName)), value));
+      // Filter the permissions cell record if client query
+      if (filterPerms && !validateFilterUser(username, filterUser, filterUserGroups)) {
+        return null;
+      }
+
+      return new Pair<>(username,
+          new TablePermission(Bytes.toString(fromNamespaceEntry(entryName)), value));
     }
 
     //Handle table and global entry
@@ -635,14 +672,71 @@ public class AccessControlLists {
       }
     }
 
-    return new Pair<>(username, new TablePermission(TableName.valueOf(entryName), permFamily, permQualifier, value));
+    // Filter the permissions cell record if client query
+    if (filterPerms) {
+      // ACL table contain 3 types of cell key entries; hbase:Acl, namespace and table. So to filter
+      // the permission cell records additional validations are required at CF, CQ and username.
+      // Here we can proceed based on client input whether it contain filterUser.
+      // Validate the filterUser when specified
+      if (filterUser != null && !validateFilterUser(username, filterUser, filterUserGroups)) {
+        return null;
+      }
+      if (!validateCFAndCQ(permFamily, cf, permQualifier, cq)) {
+        return null;
+      }
+    }
+
+    return new Pair<>(username,
+        new TablePermission(TableName.valueOf(entryName), permFamily, permQualifier, value));
+  }
+
+  /*
+   * Validate the cell key with the client filterUser if specified in the query input. 1. If cell
+   * key (username) is not a group then check whether client filterUser is equal to username 2. If
+   * cell key (username) is a group then check whether client filterUser belongs to the cell key
+   * group (username) 3. In case when both filterUser and username are group names then cell will be
+   * filtered if not equal.
+   */
+  private static boolean validateFilterUser(String username, String filterUser,
+      List<String> filterUserGroups) {
+    if (filterUserGroups == null) {
+      // Validate user name or group names whether equal
+      if (filterUser.equals(username)) {
+        return true;
+      }
+    } else {
+      // Check whether filter user belongs to the cell key group.
+      return filterUserGroups.contains(username.substring(1));
+    }
+    return false;
+  }
+
+  /*
+   * Validate the cell with client CF and CQ if specified in the query input. 1. If CF is NULL, then
+   * no need of further validation, result should include all CF and CQ. 2. IF CF specified and
+   * equal then validation required at CQ level if CF specified in client input, otherwise return
+   * all CQ records.
+   */
+  private static boolean validateCFAndCQ(byte[] permFamily, byte[] cf, byte[] permQualifier,
+      byte[] cq) {
+    boolean include = true;
+    if (cf != null) {
+      if (Bytes.equals(cf, permFamily)) {
+        if (cq != null && !Bytes.equals(cq, permQualifier)) {
+          // if CQ specified and didn't match then ignore this cell
+          include = false;
+        }
+      } else {
+        // if CF specified and didn't match then ignore this cell
+        include = false;
+      }
+    }
+    return include;
   }
 
   /**
-   * Writes a set of permissions as {@link org.apache.hadoop.io.Writable} instances
+   * Writes a set of permissions as {@link org.apache.hadoop.io.Writable} instances and returns the
-   * and returns the resulting byte array.
+   * resulting byte array. Writes a set of permission [user: table permission]
-   *
-   * Writes a set of permission [user: table permission]
    */
   public static byte[] writePermissionsAsBytes(ListMultimap<String, TablePermission> perms,
       Configuration conf) {

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

@@ -36,6 +36,7 @@ import java.util.Optional;
 import java.util.Set;
 import java.util.TreeMap;
 import java.util.TreeSet;
+
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.ArrayBackedTag;
 import org.apache.hadoop.hbase.Cell;
@@ -98,6 +99,8 @@ import org.apache.hadoop.hbase.ipc.RpcServer;
 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.AccessControlProtos.HasPermissionRequest;
+import org.apache.hadoop.hbase.protobuf.generated.AccessControlProtos.HasPermissionResponse;
 import org.apache.hadoop.hbase.quotas.GlobalQuotaSettings;
 import org.apache.hadoop.hbase.regionserver.BloomType;
 import org.apache.hadoop.hbase.regionserver.FlushLifeCycleTracker;
@@ -119,6 +122,7 @@ import org.apache.hadoop.hbase.security.User;
 import org.apache.hadoop.hbase.security.UserProvider;
 import org.apache.hadoop.hbase.security.access.Permission.Action;
 import org.apache.hadoop.hbase.snapshot.SnapshotDescriptionUtils;
+import org.apache.hadoop.hbase.shaded.protobuf.ResponseConverter;
 import org.apache.hadoop.hbase.util.ByteRange;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
@@ -240,8 +244,7 @@ public class AccessController implements MasterCoprocessor, RegionCoprocessor,
   private void initialize(RegionCoprocessorEnvironment e) throws IOException {
     final Region region = e.getRegion();
     Configuration conf = e.getConfiguration();
-    Map<byte[], ListMultimap<String,TablePermission>> tables =
+    Map<byte[], ListMultimap<String, TablePermission>> tables = AccessControlLists.loadAll(region);
-        AccessControlLists.loadAll(region);
     // For each table, write out the table's permissions to the respective
     // znode for that table.
     for (Map.Entry<byte[], ListMultimap<String,TablePermission>> t:
@@ -284,7 +287,7 @@ public class AccessController implements MasterCoprocessor, RegionCoprocessor,
       for (byte[] entry : entries) {
         currentEntry = entry;
         ListMultimap<String, TablePermission> perms =
-            AccessControlLists.getPermissions(conf, entry, t);
+            AccessControlLists.getPermissions(conf, entry, t, null, null, null, false);
         byte[] serialized = AccessControlLists.writePermissionsAsBytes(perms, conf);
         zkw.writeToZookeeper(entry, serialized);
       }
@@ -295,31 +298,29 @@ public class AccessController implements MasterCoprocessor, RegionCoprocessor,
   }
 
   /**
-   * Check the current user for authorization to perform a specific action
+   * Check the current user for authorization to perform a specific action against the given set of
-   * against the given set of row data.
+   * row data.
-   *
+   * <p>
-   * <p>Note: Ordering of the authorization checks
+   * Note: Ordering of the authorization checks has been carefully optimized to short-circuit the
-   * has been carefully optimized to short-circuit the most common requests
+   * most common requests and minimize the amount of processing required.
-   * and minimize the amount of processing required.</p>
+   * </p>
-   *
+   * @param request User request
+   * @param user User name
    * @param permRequest the action being requested
    * @param e the coprocessor environment
-   * @param families the map of column families to qualifiers present in
+   * @param tableName Table name
-   * the request
+   * @param families the map of column families to qualifiers present in the request
    * @return an authorization result
    */
   private AuthResult permissionGranted(String request, User user, Action permRequest,
-      RegionCoprocessorEnvironment e,
+      RegionCoprocessorEnvironment e, TableName tableName,
       Map<byte[], ? extends Collection<?>> families) {
-    RegionInfo hri = e.getRegion().getRegionInfo();
-    TableName tableName = hri.getTable();
-
     // 1. All users need read access to hbase:meta table.
     // this is a very common operation, so deal with it quickly.
-    if (hri.isMetaRegion()) {
+    if (TableName.META_TABLE_NAME.equals(tableName)) {
       if (permRequest == Action.READ) {
-        return AuthResult.allow(request, "All users allowed", user,
+        return AuthResult.allow(request, "All users allowed", user, permRequest, tableName,
-          permRequest, tableName, families);
+          families);
       }
     }
 
@@ -398,7 +399,8 @@ public class AccessController implements MasterCoprocessor, RegionCoprocessor,
       Map<byte [], ? extends Collection<?>> families, Action... actions) {
     AuthResult result = null;
     for (Action action: actions) {
-      result = permissionGranted(opType.toString(), user, action, e, families);
+      result = permissionGranted(opType.toString(), user, action, e,
+        e.getRegion().getRegionInfo().getTable(), families);
       if (!result.isAllowed()) {
         return result;
       }
@@ -413,14 +415,14 @@ public class AccessController implements MasterCoprocessor, RegionCoprocessor,
 
   public void requirePermission(ObserverContext<?> ctx, String request,
       Action perm) throws IOException {
-    accessChecker.requirePermission(getActiveUser(ctx), request, perm);
+    accessChecker.requirePermission(getActiveUser(ctx), request, null, perm);
   }
 
   public void requireGlobalPermission(ObserverContext<?> ctx, String request,
       Action perm, TableName tableName,
       Map<byte[], ? extends Collection<byte[]>> familyMap) throws IOException {
-    accessChecker.requireGlobalPermission(getActiveUser(ctx),
+    accessChecker.requireGlobalPermission(getActiveUser(ctx), request, perm, tableName, familyMap,
-        request, perm,tableName, familyMap);
+      null);
   }
 
   public void requireGlobalPermission(ObserverContext<?> ctx, String request,
@@ -432,7 +434,7 @@ public class AccessController implements MasterCoprocessor, RegionCoprocessor,
   public void requireNamespacePermission(ObserverContext<?> ctx, String request, String namespace,
       Action... permissions) throws IOException {
     accessChecker.requireNamespacePermission(getActiveUser(ctx),
-        request, namespace, permissions);
+        request, namespace, null, permissions);
   }
 
   public void requireNamespacePermission(ObserverContext<?> ctx, String request, String namespace,
@@ -446,7 +448,7 @@ public class AccessController implements MasterCoprocessor, RegionCoprocessor,
   public void requirePermission(ObserverContext<?> ctx, String request, TableName tableName,
       byte[] family, byte[] qualifier, Action... permissions) throws IOException {
     accessChecker.requirePermission(getActiveUser(ctx), request,
-        tableName, family, qualifier, permissions);
+        tableName, family, qualifier, null, permissions);
   }
 
   public void requireTablePermission(ObserverContext<?> ctx, String request,
@@ -936,7 +938,8 @@ public class AccessController implements MasterCoprocessor, RegionCoprocessor,
     User.runAsLoginUser(new PrivilegedExceptionAction<Void>() {
       @Override
       public Void run() throws Exception {
-        List<UserPermission> acls = AccessControlLists.getUserTablePermissions(conf, tableName);
+        List<UserPermission> acls =
+            AccessControlLists.getUserTablePermissions(conf, tableName, null, null, null, false);
         if (acls != null) {
           tableAcls.put(tableName, acls);
         }
@@ -1040,7 +1043,7 @@ public class AccessController implements MasterCoprocessor, RegionCoprocessor,
   public void preGetLocks(ObserverContext<MasterCoprocessorEnvironment> ctx)
       throws IOException {
     User user = getActiveUser(ctx);
-    accessChecker.requirePermission(user, "getLocks", Action.ADMIN);
+    accessChecker.requirePermission(user, "getLocks", null, Action.ADMIN);
   }
 
   @Override
@@ -1152,7 +1155,8 @@ public class AccessController implements MasterCoprocessor, RegionCoprocessor,
           "Snapshot owner check allowed", user, null, null, null);
       AccessChecker.logResult(result);
     } else {
-      accessChecker.requirePermission(user, "listSnapshot " + snapshot.getName(), Action.ADMIN);
+      accessChecker.requirePermission(user, "listSnapshot " + snapshot.getName(), null,
+        Action.ADMIN);
     }
   }
 
@@ -1168,7 +1172,8 @@ public class AccessController implements MasterCoprocessor, RegionCoprocessor,
         "Snapshot owner check allowed", user, null, hTableDescriptor.getTableName(), null);
       AccessChecker.logResult(result);
     } else {
-      accessChecker.requirePermission(user, "cloneSnapshot " + snapshot.getName(), Action.ADMIN);
+      accessChecker.requirePermission(user, "cloneSnapshot " + snapshot.getName(), null,
+        Action.ADMIN);
     }
   }
 
@@ -1179,9 +1184,10 @@ public class AccessController implements MasterCoprocessor, RegionCoprocessor,
     User user = getActiveUser(ctx);
     if (SnapshotDescriptionUtils.isSnapshotOwner(snapshot, user)) {
       accessChecker.requirePermission(user, "restoreSnapshot " + snapshot.getName(),
-          hTableDescriptor.getTableName(), null, null, Permission.Action.ADMIN);
+        hTableDescriptor.getTableName(), null, null, null, Permission.Action.ADMIN);
     } else {
-      accessChecker.requirePermission(user, "restoreSnapshot " + snapshot.getName(), Action.ADMIN);
+      accessChecker.requirePermission(user, "restoreSnapshot " + snapshot.getName(), null,
+        Action.ADMIN);
     }
   }
 
@@ -1195,7 +1201,8 @@ public class AccessController implements MasterCoprocessor, RegionCoprocessor,
           "Snapshot owner check allowed", user, null, null, null);
       AccessChecker.logResult(result);
     } else {
-      accessChecker.requirePermission(user, "deleteSnapshot " + snapshot.getName(), Action.ADMIN);
+      accessChecker.requirePermission(user, "deleteSnapshot " + snapshot.getName(), null,
+        Action.ADMIN);
     }
   }
 
@@ -1255,8 +1262,8 @@ public class AccessController implements MasterCoprocessor, RegionCoprocessor,
     while (itr.hasNext()) {
       NamespaceDescriptor desc = itr.next();
       try {
-        accessChecker.requireNamespacePermission(user, "listNamespaces",
+        accessChecker.requireNamespacePermission(user, "listNamespaces", desc.getName(), null,
-            desc.getName(), Action.ADMIN);
+          Action.ADMIN);
       } catch (AccessDeniedException e) {
         itr.remove();
       }
@@ -1971,10 +1978,8 @@ public class AccessController implements MasterCoprocessor, RegionCoprocessor,
     User user = getActiveUser(ctx);
     for(Pair<byte[],String> el : familyPaths) {
       accessChecker.requirePermission(user, "preBulkLoadHFile",
-          ctx.getEnvironment().getRegion().getTableDescriptor().getTableName(),
+        ctx.getEnvironment().getRegion().getTableDescriptor().getTableName(), el.getFirst(), null,
-          el.getFirst(),
+        null, Action.CREATE);
-          null,
-          Action.CREATE);
     }
   }
 
@@ -2048,11 +2053,11 @@ public class AccessController implements MasterCoprocessor, RegionCoprocessor,
           case Global :
           case Table :
             accessChecker.requirePermission(caller, "grant", perm.getTableName(),
-                perm.getFamily(), perm.getQualifier(), Action.ADMIN);
+                perm.getFamily(), perm.getQualifier(), null, Action.ADMIN);
             break;
           case Namespace :
             accessChecker.requireNamespacePermission(caller, "grant", perm.getNamespace(),
-                Action.ADMIN);
+                null, Action.ADMIN);
            break;
         }
 
@@ -2106,11 +2111,11 @@ public class AccessController implements MasterCoprocessor, RegionCoprocessor,
           case Global :
           case Table :
             accessChecker.requirePermission(caller, "revoke", perm.getTableName(), perm.getFamily(),
-              perm.getQualifier(), Action.ADMIN);
+              perm.getQualifier(), null, Action.ADMIN);
             break;
           case Namespace :
             accessChecker.requireNamespacePermission(caller, "revoke", perm.getNamespace(),
-                Action.ADMIN);
+                null, Action.ADMIN);
             break;
         }
 
@@ -2156,44 +2161,75 @@ public class AccessController implements MasterCoprocessor, RegionCoprocessor,
         User caller = RpcServer.getRequestUser().orElse(null);
 
         List<UserPermission> perms = null;
+        // Initialize username, cf and cq. Set to null if request doesn't have.
+        final String userName = request.hasUserName() ? request.getUserName().toStringUtf8() : null;
+        final byte[] cf =
+            request.hasColumnFamily() ? request.getColumnFamily().toByteArray() : null;
+        final byte[] cq =
+            request.hasColumnQualifier() ? request.getColumnQualifier().toByteArray() : null;
+
         if (request.getType() == AccessControlProtos.Permission.Type.Table) {
           final TableName table = request.hasTableName() ?
             ProtobufUtil.toTableName(request.getTableName()) : null;
-          accessChecker.requirePermission(caller, "userPermissions",
+          accessChecker.requirePermission(caller, "userPermissions", table, cf, cq, userName,
-              table, null, null, Action.ADMIN);
+            Action.ADMIN);
           perms = User.runAsLoginUser(new PrivilegedExceptionAction<List<UserPermission>>() {
             @Override
             public List<UserPermission> run() throws Exception {
-              return AccessControlLists.getUserTablePermissions(regionEnv.getConfiguration(), table);
+              if (cf != null || userName != null) {
+                // retrieve permission based on the requested parameters
+                return AccessControlLists.getUserTablePermissions(regionEnv.getConfiguration(),
+                  table, cf, cq, userName, true);
+              } else {
+                return AccessControlLists.getUserTablePermissions(regionEnv.getConfiguration(),
+                  table, null, null, null, false);
+              }
             }
           });
         } else if (request.getType() == AccessControlProtos.Permission.Type.Namespace) {
           final String namespace = request.getNamespaceName().toStringUtf8();
           accessChecker.requireNamespacePermission(caller, "userPermissions",
-              namespace, Action.ADMIN);
+            namespace,userName, Action.ADMIN);
           perms = User.runAsLoginUser(new PrivilegedExceptionAction<List<UserPermission>>() {
             @Override
             public List<UserPermission> run() throws Exception {
+              if (userName != null) {
+                // retrieve permission based on the requested parameters
                 return AccessControlLists.getUserNamespacePermissions(regionEnv.getConfiguration(),
-                namespace);
+                  namespace, userName, true);
+              } else {
+                return AccessControlLists.getUserNamespacePermissions(regionEnv.getConfiguration(),
+                  namespace, null, false);
+              }
             }
           });
         } else {
-          accessChecker.requirePermission(caller, "userPermissions", Action.ADMIN);
+          accessChecker.requirePermission(caller, "userPermissions", userName, Action.ADMIN);
           perms = User.runAsLoginUser(new PrivilegedExceptionAction<List<UserPermission>>() {
             @Override
             public List<UserPermission> run() throws Exception {
-              return AccessControlLists.getUserPermissions(regionEnv.getConfiguration(), null);
+              if (userName != null) {
+                // retrieve permission based on the requested parameters
+                return AccessControlLists.getUserPermissions(regionEnv.getConfiguration(), null,
+                  null, null, userName, true);
+              } else {
+                return AccessControlLists.getUserPermissions(regionEnv.getConfiguration(), null,
+                  null, null, null, false);
+              }
             }
           });
-          // Adding superusers explicitly to the result set as AccessControlLists do not store them.
+
-          // Also using acl as table name to be inline  with the results of global admin and will
+          // Skip super users when filter user is specified
-          // help in avoiding any leakage of information about being superusers.
+          if (userName == null) {
+            // Adding superusers explicitly to the result set as AccessControlLists do not store
+            // them. Also using acl as table name to be inline with the results of global admin and
+            // will help in avoiding any leakage of information about being superusers.
             for (String user : Superusers.getSuperUsers()) {
               perms.add(new UserPermission(Bytes.toBytes(user), AccessControlLists.ACL_TABLE_NAME,
                   null, Action.values()));
             }
           }
+        }
         response = AccessControlUtil.buildGetUserPermissionsResponse(perms);
       } else {
         throw new CoprocessorException(AccessController.class, "This method "
@@ -2243,7 +2279,7 @@ public class AccessController implements MasterCoprocessor, RegionCoprocessor,
             }
 
             AuthResult result = permissionGranted("checkPermissions", user, action, regionEnv,
-              familyMap);
+              regionEnv.getRegion().getRegionInfo().getTable(), familyMap);
             AccessChecker.logResult(result);
             if (!result.isAllowed()) {
               // Even if passive we need to throw an exception here, we support checking
@@ -2550,4 +2586,51 @@ public class AccessController implements MasterCoprocessor, RegionCoprocessor,
     }
     return userProvider.getCurrent();
   }
+
+  @Override
+  public void hasPermission(RpcController controller, HasPermissionRequest request,
+      RpcCallback<HasPermissionResponse> done) {
+    // Converts proto to a TablePermission object.
+    TablePermission tPerm = AccessControlUtil.toTablePermission(request.getTablePermission());
+    // Check input user name
+    if (!request.hasUserName()) {
+      throw new IllegalStateException("Input username cannot be empty");
+    }
+    final String inputUserName = request.getUserName().toStringUtf8();
+    AccessControlProtos.HasPermissionResponse response = null;
+    try {
+      User caller = RpcServer.getRequestUser().orElse(null);
+      // User instance for the input user name
+      User filterUser = accessChecker.validateCallerWithFilterUser(caller, tPerm, inputUserName);
+
+      // Initialize family and qualifier map
+      Map<byte[], Set<byte[]>> familyMap = new TreeMap<byte[], Set<byte[]>>(Bytes.BYTES_COMPARATOR);
+      if (tPerm.getFamily() != null) {
+        if (tPerm.getQualifier() != null) {
+          Set<byte[]> qualifiers = Sets.newTreeSet(Bytes.BYTES_COMPARATOR);
+          qualifiers.add(tPerm.getQualifier());
+          familyMap.put(tPerm.getFamily(), qualifiers);
+        } else {
+          familyMap.put(tPerm.getFamily(), null);
+        }
+      }
+
+      // Iterate each action and check whether permission granted
+      boolean hasPermission = false;
+      for (Action action : tPerm.getActions()) {
+        AuthResult result = permissionGranted("hasPermission", filterUser, action, regionEnv,
+          tPerm.getTableName(), familyMap);
+        if (!result.isAllowed()) {
+          hasPermission = false;
+          // Break the loop is any action is not allowed
+          break;
+        }
+        hasPermission = true;
+      }
+      response = ResponseConverter.buildHasPermissionResponse(hasPermission);
+    } catch (IOException ioe) {
+      ResponseConverter.setControllerException(controller, ioe);
+    }
+    done.run(response);
+  }
 }

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

@@ -19,7 +19,9 @@
 package org.apache.hadoop.hbase.security.access;
 
 import java.util.Collection;
+import java.util.HashMap;
 import java.util.Map;
+import java.util.Map.Entry;
 
 import org.apache.yetus.audience.InterfaceAudience;
 import org.apache.hadoop.hbase.Cell;
@@ -254,6 +256,13 @@ public class AuthResult {
     private Map<byte[], ? extends Collection<?>> families = null;
     byte[] family = null;
     byte[] qualifier = null;
+    // For extra parameters to be shown in audit log
+    private final Map<String, String> extraParams = new HashMap<String, String>(2);
+
+    public Params addExtraParam(String key, String value) {
+      extraParams.put(key, value);
+      return this;
+    }
 
     public Params setNamespace(String namespace) {
       this.namespace = namespace;
@@ -286,10 +295,29 @@ public class AuthResult {
       String[] params = new String[] {
           namespace != null ? "namespace=" + namespace : null,
           tableName != null ? "table=" + tableName.getNameWithNamespaceInclAsString() : null,
-          familiesString.length() > 0 ? "family=" + familiesString : null
+          familiesString.length() > 0 ? "family=" + familiesString : null,
+          extraParams.isEmpty() ? null : concatenateExtraParams()
       };
       return Joiner.on(",").skipNulls().join(params);
     }
 
+    /**
+     * @return extra parameter key/value string
+     */
+    private String concatenateExtraParams() {
+      final StringBuilder sb = new StringBuilder();
+      boolean first = true;
+      for (Entry<String, String> entry : extraParams.entrySet()) {
+        if (entry.getKey() != null && entry.getValue() != null) {
+          if (!first) {
+            sb.append(',');
+          }
+          first = false;
+          sb.append(entry.getKey() + '=');
+          sb.append(entry.getValue());
+        }
+      }
+      return sb.toString();
+    }
   }
 }

hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestMasterCoprocessorServices.java

@@ -41,6 +41,8 @@ import org.apache.hadoop.hbase.protobuf.generated.AccessControlProtos.GetUserPer
 import org.apache.hadoop.hbase.protobuf.generated.AccessControlProtos.GetUserPermissionsResponse;
 import org.apache.hadoop.hbase.protobuf.generated.AccessControlProtos.GrantRequest;
 import org.apache.hadoop.hbase.protobuf.generated.AccessControlProtos.GrantResponse;
+import org.apache.hadoop.hbase.protobuf.generated.AccessControlProtos.HasPermissionRequest;
+import org.apache.hadoop.hbase.protobuf.generated.AccessControlProtos.HasPermissionResponse;
 import org.apache.hadoop.hbase.protobuf.generated.AccessControlProtos.RevokeRequest;
 import org.apache.hadoop.hbase.protobuf.generated.AccessControlProtos.RevokeResponse;
 import org.apache.hadoop.hbase.protobuf.generated.VisibilityLabelsProtos.GetAuthsRequest;
@@ -88,6 +90,11 @@ public class TestMasterCoprocessorServices {
     @Override
     public void checkPermissions(RpcController controller, CheckPermissionsRequest request,
         RpcCallback<CheckPermissionsResponse> done) {}
+
+    @Override
+    public void hasPermission(RpcController controller, HasPermissionRequest request,
+        RpcCallback<HasPermissionResponse> done) {
+    }
   }
 
   private static class MockVisibilityController implements VisibilityLabelsService.Interface,

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

@@ -38,6 +38,7 @@ import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.CommonConfigurationKeys;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
@@ -127,6 +128,9 @@ import org.apache.hadoop.hbase.tool.LoadIncrementalHFiles;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.hbase.util.JVMClusterUtil;
 import org.apache.hadoop.hbase.util.Threads;
+import org.apache.hadoop.security.GroupMappingServiceProvider;
+import org.apache.hadoop.security.ShellBasedUnixGroupsMapping;
+import org.apache.hadoop.security.UserGroupInformation;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
 import org.junit.ClassRule;
@@ -212,6 +216,11 @@ public class TestAccessController extends SecureTestUtil {
     conf = TEST_UTIL.getConfiguration();
     // Up the handlers; this test needs more than usual.
     conf.setInt(HConstants.REGION_SERVER_HIGH_PRIORITY_HANDLER_COUNT, 10);
+
+    conf.set(CommonConfigurationKeys.HADOOP_SECURITY_GROUP_MAPPING,
+      MyShellBasedUnixGroupsMapping.class.getName());
+    UserGroupInformation.setConfiguration(conf);
+
     // Enable security
     enableSecurity(conf);
     // In this particular test case, we can't use SecureBulkLoadEndpoint because its doAs will fail
@@ -2215,8 +2224,12 @@ public class TestAccessController extends SecureTestUtil {
   }
 
   private void createTestTable(TableName tname) throws Exception {
+    createTestTable(tname, TEST_FAMILY);
+  }
+
+  private void createTestTable(TableName tname, byte[] cf) throws Exception {
     HTableDescriptor htd = new HTableDescriptor(tname);
-    HColumnDescriptor hcd = new HColumnDescriptor(TEST_FAMILY);
+    HColumnDescriptor hcd = new HColumnDescriptor(cf);
     hcd.setMaxVersions(100);
     htd.addFamily(hcd);
     htd.setOwner(USER_OWNER);
@@ -3105,4 +3118,470 @@ public class TestAccessController extends SecureTestUtil {
     verifyAllowed(action, SUPERUSER);
     verifyDenied(action, USER_CREATE, USER_RW, USER_RO, USER_NONE, USER_OWNER, USER_ADMIN);
   }
+
+  @Test(timeout = 180000)
+  public void testGetUserPermissions() throws Throwable {
+    Connection conn = null;
+    try {
+      conn = ConnectionFactory.createConnection(conf);
+      User nSUser1 = User.createUserForTesting(conf, "nsuser1", new String[0]);
+      User nSUser2 = User.createUserForTesting(conf, "nsuser2", new String[0]);
+      User nSUser3 = User.createUserForTesting(conf, "nsuser3", new String[0]);
+
+      // Global access groups
+      User globalGroupUser1 =
+          User.createUserForTesting(conf, "globalGroupUser1", new String[] { "group_admin" });
+      User globalGroupUser2 = User.createUserForTesting(conf, "globalGroupUser2",
+        new String[] { "group_admin", "group_create" });
+      // Namespace access groups
+      User nsGroupUser1 =
+          User.createUserForTesting(conf, "nsGroupUser1", new String[] { "ns_group1" });
+      User nsGroupUser2 =
+          User.createUserForTesting(conf, "nsGroupUser2", new String[] { "ns_group2" });
+      // table Access groups
+      User tableGroupUser1 =
+          User.createUserForTesting(conf, "tableGroupUser1", new String[] { "table_group1" });
+      User tableGroupUser2 =
+          User.createUserForTesting(conf, "tableGroupUser2", new String[] { "table_group2" });
+
+      // Create namespaces
+      String nsPrefix = "testNS";
+      final String namespace1 = nsPrefix + "1";
+      NamespaceDescriptor desc1 = NamespaceDescriptor.create(namespace1).build();
+      createNamespace(TEST_UTIL, desc1);
+      String namespace2 = nsPrefix + "2";
+      NamespaceDescriptor desc2 = NamespaceDescriptor.create(namespace2).build();
+      createNamespace(TEST_UTIL, desc2);
+
+      // Grant namespace permission
+      grantOnNamespace(TEST_UTIL, nSUser1.getShortName(), namespace1, Permission.Action.ADMIN);
+      grantOnNamespace(TEST_UTIL, nSUser3.getShortName(), namespace1, Permission.Action.READ);
+      grantOnNamespace(TEST_UTIL, toGroupEntry("ns_group1"), namespace1, Permission.Action.ADMIN);
+      grantOnNamespace(TEST_UTIL, nSUser2.getShortName(), namespace2, Permission.Action.ADMIN);
+      grantOnNamespace(TEST_UTIL, nSUser3.getShortName(), namespace2, Permission.Action.ADMIN);
+      grantOnNamespace(TEST_UTIL, toGroupEntry("ns_group2"), namespace2, Permission.Action.READ,
+        Permission.Action.WRITE);
+
+      // Create tables
+      TableName table1 = TableName.valueOf(namespace1 + TableName.NAMESPACE_DELIM + "t1");
+      TableName table2 = TableName.valueOf(namespace2 + TableName.NAMESPACE_DELIM + "t2");
+      byte[] TEST_FAMILY2 = Bytes.toBytes("f2");
+      byte[] TEST_QUALIFIER2 = Bytes.toBytes("q2");
+      createTestTable(table1, TEST_FAMILY);
+      createTestTable(table2, TEST_FAMILY2);
+
+      // Grant table permissions
+      grantOnTable(TEST_UTIL, toGroupEntry("table_group1"), table1, null, null,
+        Permission.Action.ADMIN);
+      grantOnTable(TEST_UTIL, USER_ADMIN.getShortName(), table1, null, null,
+        Permission.Action.ADMIN);
+      grantOnTable(TEST_UTIL, USER_ADMIN_CF.getShortName(), table1, TEST_FAMILY, null,
+        Permission.Action.ADMIN);
+      grantOnTable(TEST_UTIL, USER_RW.getShortName(), table1, TEST_FAMILY, TEST_QUALIFIER,
+        Permission.Action.READ);
+      grantOnTable(TEST_UTIL, USER_RW.getShortName(), table1, TEST_FAMILY, TEST_QUALIFIER2,
+        Permission.Action.WRITE);
+
+      grantOnTable(TEST_UTIL, toGroupEntry("table_group2"), table2, null, null,
+        Permission.Action.ADMIN);
+      grantOnTable(TEST_UTIL, USER_ADMIN.getShortName(), table2, null, null,
+        Permission.Action.ADMIN);
+      grantOnTable(TEST_UTIL, USER_ADMIN_CF.getShortName(), table2, TEST_FAMILY2, null,
+        Permission.Action.ADMIN);
+      grantOnTable(TEST_UTIL, USER_RW.getShortName(), table2, TEST_FAMILY2, TEST_QUALIFIER,
+        Permission.Action.READ);
+      grantOnTable(TEST_UTIL, USER_RW.getShortName(), table2, TEST_FAMILY2, TEST_QUALIFIER2,
+        Permission.Action.WRITE);
+
+      List<UserPermission> userPermissions = null;
+      Collection<String> superUsers = Superusers.getSuperUsers();
+      int superUserCount = superUsers.size();
+
+      // Global User ACL
+      validateGlobalUserACLForGetUserPermissions(conn, nSUser1, globalGroupUser1, globalGroupUser2,
+        superUsers, superUserCount);
+
+      // Namespace ACL
+      validateNamespaceUserACLForGetUserPermissions(conn, nSUser1, nSUser3, nsGroupUser1,
+        nsGroupUser2, nsPrefix, namespace1, namespace2);
+
+      // Table + Users
+      validateTableACLForGetUserPermissions(conn, nSUser1, tableGroupUser1, tableGroupUser2,
+        nsPrefix, table1, table2, TEST_QUALIFIER2, superUsers);
+
+      // exception scenarios
+
+      try {
+        // test case with table name as null
+        assertEquals(3, AccessControlClient.getUserPermissions(conn, null, TEST_FAMILY).size());
+        fail("this should have thrown IllegalArgumentException");
+      } catch (IllegalArgumentException ex) {
+        // expected
+      }
+      try {
+        // test case with table name as emplty
+        assertEquals(3, AccessControlClient
+            .getUserPermissions(conn, HConstants.EMPTY_STRING, TEST_FAMILY).size());
+        fail("this should have thrown IllegalArgumentException");
+      } catch (IllegalArgumentException ex) {
+        // expected
+      }
+      try {
+        // test case with table name as namespace name
+        assertEquals(3,
+          AccessControlClient.getUserPermissions(conn, "@" + namespace2, TEST_FAMILY).size());
+        fail("this should have thrown IllegalArgumentException");
+      } catch (IllegalArgumentException ex) {
+        // expected
+      }
+
+      // Clean the table and namespace
+      deleteTable(TEST_UTIL, table1);
+      deleteTable(TEST_UTIL, table2);
+      deleteNamespace(TEST_UTIL, namespace1);
+      deleteNamespace(TEST_UTIL, namespace2);
+    } finally {
+      if (conn != null) {
+        conn.close();
+      }
+    }
+  }
+
+  @Test(timeout = 180000)
+  public void testHasPermission() throws Throwable {
+    Connection conn = null;
+    try {
+      conn = ConnectionFactory.createConnection(conf);
+      // Create user and set namespace ACL
+      User user1 = User.createUserForTesting(conf, "testHasPermissionUser1", new String[0]);
+      // Grant namespace permission
+      grantOnNamespaceUsingAccessControlClient(TEST_UTIL, conn, user1.getShortName(),
+        NamespaceDescriptor.DEFAULT_NAMESPACE.getName(), Permission.Action.ADMIN,
+        Permission.Action.CREATE, Permission.Action.READ);
+
+      // Create user and set table ACL
+      User user2 = User.createUserForTesting(conf, "testHasPermissionUser2", new String[0]);
+      // Grant namespace permission
+      grantOnTableUsingAccessControlClient(TEST_UTIL, conn, user2.getShortName(), TEST_TABLE,
+        TEST_FAMILY, TEST_QUALIFIER, Permission.Action.READ, Permission.Action.WRITE);
+
+      // Verify action privilege
+      AccessTestAction hasPermissionAction = new AccessTestAction() {
+        @Override
+        public Object run() throws Exception {
+          try (Connection conn = ConnectionFactory.createConnection(conf);
+              Table acl = conn.getTable(AccessControlLists.ACL_TABLE_NAME)) {
+            BlockingRpcChannel service = acl.coprocessorService(TEST_TABLE.getName());
+            AccessControlService.BlockingInterface protocol =
+                AccessControlService.newBlockingStub(service);
+            Permission.Action[] actions = { Permission.Action.READ, Permission.Action.WRITE };
+            AccessControlUtil.hasPermission(null, protocol, TEST_TABLE, TEST_FAMILY,
+              HConstants.EMPTY_BYTE_ARRAY, "dummy", actions);
+          }
+          return null;
+        }
+      };
+      verifyAllowed(hasPermissionAction, SUPERUSER, USER_ADMIN, USER_GROUP_ADMIN, USER_OWNER,
+        USER_ADMIN_CF, user1);
+      verifyDenied(hasPermissionAction, USER_CREATE, USER_RW, USER_RO, USER_NONE, user2);
+
+      // Check for global user
+      assertTrue(AccessControlClient.hasPermission(conn, TEST_TABLE.getNameAsString(),
+        HConstants.EMPTY_BYTE_ARRAY, HConstants.EMPTY_BYTE_ARRAY, USER_ADMIN.getShortName(),
+        Permission.Action.READ, Permission.Action.WRITE, Permission.Action.CREATE,
+        Permission.Action.ADMIN));
+      assertFalse(AccessControlClient.hasPermission(conn, TEST_TABLE.getNameAsString(),
+        HConstants.EMPTY_BYTE_ARRAY, HConstants.EMPTY_BYTE_ARRAY, USER_ADMIN.getShortName(),
+        Permission.Action.READ, Permission.Action.WRITE, Permission.Action.CREATE,
+        Permission.Action.ADMIN, Permission.Action.EXEC));
+
+      // Check for namespace access user
+      assertTrue(AccessControlClient.hasPermission(conn, TEST_TABLE.getNameAsString(),
+        HConstants.EMPTY_BYTE_ARRAY, HConstants.EMPTY_BYTE_ARRAY, user1.getShortName(),
+        Permission.Action.ADMIN, Permission.Action.CREATE));
+      assertFalse(AccessControlClient.hasPermission(conn, TEST_TABLE.getNameAsString(),
+        HConstants.EMPTY_BYTE_ARRAY, HConstants.EMPTY_BYTE_ARRAY, user1.getShortName(),
+        Permission.Action.ADMIN, Permission.Action.READ, Permission.Action.EXEC));
+
+      // Check for table owner
+      assertTrue(AccessControlClient.hasPermission(conn, TEST_TABLE.getNameAsString(),
+        HConstants.EMPTY_BYTE_ARRAY, HConstants.EMPTY_BYTE_ARRAY, USER_OWNER.getShortName(),
+        Permission.Action.READ, Permission.Action.WRITE, Permission.Action.EXEC,
+        Permission.Action.CREATE, Permission.Action.ADMIN));
+
+      // Check for table user
+      assertTrue(AccessControlClient.hasPermission(conn, TEST_TABLE.getNameAsString(),
+        HConstants.EMPTY_BYTE_ARRAY, HConstants.EMPTY_BYTE_ARRAY, USER_CREATE.getShortName(),
+        Permission.Action.READ, Permission.Action.WRITE));
+      assertFalse(AccessControlClient.hasPermission(conn, TEST_TABLE.getNameAsString(),
+        HConstants.EMPTY_BYTE_ARRAY, HConstants.EMPTY_BYTE_ARRAY, USER_RO.getShortName(),
+        Permission.Action.READ, Permission.Action.WRITE));
+
+      // Check for family access user
+      assertTrue(AccessControlClient.hasPermission(conn, TEST_TABLE.getNameAsString(), TEST_FAMILY,
+        HConstants.EMPTY_BYTE_ARRAY, USER_RO.getShortName(), Permission.Action.READ));
+      assertTrue(AccessControlClient.hasPermission(conn, TEST_TABLE.getNameAsString(), TEST_FAMILY,
+        HConstants.EMPTY_BYTE_ARRAY, USER_RW.getShortName(), Permission.Action.READ,
+        Permission.Action.WRITE));
+      assertFalse(AccessControlClient.hasPermission(conn, TEST_TABLE.getNameAsString(),
+        HConstants.EMPTY_BYTE_ARRAY, HConstants.EMPTY_BYTE_ARRAY, USER_ADMIN_CF.getShortName(),
+        Permission.Action.ADMIN, Permission.Action.CREATE));
+      assertTrue(AccessControlClient.hasPermission(conn, TEST_TABLE.getNameAsString(), TEST_FAMILY,
+        HConstants.EMPTY_BYTE_ARRAY, USER_ADMIN_CF.getShortName(), Permission.Action.ADMIN,
+        Permission.Action.CREATE));
+      assertFalse(AccessControlClient.hasPermission(conn, TEST_TABLE.getNameAsString(), TEST_FAMILY,
+        HConstants.EMPTY_BYTE_ARRAY, USER_ADMIN_CF.getShortName(), Permission.Action.READ));
+
+      // Check for qualifier access user
+      assertTrue(AccessControlClient.hasPermission(conn, TEST_TABLE.getNameAsString(), TEST_FAMILY,
+        TEST_QUALIFIER, user2.getShortName(), Permission.Action.READ, Permission.Action.WRITE));
+      assertFalse(AccessControlClient.hasPermission(conn, TEST_TABLE.getNameAsString(), TEST_FAMILY,
+        TEST_QUALIFIER, user2.getShortName(), Permission.Action.EXEC, Permission.Action.READ));
+      assertFalse(AccessControlClient.hasPermission(conn, TEST_TABLE.getNameAsString(),
+        HConstants.EMPTY_BYTE_ARRAY, TEST_QUALIFIER, USER_RW.getShortName(),
+        Permission.Action.WRITE, Permission.Action.READ));
+
+      // exception scenarios
+      try {
+        // test case with table name as null
+        assertTrue(AccessControlClient.hasPermission(conn, null, HConstants.EMPTY_BYTE_ARRAY,
+          HConstants.EMPTY_BYTE_ARRAY, null, Permission.Action.READ));
+        fail("this should have thrown IllegalArgumentException");
+      } catch (IllegalArgumentException ex) {
+        // expected
+      }
+      try {
+        // test case with username as null
+        assertTrue(AccessControlClient.hasPermission(conn, TEST_TABLE.getNameAsString(),
+          HConstants.EMPTY_BYTE_ARRAY, HConstants.EMPTY_BYTE_ARRAY, null, Permission.Action.READ));
+        fail("this should have thrown IllegalArgumentException");
+      } catch (IllegalArgumentException ex) {
+        // expected
+      }
+
+      revokeFromNamespaceUsingAccessControlClient(TEST_UTIL, conn, user1.getShortName(),
+        NamespaceDescriptor.DEFAULT_NAMESPACE.getName(), Permission.Action.ADMIN,
+        Permission.Action.CREATE, Permission.Action.READ);
+      revokeFromTableUsingAccessControlClient(TEST_UTIL, conn, user2.getShortName(), TEST_TABLE,
+        TEST_FAMILY, TEST_QUALIFIER, Permission.Action.READ, Permission.Action.WRITE);
+    } finally {
+      if (conn != null) {
+        conn.close();
+      }
+    }
+  }
+
+  /*
+   * Validate Global User ACL
+   */
+  private void validateGlobalUserACLForGetUserPermissions(final Connection conn, User nSUser1,
+      User globalGroupUser1, User globalGroupUser2, Collection<String> superUsers,
+      int superUserCount) throws Throwable {
+    // Verify action privilege
+    AccessTestAction globalUserPermissionAction = new AccessTestAction() {
+      @Override
+      public Object run() throws Exception {
+        try (Connection conn = ConnectionFactory.createConnection(conf);
+            Table acl = conn.getTable(AccessControlLists.ACL_TABLE_NAME)) {
+          BlockingRpcChannel service = acl.coprocessorService(TEST_TABLE.getName());
+          AccessControlService.BlockingInterface protocol =
+              AccessControlService.newBlockingStub(service);
+          AccessControlUtil.getUserPermissions(null, protocol, "dummy");
+        }
+        return null;
+      }
+    };
+    verifyAllowed(globalUserPermissionAction, SUPERUSER, USER_ADMIN, USER_GROUP_ADMIN);
+    verifyDenied(globalUserPermissionAction, USER_GROUP_CREATE, USER_GROUP_READ, USER_GROUP_WRITE);
+
+    // Validate global user permission
+    List<UserPermission> userPermissions;
+    assertEquals(5 + superUserCount, AccessControlClient.getUserPermissions(conn, null).size());
+    assertEquals(5 + superUserCount,
+      AccessControlClient.getUserPermissions(conn, HConstants.EMPTY_STRING).size());
+    assertEquals(5 + superUserCount,
+      AccessControlClient.getUserPermissions(conn, null, HConstants.EMPTY_STRING).size());
+    userPermissions = AccessControlClient.getUserPermissions(conn, null, USER_ADMIN.getName());
+    verifyGetUserPermissionResult(userPermissions, 1, null, null, USER_ADMIN.getName(), superUsers);
+    assertEquals(0, AccessControlClient.getUserPermissions(conn, null, nSUser1.getName()).size());
+    // Global group user ACL
+    assertEquals(1,
+      AccessControlClient.getUserPermissions(conn, null, globalGroupUser1.getName()).size());
+    assertEquals(2,
+      AccessControlClient.getUserPermissions(conn, null, globalGroupUser2.getName()).size());
+  }
+
+  /*
+   * Validate Namespace User ACL
+   */
+  private void validateNamespaceUserACLForGetUserPermissions(final Connection conn, User nSUser1,
+      User nSUser3, User nsGroupUser1, User nsGroupUser2, String nsPrefix, final String namespace1,
+      String namespace2) throws Throwable {
+    AccessTestAction namespaceUserPermissionAction = new AccessTestAction() {
+      @Override
+      public Object run() throws Exception {
+        try (Connection conn = ConnectionFactory.createConnection(conf);
+            Table acl = conn.getTable(AccessControlLists.ACL_TABLE_NAME)) {
+          BlockingRpcChannel service = acl.coprocessorService(TEST_TABLE.getName());
+          AccessControlService.BlockingInterface protocol =
+              AccessControlService.newBlockingStub(service);
+          AccessControlUtil.getUserPermissions(null, protocol, Bytes.toBytes(namespace1), "dummy");
+        }
+        return null;
+      }
+    };
+    verifyAllowed(namespaceUserPermissionAction, SUPERUSER, USER_GROUP_ADMIN, USER_ADMIN, nSUser1,
+      nsGroupUser1);
+    verifyDenied(namespaceUserPermissionAction, USER_GROUP_CREATE, USER_GROUP_READ,
+      USER_GROUP_WRITE, nSUser3, nsGroupUser2);
+
+    List<UserPermission> userPermissions;
+    assertEquals(6, AccessControlClient.getUserPermissions(conn, "@" + nsPrefix + ".*").size());
+    assertEquals(3, AccessControlClient.getUserPermissions(conn, "@" + namespace1).size());
+    assertEquals(3, AccessControlClient
+        .getUserPermissions(conn, "@" + namespace1, HConstants.EMPTY_STRING).size());
+    userPermissions =
+        AccessControlClient.getUserPermissions(conn, "@" + namespace1, nSUser1.getName());
+    verifyGetUserPermissionResult(userPermissions, 1, null, null, nSUser1.getName(), null);
+    userPermissions =
+        AccessControlClient.getUserPermissions(conn, "@" + namespace1, nSUser3.getName());
+    verifyGetUserPermissionResult(userPermissions, 1, null, null, nSUser3.getName(), null);
+    assertEquals(0,
+      AccessControlClient.getUserPermissions(conn, "@" + namespace1, USER_ADMIN.getName()).size());
+    // Namespace group user ACL
+    assertEquals(1, AccessControlClient
+        .getUserPermissions(conn, "@" + namespace1, nsGroupUser1.getName()).size());
+    assertEquals(1, AccessControlClient
+        .getUserPermissions(conn, "@" + namespace2, nsGroupUser2.getName()).size());
+  }
+
+  /*
+   * Validate Table User ACL
+   */
+  private void validateTableACLForGetUserPermissions(final Connection conn, User nSUser1,
+      User tableGroupUser1, User tableGroupUser2, String nsPrefix, TableName table1,
+      TableName table2, byte[] TEST_QUALIFIER2, Collection<String> superUsers) throws Throwable {
+    AccessTestAction tableUserPermissionAction = new AccessTestAction() {
+      @Override
+      public Object run() throws Exception {
+        try (Connection conn = ConnectionFactory.createConnection(conf);
+            Table acl = conn.getTable(AccessControlLists.ACL_TABLE_NAME)) {
+          BlockingRpcChannel service = acl.coprocessorService(TEST_TABLE.getName());
+          AccessControlService.BlockingInterface protocol =
+              AccessControlService.newBlockingStub(service);
+          AccessControlUtil.getUserPermissions(null, protocol, TEST_TABLE, TEST_FAMILY,
+            TEST_QUALIFIER, "dummy");
+        }
+        return null;
+      }
+    };
+    verifyAllowed(tableUserPermissionAction, SUPERUSER, USER_ADMIN, USER_OWNER, USER_ADMIN_CF);
+    verifyDenied(tableUserPermissionAction, USER_CREATE, USER_RW, USER_RO, USER_NONE, USER_CREATE);
+
+    List<UserPermission> userPermissions;
+    assertEquals(12, AccessControlClient.getUserPermissions(conn, nsPrefix + ".*").size());
+    assertEquals(6, AccessControlClient.getUserPermissions(conn, table1.getNameAsString()).size());
+    assertEquals(6, AccessControlClient
+        .getUserPermissions(conn, table1.getNameAsString(), HConstants.EMPTY_STRING).size());
+    userPermissions = AccessControlClient.getUserPermissions(conn, table1.getNameAsString(),
+      USER_ADMIN_CF.getName());
+    verifyGetUserPermissionResult(userPermissions, 1, null, null, USER_ADMIN_CF.getName(), null);
+    assertEquals(0, AccessControlClient
+        .getUserPermissions(conn, table1.getNameAsString(), nSUser1.getName()).size());
+    // Table group user ACL
+    assertEquals(1, AccessControlClient
+        .getUserPermissions(conn, table1.getNameAsString(), tableGroupUser1.getName()).size());
+    assertEquals(1, AccessControlClient
+        .getUserPermissions(conn, table2.getNameAsString(), tableGroupUser2.getName()).size());
+
+    // Table Users + CF
+    assertEquals(12, AccessControlClient
+        .getUserPermissions(conn, nsPrefix + ".*", HConstants.EMPTY_BYTE_ARRAY).size());
+    userPermissions = AccessControlClient.getUserPermissions(conn, nsPrefix + ".*", TEST_FAMILY);
+    verifyGetUserPermissionResult(userPermissions, 3, TEST_FAMILY, null, null, null);
+    assertEquals(0, AccessControlClient
+        .getUserPermissions(conn, table1.getNameAsString(), Bytes.toBytes("dummmyCF")).size());
+
+    // Table Users + CF + User
+    assertEquals(3,
+      AccessControlClient
+          .getUserPermissions(conn, table1.getNameAsString(), TEST_FAMILY, HConstants.EMPTY_STRING)
+          .size());
+    userPermissions = AccessControlClient.getUserPermissions(conn, table1.getNameAsString(),
+      TEST_FAMILY, USER_ADMIN_CF.getName());
+    verifyGetUserPermissionResult(userPermissions, 1, null, null, USER_ADMIN_CF.getName(),
+      superUsers);
+    assertEquals(0, AccessControlClient
+        .getUserPermissions(conn, table1.getNameAsString(), TEST_FAMILY, nSUser1.getName()).size());
+
+    // Table Users + CF + CQ
+    assertEquals(3, AccessControlClient.getUserPermissions(conn, table1.getNameAsString(),
+      TEST_FAMILY, HConstants.EMPTY_BYTE_ARRAY).size());
+    assertEquals(1, AccessControlClient
+        .getUserPermissions(conn, table1.getNameAsString(), TEST_FAMILY, TEST_QUALIFIER).size());
+    assertEquals(1, AccessControlClient
+        .getUserPermissions(conn, table1.getNameAsString(), TEST_FAMILY, TEST_QUALIFIER2).size());
+    assertEquals(2, AccessControlClient.getUserPermissions(conn, table1.getNameAsString(),
+      HConstants.EMPTY_BYTE_ARRAY, HConstants.EMPTY_BYTE_ARRAY, USER_RW.getName()).size());
+    assertEquals(0, AccessControlClient
+        .getUserPermissions(conn, table1.getNameAsString(), TEST_FAMILY, Bytes.toBytes("dummmyCQ"))
+        .size());
+
+    // Table Users + CF + CQ + User
+    assertEquals(3, AccessControlClient.getUserPermissions(conn, table1.getNameAsString(),
+      TEST_FAMILY, HConstants.EMPTY_BYTE_ARRAY, HConstants.EMPTY_STRING).size());
+    assertEquals(1, AccessControlClient.getUserPermissions(conn, table1.getNameAsString(),
+      TEST_FAMILY, TEST_QUALIFIER, USER_RW.getName()).size());
+    assertEquals(1, AccessControlClient.getUserPermissions(conn, table1.getNameAsString(),
+      TEST_FAMILY, TEST_QUALIFIER2, USER_RW.getName()).size());
+    assertEquals(0, AccessControlClient.getUserPermissions(conn, table1.getNameAsString(),
+      TEST_FAMILY, TEST_QUALIFIER2, nSUser1.getName()).size());
+  }
+
+  /*
+   * Validate the user permission against the specified column family, column qualifier and user
+   * name.
+   */
+  private void verifyGetUserPermissionResult(List<UserPermission> userPermissions, int resultCount,
+      byte[] cf, byte[] cq, String userName, Collection<String> superUsers) {
+    assertEquals(resultCount, userPermissions.size());
+
+    for (UserPermission perm : userPermissions) {
+      if (cf != null) {
+        assertTrue(Bytes.equals(cf, perm.getFamily()));
+      }
+      if (cq != null) {
+        assertTrue(Bytes.equals(cq, perm.getQualifier()));
+      }
+      if (userName != null
+          && (superUsers == null || !superUsers.contains(Bytes.toString(perm.getUser())))) {
+        assertTrue(userName.equals(Bytes.toString(perm.getUser())));
+      }
+    }
+  }
+
+  /*
+   * Dummy ShellBasedUnixGroupsMapping class to retrieve the groups for the test users.
+   */
+  public static class MyShellBasedUnixGroupsMapping extends ShellBasedUnixGroupsMapping
+      implements GroupMappingServiceProvider {
+    @Override
+    public List<String> getGroups(String user) throws IOException {
+      if (user.equals("globalGroupUser1")) {
+        return Arrays.asList(new String[] { "group_admin" });
+      } else if (user.equals("globalGroupUser2")) {
+        return Arrays.asList(new String[] { "group_admin", "group_create" });
+      } else if (user.equals("nsGroupUser1")) {
+        return Arrays.asList(new String[] { "ns_group1" });
+      } else if (user.equals("nsGroupUser2")) {
+        return Arrays.asList(new String[] { "ns_group2" });
+      } else if (user.equals("tableGroupUser1")) {
+        return Arrays.asList(new String[] { "table_group1" });
+      } else if (user.equals("tableGroupUser2")) {
+        return Arrays.asList(new String[] { "table_group2" });
+      } else {
+        return super.getGroups(user);
+      }
+    }
+  }
 }