[#5892] fix(auth): Fix to grant privilege for the metalake (#5919)

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

Fix to grant privilege for the metalake

### Why are the changes needed?

Fix: #5892 

### Does this PR introduce _any_ user-facing change?

No.

### How was this patch tested?

Add a UT.

Author: jerqi

authorizations/authorization-ranger/src/test/java/org/apache/gravitino/authorization/ranger/integration/test/RangerBaseE2EIT.java

@@ -984,4 +984,37 @@ void testDenyPrivileges() throws InterruptedException {
     catalog.asSchemas().dropSchema(schemaName, false);
     metalake.deleteRole(roleName);
   }
+
+  // ISSUE-5892 Fix to grant privilege for the metalake
+  @Test
+  void testGrantPrivilegesForMetalake() throws InterruptedException {
+    // Choose a catalog
+    useCatalog();
+
+    // Create a schema
+    String roleName = currentFunName();
+    metalake.createRole(roleName, Collections.emptyMap(), Collections.emptyList());
+
+    // Grant a create schema privilege
+    metalake.grantPrivilegesToRole(
+        roleName,
+        MetadataObjects.of(null, metalakeName, MetadataObject.Type.METALAKE),
+        Lists.newArrayList(Privileges.CreateSchema.allow()));
+
+    // Fail to create a schema
+    Assertions.assertThrows(
+        AccessControlException.class, () -> sparkSession.sql(SQL_CREATE_SCHEMA));
+
+    // Granted this role to the spark execution user `HADOOP_USER_NAME`
+    String userName1 = System.getenv(HADOOP_USER_NAME);
+    metalake.grantRolesToUser(Lists.newArrayList(roleName), userName1);
+
+    waitForUpdatingPolicies();
+
+    Assertions.assertDoesNotThrow(() -> sparkSession.sql(SQL_CREATE_SCHEMA));
+
+    // Clean up
+    catalog.asSchemas().dropSchema(schemaName, false);
+    metalake.deleteRole(roleName);
+  }
 }

core/src/main/java/org/apache/gravitino/authorization/AuthorizationUtils.java

@@ -173,9 +173,11 @@ public static void callAuthorizationPluginForMetadataObject(
       String metalake, MetadataObject metadataObject, Consumer<AuthorizationPlugin> consumer) {
     CatalogManager catalogManager = GravitinoEnv.getInstance().catalogManager();
     if (needApplyAuthorizationPluginAllCatalogs(metadataObject.type())) {
-      Catalog[] catalogs = catalogManager.listCatalogsInfo(Namespace.of(metalake));
-      for (Catalog catalog : catalogs) {
-        callAuthorizationPluginImpl(consumer, catalog);
+      NameIdentifier[] catalogs = catalogManager.listCatalogs(Namespace.of(metalake));
+      // ListCatalogsInfo return `CatalogInfo` instead of `BaseCatalog`, we need `BaseCatalog` to
+      // call authorization plugin method.
+      for (NameIdentifier catalog : catalogs) {
+        callAuthorizationPluginImpl(consumer, catalogManager.loadCatalog(catalog));
       }
     } else if (needApplyAuthorization(metadataObject.type())) {
       NameIdentifier catalogIdent =
@@ -269,6 +271,11 @@ private static void callAuthorizationPluginImpl(
       if (baseCatalog.getAuthorizationPlugin() != null) {
         consumer.accept(baseCatalog.getAuthorizationPlugin());
       }
+    } else {
+      throw new IllegalArgumentException(
+          String.format(
+              "Catalog %s is not a BaseCatalog, we don't support authorization plugin for it",
+              catalog.type()));
     }
   }
 

core/src/test/java/org/apache/gravitino/authorization/TestAccessControlManagerForPermissions.java

@@ -28,14 +28,14 @@
 import java.time.Instant;
 import java.util.List;
 import org.apache.commons.lang3.reflect.FieldUtils;
-import org.apache.gravitino.Catalog;
 import org.apache.gravitino.Config;
 import org.apache.gravitino.Configs;
 import org.apache.gravitino.Entity;
 import org.apache.gravitino.EntityStore;
 import org.apache.gravitino.GravitinoEnv;
 import org.apache.gravitino.MetadataObject;
 import org.apache.gravitino.MetadataObjects;
+import org.apache.gravitino.NameIdentifier;
 import org.apache.gravitino.Namespace;
 import org.apache.gravitino.catalog.CatalogManager;
 import org.apache.gravitino.connector.BaseCatalog;
@@ -172,8 +172,8 @@ public static void setUp() throws Exception {
     FieldUtils.writeField(GravitinoEnv.getInstance(), "catalogManager", catalogManager, true);
     BaseCatalog catalog = Mockito.mock(BaseCatalog.class);
     Mockito.when(catalogManager.loadCatalog(any())).thenReturn(catalog);
-    Mockito.when(catalogManager.listCatalogsInfo(Mockito.any()))
-        .thenReturn(new Catalog[] {catalog});
+    Mockito.when(catalogManager.listCatalogs(Mockito.any()))
+        .thenReturn(new NameIdentifier[] {NameIdentifier.of("metalake", "catalog")});
     authorizationPlugin = Mockito.mock(AuthorizationPlugin.class);
     Mockito.when(catalog.getAuthorizationPlugin()).thenReturn(authorizationPlugin);
   }

core/src/test/java/org/apache/gravitino/authorization/TestOwnerManager.java

@@ -31,6 +31,7 @@
 import static org.apache.gravitino.Configs.TREE_LOCK_MAX_NODE_IN_MEMORY;
 import static org.apache.gravitino.Configs.TREE_LOCK_MIN_NODE_IN_MEMORY;
 import static org.apache.gravitino.Configs.VERSION_RETENTION_COUNT;
+import static org.mockito.ArgumentMatchers.any;
 
 import com.google.common.collect.Lists;
 import java.io.File;
@@ -40,13 +41,13 @@
 import java.util.UUID;
 import org.apache.commons.io.FileUtils;
 import org.apache.commons.lang3.reflect.FieldUtils;
-import org.apache.gravitino.Catalog;
 import org.apache.gravitino.Config;
 import org.apache.gravitino.EntityStore;
 import org.apache.gravitino.EntityStoreFactory;
 import org.apache.gravitino.GravitinoEnv;
 import org.apache.gravitino.MetadataObject;
 import org.apache.gravitino.MetadataObjects;
+import org.apache.gravitino.NameIdentifier;
 import org.apache.gravitino.catalog.CatalogManager;
 import org.apache.gravitino.connector.BaseCatalog;
 import org.apache.gravitino.connector.authorization.AuthorizationPlugin;
@@ -145,8 +146,9 @@ public static void setUp() throws IOException, IllegalAccessException {
 
     ownerManager = new OwnerManager(entityStore);
     BaseCatalog catalog = Mockito.mock(BaseCatalog.class);
-    Mockito.when(catalogManager.listCatalogsInfo(Mockito.any()))
-        .thenReturn(new Catalog[] {catalog});
+    Mockito.when(catalogManager.loadCatalog(any())).thenReturn(catalog);
+    Mockito.when(catalogManager.listCatalogs(Mockito.any()))
+        .thenReturn(new NameIdentifier[] {NameIdentifier.of("metalake", "catalog")});
     Mockito.when(catalog.getAuthorizationPlugin()).thenReturn(authorizationPlugin);
   }