[apache#4903] feat(core,server): Support to grant or revoke privilege…

…s for a role (apache#5021)

### What changes were proposed in this pull request?

Supports to grant or revoke privileges for  a role

### Why are the changes needed?

Fix: apache#4903

### Does this PR introduce _any_ user-facing change?
I will add the document later.

### How was this patch tested?
Add the UT.

api/src/main/java/org/apache/gravitino/authorization/SecurableObjects.java

@@ -21,6 +21,7 @@
 import com.google.common.base.Splitter;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
 import java.util.List;
 import java.util.Objects;
 import java.util.stream.Collectors;
@@ -125,7 +126,8 @@ private static class SecurableObjectImpl extends MetadataObjectImpl implements S
 
     SecurableObjectImpl(String parent, String name, Type type, List<Privilege> privileges) {
       super(parent, name, type);
-      this.privileges = ImmutableList.copyOf(privileges);
+      // Remove duplicated privileges
+      this.privileges = ImmutableList.copyOf(Sets.newHashSet(privileges));
     }
 
     @Override

api/src/main/java/org/apache/gravitino/exceptions/IllegalPrivilegeException.java

@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.exceptions;
+
+import com.google.errorprone.annotations.FormatMethod;
+import com.google.errorprone.annotations.FormatString;
+
+/** An exception thrown when a privilege is invalid. */
+public class IllegalPrivilegeException extends IllegalArgumentException {
+  /**
+   * Constructs a new exception with the specified detail message.
+   *
+   * @param message the detail message.
+   * @param args the arguments to the message.
+   */
+  @FormatMethod
+  public IllegalPrivilegeException(@FormatString String message, Object... args) {
+    super(String.format(message, args));
+  }
+
+  /**
+   * Constructs a new exception with the specified detail message and cause.
+   *
+   * @param cause the cause.
+   * @param message the detail message.
+   * @param args the arguments to the message.
+   */
+  @FormatMethod
+  public IllegalPrivilegeException(Throwable cause, @FormatString String message, Object... args) {
+    super(String.format(message, args), cause);
+  }
+
+  /**
+   * Constructs a new exception with the specified cause.
+   *
+   * @param cause the cause.
+   */
+  public IllegalPrivilegeException(Throwable cause) {
+    super(cause);
+  }
+
+  /** Constructs a new exception with the specified detail message and cause. */
+  public IllegalPrivilegeException() {
+    super();
+  }
+}

clients/client-java/src/main/java/org/apache/gravitino/client/DTOConverters.java

@@ -20,11 +20,14 @@
 
 import static org.apache.gravitino.dto.util.DTOConverters.toFunctionArg;
 
+import java.util.List;
+import java.util.stream.Collectors;
 import org.apache.gravitino.Catalog;
 import org.apache.gravitino.CatalogChange;
 import org.apache.gravitino.MetalakeChange;
 import org.apache.gravitino.Namespace;
 import org.apache.gravitino.SchemaChange;
+import org.apache.gravitino.authorization.Privilege;
 import org.apache.gravitino.authorization.SecurableObject;
 import org.apache.gravitino.dto.AuditDTO;
 import org.apache.gravitino.dto.CatalogDTO;
@@ -302,19 +305,21 @@ static SecurableObjectDTO toSecurableObject(SecurableObject securableObject) {
     return SecurableObjectDTO.builder()
         .withFullName(securableObject.fullName())
         .withType(securableObject.type())
-        .withPrivileges(
-            securableObject.privileges().stream()
-                .map(
-                    privilege -> {
-                      return PrivilegeDTO.builder()
-                          .withCondition(privilege.condition())
-                          .withName(privilege.name())
-                          .build();
-                    })
-                .toArray(PrivilegeDTO[]::new))
+        .withPrivileges(toPrivileges(securableObject.privileges()).toArray(new PrivilegeDTO[0]))
         .build();
   }
 
+  static List<PrivilegeDTO> toPrivileges(List<Privilege> privileges) {
+    return privileges.stream()
+        .map(
+            privilege ->
+                PrivilegeDTO.builder()
+                    .withCondition(privilege.condition())
+                    .withName(privilege.name())
+                    .build())
+        .collect(Collectors.toList());
+  }
+
   static TagUpdateRequest toTagUpdateRequest(TagChange change) {
     if (change instanceof TagChange.RenameTag) {
       return new TagUpdateRequest.RenameTagRequest(((TagChange.RenameTag) change).getNewName());

clients/client-java/src/main/java/org/apache/gravitino/client/ErrorHandlers.java

@@ -31,6 +31,7 @@
 import org.apache.gravitino.exceptions.ConnectionFailedException;
 import org.apache.gravitino.exceptions.FilesetAlreadyExistsException;
 import org.apache.gravitino.exceptions.GroupAlreadyExistsException;
+import org.apache.gravitino.exceptions.IllegalPrivilegeException;
 import org.apache.gravitino.exceptions.MetalakeAlreadyExistsException;
 import org.apache.gravitino.exceptions.NoSuchCatalogException;
 import org.apache.gravitino.exceptions.NoSuchFilesetException;
@@ -626,7 +627,11 @@ public void accept(ErrorResponse errorResponse) {
 
       switch (errorResponse.getCode()) {
         case ErrorConstants.ILLEGAL_ARGUMENTS_CODE:
-          throw new IllegalArgumentException(errorMessage);
+          if (errorResponse.getType().equals(IllegalPrivilegeException.class.getSimpleName())) {
+            throw new IllegalPrivilegeException(errorMessage);
+          } else {
+            throw new IllegalArgumentException(errorMessage);
+          }
 
         case ErrorConstants.NOT_FOUND_CODE:
           if (errorResponse.getType().equals(NoSuchMetalakeException.class.getSimpleName())) {
@@ -669,7 +674,11 @@ public void accept(ErrorResponse errorResponse) {
 
       switch (errorResponse.getCode()) {
         case ErrorConstants.ILLEGAL_ARGUMENTS_CODE:
-          throw new IllegalArgumentException(errorMessage);
+          if (errorResponse.getType().equals(IllegalPrivilegeException.class.getSimpleName())) {
+            throw new IllegalPrivilegeException(errorMessage);
+          } else {
+            throw new IllegalArgumentException(errorMessage);
+          }
 
         case ErrorConstants.NOT_FOUND_CODE:
           if (errorResponse.getType().equals(NoSuchMetalakeException.class.getSimpleName())) {
@@ -680,6 +689,10 @@ public void accept(ErrorResponse errorResponse) {
             throw new NoSuchGroupException(errorMessage);
           } else if (errorResponse.getType().equals(NoSuchRoleException.class.getSimpleName())) {
             throw new NoSuchRoleException(errorMessage);
+          } else if (errorResponse
+              .getType()
+              .equals(NoSuchMetadataObjectException.class.getSimpleName())) {
+            throw new NoSuchMetadataObjectException(errorMessage);
           } else {
             throw new NotFoundException(errorMessage);
           }

clients/client-java/src/main/java/org/apache/gravitino/client/GravitinoClient.java

@@ -29,11 +29,13 @@
 import org.apache.gravitino.SupportsCatalogs;
 import org.apache.gravitino.authorization.Group;
 import org.apache.gravitino.authorization.Owner;
+import org.apache.gravitino.authorization.Privilege;
 import org.apache.gravitino.authorization.Role;
 import org.apache.gravitino.authorization.SecurableObject;
 import org.apache.gravitino.authorization.User;
 import org.apache.gravitino.exceptions.CatalogAlreadyExistsException;
 import org.apache.gravitino.exceptions.GroupAlreadyExistsException;
+import org.apache.gravitino.exceptions.IllegalPrivilegeException;
 import org.apache.gravitino.exceptions.NoSuchCatalogException;
 import org.apache.gravitino.exceptions.NoSuchGroupException;
 import org.apache.gravitino.exceptions.NoSuchMetadataObjectException;
@@ -389,6 +391,47 @@ public String[] listRoleNames() throws NoSuchMetalakeException {
     return getMetalake().listRoleNames();
   }
 
+  /**
+   * Grant privileges to a role.
+   *
+   * @param role The name of the role.
+   * @param privileges The privileges to grant.
+   * @param object The object is associated with privileges to grant.
+   * @return The role after granted.
+   * @throws NoSuchRoleException If the role with the given name does not exist.
+   * @throws NoSuchMetadataObjectException If the metadata object with the given name does not
+   *     exist.
+   * @throws NoSuchMetalakeException If the Metalake with the given name does not exist.
+   * @throws IllegalPrivilegeException If any privilege can't be bind to the metadata object.
+   * @throws RuntimeException If granting roles to a role encounters storage issues.
+   */
+  public Role grantPrivilegesToRole(String role, MetadataObject object, List<Privilege> privileges)
+      throws NoSuchRoleException, NoSuchMetadataObjectException, NoSuchMetalakeException,
+          IllegalPrivilegeException {
+    return getMetalake().grantPrivilegesToRole(role, object, privileges);
+  }
+
+  /**
+   * Revoke privileges from a role.
+   *
+   * @param role The name of the role.
+   * @param privileges The privileges to revoke.
+   * @param object The object is associated with privileges to revoke.
+   * @return The role after revoked.
+   * @throws NoSuchRoleException If the role with the given name does not exist.
+   * @throws NoSuchMetadataObjectException If the metadata object with the given name does not
+   *     exist.
+   * @throws NoSuchMetalakeException If the Metalake with the given name does not exist.
+   * @throws IllegalPrivilegeException If any privilege can't be bind to the metadata object.
+   * @throws RuntimeException If revoking privileges from a role encounters storage issues.
+   */
+  public Role revokePrivilegesFromRole(
+      String role, MetadataObject object, List<Privilege> privileges)
+      throws NoSuchRoleException, NoSuchMetadataObjectException, NoSuchMetalakeException,
+          IllegalPrivilegeException {
+    return getMetalake().revokePrivilegesFromRole(role, object, privileges);
+  }
+
   /**
    * Creates a new builder for constructing a GravitinoClient.
    *

clients/client-java/src/main/java/org/apache/gravitino/client/GravitinoMetalake.java

@@ -37,6 +37,7 @@
 import org.apache.gravitino.SupportsCatalogs;
 import org.apache.gravitino.authorization.Group;
 import org.apache.gravitino.authorization.Owner;
+import org.apache.gravitino.authorization.Privilege;
 import org.apache.gravitino.authorization.Role;
 import org.apache.gravitino.authorization.SecurableObject;
 import org.apache.gravitino.authorization.SupportsRoles;
@@ -49,6 +50,8 @@
 import org.apache.gravitino.dto.requests.CatalogUpdatesRequest;
 import org.apache.gravitino.dto.requests.GroupAddRequest;
 import org.apache.gravitino.dto.requests.OwnerSetRequest;
+import org.apache.gravitino.dto.requests.PrivilegeGrantRequest;
+import org.apache.gravitino.dto.requests.PrivilegeRevokeRequest;
 import org.apache.gravitino.dto.requests.RoleCreateRequest;
 import org.apache.gravitino.dto.requests.RoleGrantRequest;
 import org.apache.gravitino.dto.requests.RoleRevokeRequest;
@@ -75,6 +78,7 @@
 import org.apache.gravitino.dto.responses.UserResponse;
 import org.apache.gravitino.exceptions.CatalogAlreadyExistsException;
 import org.apache.gravitino.exceptions.GroupAlreadyExistsException;
+import org.apache.gravitino.exceptions.IllegalPrivilegeException;
 import org.apache.gravitino.exceptions.NoSuchCatalogException;
 import org.apache.gravitino.exceptions.NoSuchGroupException;
 import org.apache.gravitino.exceptions.NoSuchMetadataObjectException;
@@ -894,6 +898,89 @@ public Group revokeRolesFromGroup(List<String> roles, String group)
     return resp.getGroup();
   }
 
+  /**
+   * Grant privileges to a role.
+   *
+   * @param role The name of the role.
+   * @param privileges The privileges to grant.
+   * @param object The object is associated with privileges to grant.
+   * @return The role after granted.
+   * @throws NoSuchRoleException If the role with the given name does not exist.
+   * @throws NoSuchMetadataObjectException If the metadata object with the given name does not
+   *     exist.
+   * @throws NoSuchMetalakeException If the Metalake with the given name does not exist.
+   * @throws IllegalPrivilegeException If any privilege can't be bind to the metadata object.
+   * @throws RuntimeException If granting privileges to a role encounters storage issues.
+   */
+  public Role grantPrivilegesToRole(String role, MetadataObject object, List<Privilege> privileges)
+      throws NoSuchRoleException, NoSuchMetadataObjectException, NoSuchMetalakeException,
+          IllegalPrivilegeException {
+    PrivilegeGrantRequest request =
+        new PrivilegeGrantRequest(DTOConverters.toPrivileges(privileges));
+    request.validate();
+
+    RoleResponse resp =
+        restClient.put(
+            String.format(
+                API_PERMISSION_PATH,
+                this.name(),
+                String.format(
+                    "roles/%s/%s/%s/grant",
+                    RESTUtils.encodeString(role),
+                    object.type().name().toLowerCase(Locale.ROOT),
+                    object.fullName())),
+            request,
+            RoleResponse.class,
+            Collections.emptyMap(),
+            ErrorHandlers.permissionOperationErrorHandler());
+
+    resp.validate();
+
+    return resp.getRole();
+  }
+
+  /**
+   * Revoke privileges from a role.
+   *
+   * @param role The name of the role.
+   * @param privileges The privileges to revoke.
+   * @param object The object is associated with privileges to revoke.
+   * @return The role after revoked.
+   * @throws NoSuchRoleException If the role with the given name does not exist.
+   * @throws NoSuchMetadataObjectException If the metadata object with the given name does not
+   *     exist.
+   * @throws NoSuchMetalakeException If the Metalake with the given name does not exist.
+   * @throws IllegalPrivilegeException If any privilege can't be bind to the metadata object.
+   * @throws RuntimeException If revoking privileges from a role encounters storage issues.
+   */
+  public Role revokePrivilegesFromRole(
+      String role, MetadataObject object, List<Privilege> privileges)
+      throws NoSuchRoleException, NoSuchMetadataObjectException, NoSuchMetalakeException,
+          IllegalPrivilegeException {
+    PrivilegeRevokeRequest request =
+        new PrivilegeRevokeRequest(DTOConverters.toPrivileges(privileges));
+    request.validate();
+
+    RoleResponse resp =
+        restClient.put(
+            String.format(
+                API_PERMISSION_PATH,
+                this.name(),
+                String.format(
+                    "roles/%s/%s/%s/revoke",
+                    RESTUtils.encodeString(role),
+                    object.type().name().toLowerCase(Locale.ROOT),
+                    object.fullName())),
+            request,
+            RoleResponse.class,
+            Collections.emptyMap(),
+            ErrorHandlers.permissionOperationErrorHandler());
+
+    resp.validate();
+
+    return resp.getRole();
+  }
+
   /**
    * Get the owner of a metadata object.
    *