[#5750] improvement(auth): Add metalake name in the authorization plugin (#5751)

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

Add metalake name variable in the `BaseAuthorization::newPlugin()`
params.

### Why are the changes needed?

Fix: #5750

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

N/A

### How was this patch tested?

Add ITs.

Author: xunliu

authorizations/authorization-ranger/src/main/java/org/apache/gravitino/authorization/ranger/RangerAuthorization.java

@@ -30,12 +30,13 @@ public String shortName() {
   }
 
   @Override
-  protected AuthorizationPlugin newPlugin(String catalogProvider, Map<String, String> config) {
+  protected AuthorizationPlugin newPlugin(
+      String metalake, String catalogProvider, Map<String, String> config) {
     switch (catalogProvider) {
       case "hive":
       case "lakehouse-iceberg":
       case "lakehouse-paimon":
-        return RangerAuthorizationHadoopSQLPlugin.getInstance(config);
+        return RangerAuthorizationHadoopSQLPlugin.getInstance(metalake, config);
       default:
         throw new IllegalArgumentException("Unknown catalog provider: " + catalogProvider);
     }

authorizations/authorization-ranger/src/main/java/org/apache/gravitino/authorization/ranger/RangerAuthorizationHadoopSQLPlugin.java

@@ -49,16 +49,16 @@ public class RangerAuthorizationHadoopSQLPlugin extends RangerAuthorizationPlugi
       LoggerFactory.getLogger(RangerAuthorizationHadoopSQLPlugin.class);
   private static volatile RangerAuthorizationHadoopSQLPlugin instance = null;
 
-  private RangerAuthorizationHadoopSQLPlugin(Map<String, String> config) {
-    super(config);
+  private RangerAuthorizationHadoopSQLPlugin(String metalake, Map<String, String> config) {
+    super(metalake, config);
   }
 
   public static synchronized RangerAuthorizationHadoopSQLPlugin getInstance(
-      Map<String, String> config) {
+      String metalake, Map<String, String> config) {
     if (instance == null) {
       synchronized (RangerAuthorizationHadoopSQLPlugin.class) {
         if (instance == null) {
-          instance = new RangerAuthorizationHadoopSQLPlugin(config);
+          instance = new RangerAuthorizationHadoopSQLPlugin(metalake, config);
         }
       }
     }

authorizations/authorization-ranger/src/main/java/org/apache/gravitino/authorization/ranger/RangerAuthorizationPlugin.java

@@ -80,12 +80,14 @@ public abstract class RangerAuthorizationPlugin
     implements AuthorizationPlugin, AuthorizationPrivilegesMappingProvider {
   private static final Logger LOG = LoggerFactory.getLogger(RangerAuthorizationPlugin.class);
 
+  protected String metalake;
   protected final String rangerServiceName;
   protected final RangerClientExtension rangerClient;
   private final RangerHelper rangerHelper;
   @VisibleForTesting public final String rangerAdminName;
 
-  protected RangerAuthorizationPlugin(Map<String, String> config) {
+  protected RangerAuthorizationPlugin(String metalake, Map<String, String> config) {
+    this.metalake = metalake;
     String rangerUrl = config.get(AuthorizationPropertiesMeta.RANGER_ADMIN_URL);
     String authType = config.get(AuthorizationPropertiesMeta.RANGER_AUTH_TYPE);
     rangerAdminName = config.get(AuthorizationPropertiesMeta.RANGER_USERNAME);
@@ -108,6 +110,11 @@ protected RangerAuthorizationPlugin(Map<String, String> config) {
             policyResourceDefinesRule());
   }
 
+  @VisibleForTesting
+  public String getMetalake() {
+    return metalake;
+  }
+
   /**
    * Set the Ranger policy resource defines rule.
    *
@@ -251,18 +258,22 @@ public Boolean onMetadataUpdated(MetadataObjectChange... changes) throws Runtime
             ((MetadataObjectChange.RenameMetadataObject) change).metadataObject();
         MetadataObject newMetadataObject =
             ((MetadataObjectChange.RenameMetadataObject) change).newMetadataObject();
-        AuthorizationMetadataObject AuthorizationMetadataObject =
-            translateMetadataObject(metadataObject);
-        AuthorizationMetadataObject newAuthorizationMetadataObject =
+        if (metadataObject.type() == MetadataObject.Type.METALAKE
+            && newMetadataObject.type() == MetadataObject.Type.METALAKE) {
+          // Modify the metalake name
+          this.metalake = newMetadataObject.name();
+        }
+        AuthorizationMetadataObject oldAuthMetadataObject = translateMetadataObject(metadataObject);
+        AuthorizationMetadataObject newAuthMetadataObject =
             translateMetadataObject(newMetadataObject);
-        if (AuthorizationMetadataObject.equals(newAuthorizationMetadataObject)) {
+        if (oldAuthMetadataObject.equals(newAuthMetadataObject)) {
           LOG.info(
               "The metadata object({}) and new metadata object({}) are equal, so ignore rename!",
-              AuthorizationMetadataObject.fullName(),
-              newAuthorizationMetadataObject.fullName());
+              oldAuthMetadataObject.fullName(),
+              newAuthMetadataObject.fullName());
           continue;
         }
-        doRenameMetadataObject(AuthorizationMetadataObject, newAuthorizationMetadataObject);
+        doRenameMetadataObject(oldAuthMetadataObject, newAuthMetadataObject);
       } else if (change instanceof MetadataObjectChange.RemoveMetadataObject) {
         MetadataObject metadataObject =
             ((MetadataObjectChange.RemoveMetadataObject) change).metadataObject();

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

@@ -697,9 +697,8 @@ public void testMetadataObjectChangeRenameMetalake() {
     Assertions.assertTrue(rangerAuthHivePlugin.onRoleCreated(role));
     assertFindManagedPolicyItems(role, true);
 
-    MetadataObject newMetadataObject =
-        MetadataObjects.parse(
-            String.format("metalake-new-%s", currentFunName), oldMetadataObject.type());
+    String newMetalake = String.format("metalake-new-%s", currentFunName);
+    MetadataObject newMetadataObject = MetadataObjects.parse(newMetalake, oldMetadataObject.type());
     Assertions.assertTrue(
         rangerAuthHivePlugin.onMetadataUpdated(
             MetadataObjectChange.rename(oldMetadataObject, newMetadataObject)));
@@ -716,6 +715,7 @@ public void testMetadataObjectChangeRenameMetalake() {
             .withSecurableObjects(Lists.newArrayList(newSecurableObject1))
             .build();
     assertFindManagedPolicyItems(newRole, true);
+    Assertions.assertEquals(newMetalake, rangerAuthHivePlugin.getMetalake());
   }
 
   @Test

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

@@ -89,6 +89,7 @@ public static void init() {
 
     rangerAuthHivePlugin =
         RangerAuthorizationHadoopSQLPlugin.getInstance(
+            "metalake",
             ImmutableMap.of(
                 AuthorizationPropertiesMeta.RANGER_ADMIN_URL,
                 String.format(

core/src/main/java/org/apache/gravitino/connector/BaseCatalog.java

@@ -184,7 +184,7 @@ public AuthorizationPlugin getAuthorizationPlugin() {
     if (authorization == null) {
       return null;
     }
-    return authorization.plugin(provider(), this.conf);
+    return authorization.plugin(entity.namespace().level(0), provider(), this.conf);
   }
 
   public void initAuthorizationPluginInstance(IsolatedClassLoader classLoader) {

core/src/main/java/org/apache/gravitino/connector/authorization/BaseAuthorization.java

@@ -43,13 +43,14 @@ public abstract class BaseAuthorization<T extends BaseAuthorization>
    * @return A new instance of AuthorizationHook.
    */
   protected abstract AuthorizationPlugin newPlugin(
-      String catalogProvider, Map<String, String> config);
+      String metalake, String catalogProvider, Map<String, String> config);
 
-  public AuthorizationPlugin plugin(String catalogProvider, Map<String, String> config) {
+  public AuthorizationPlugin plugin(
+      String metalake, String catalogProvider, Map<String, String> config) {
     if (plugin == null) {
       synchronized (this) {
         if (plugin == null) {
-          plugin = newPlugin(catalogProvider, config);
+          plugin = newPlugin(metalake, catalogProvider, config);
         }
       }
     }

core/src/main/java/org/apache/gravitino/hook/MetalakeHookDispatcher.java

@@ -25,6 +25,7 @@
 import org.apache.gravitino.MetalakeChange;
 import org.apache.gravitino.NameIdentifier;
 import org.apache.gravitino.authorization.AccessControlDispatcher;
+import org.apache.gravitino.authorization.AuthorizationUtils;
 import org.apache.gravitino.authorization.Owner;
 import org.apache.gravitino.authorization.OwnerManager;
 import org.apache.gravitino.exceptions.MetalakeAlreadyExistsException;
@@ -85,10 +86,18 @@ public Metalake createMetalake(
   @Override
   public Metalake alterMetalake(NameIdentifier ident, MetalakeChange... changes)
       throws NoSuchMetalakeException, IllegalArgumentException {
-    // For underlying authorization plugins, the privilege information shouldn't
-    // contain metalake information, so metalake rename won't affect the privileges
-    // of the authorization plugin.
-    return dispatcher.alterMetalake(ident, changes);
+    Metalake alterMetalake = dispatcher.alterMetalake(ident, changes);
+    MetalakeChange.RenameMetalake lastRenameChange = null;
+    for (MetalakeChange change : changes) {
+      if (change instanceof MetalakeChange.RenameMetalake) {
+        lastRenameChange = (MetalakeChange.RenameMetalake) change;
+      }
+    }
+    if (lastRenameChange != null) {
+      AuthorizationUtils.authorizationPluginRenamePrivileges(
+          ident, Entity.EntityType.METALAKE, lastRenameChange.getNewName());
+    }
+    return alterMetalake;
   }
 
   @Override

core/src/test/java/org/apache/gravitino/connector/authorization/mysql/TestMySQLAuthorization.java

@@ -32,7 +32,8 @@ public String shortName() {
   }
 
   @Override
-  protected AuthorizationPlugin newPlugin(String catalogProvider, Map<String, String> config) {
+  protected AuthorizationPlugin newPlugin(
+      String metalake, String catalogProvider, Map<String, String> config) {
     return new TestMySQLAuthorizationPlugin();
   }
 }

core/src/test/java/org/apache/gravitino/connector/authorization/ranger/TestRangerAuthorization.java

@@ -32,7 +32,8 @@ public String shortName() {
   }
 
   @Override
-  protected AuthorizationPlugin newPlugin(String catalogProvider, Map<String, String> config) {
+  protected AuthorizationPlugin newPlugin(
+      String metalake, String catalogProvider, Map<String, String> config) {
     return new TestRangerAuthorizationPlugin();
   }
 }