[#10772] feat(authorization): refactor JcasbinAuthorizer with version-validated cache

core/src/main/java/org/apache/gravitino/Configs.java

@@ -316,7 +316,10 @@ private Configs() {}
 
   public static final ConfigEntry<Long> GRAVITINO_AUTHORIZATION_ROLE_CACHE_SIZE =
       new ConfigBuilder("gravitino.authorization.jcasbin.roleCacheSize")
-          .doc("The maximum size of the role cache for authorization")
+          .doc(
+              "The maximum size of the role-related caches used by the JcasbinAuthorizer. "
+                  + "Shared by the user-role, group-role, and loaded-role caches, so the "
+                  + "effective memory footprint is up to roughly 3x this value.")
           .version(ConfigConstants.VERSION_1_1_1)
           .longConf()
           .createWithDefault(DEFAULT_GRAVITINO_AUTHORIZATION_ROLE_CACHE_SIZE);

server-common/src/main/java/org/apache/gravitino/server/authorization/jcasbin/CachedGroupRoles.java

@@ -0,0 +1,51 @@
+/*
+ * 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.server.authorization.jcasbin;
+
+import java.util.List;
+
+/**
+ * Cached snapshot of a group's role assignments. The {@code updatedAt} timestamp corresponds to the
+ * {@code group_meta.updated_at} column and is used as a version sentinel: if the DB value is newer,
+ * the cached role list is stale and must be reloaded.
+ */
+final class CachedGroupRoles {
+
+  private final long groupId;
+  private final long updatedAt;
+  private final List<Long> roleIds;
+
+  CachedGroupRoles(long groupId, long updatedAt, List<Long> roleIds) {
+    this.groupId = groupId;
+    this.updatedAt = updatedAt;
+    this.roleIds = roleIds;
+  }
+
+  long getGroupId() {
+    return groupId;
+  }
+
+  long getUpdatedAt() {
+    return updatedAt;
+  }
+
+  List<Long> getRoleIds() {
+    return roleIds;
+  }
+}

server-common/src/main/java/org/apache/gravitino/server/authorization/jcasbin/CachedUserRoles.java

@@ -0,0 +1,51 @@
+/*
+ * 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.server.authorization.jcasbin;
+
+import java.util.List;
+
+/**
+ * Cached snapshot of a user's direct role assignments. The {@code updatedAt} timestamp corresponds
+ * to the {@code user_meta.updated_at} column and is used as a version sentinel: if the DB value is
+ * newer, the cached role list is stale and must be reloaded.
+ */
+final class CachedUserRoles {
+
+  private final long userId;
+  private final long updatedAt;
+  private final List<Long> roleIds;
+
+  CachedUserRoles(long userId, long updatedAt, List<Long> roleIds) {
+    this.userId = userId;
+    this.updatedAt = updatedAt;
+    this.roleIds = roleIds;
+  }
+
+  long getUserId() {
+    return userId;
+  }
+
+  long getUpdatedAt() {
+    return updatedAt;
+  }
+
+  List<Long> getRoleIds() {
+    return roleIds;
+  }
+}

server-common/src/main/java/org/apache/gravitino/server/authorization/jcasbin/JcasbinAuthorizationCacheKeys.java

@@ -0,0 +1,70 @@
+/*
+ * 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.server.authorization.jcasbin;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.gravitino.MetadataObject;
+
+/** Cache key factory for JCasbin authorization caches. */
+final class JcasbinAuthorizationCacheKeys {
+
+  /** Unit Separator for internal cache keys. */
+  static final String SEPARATOR = "\u001F";
+
+  private JcasbinAuthorizationCacheKeys() {}
+
+  /**
+   * Builds a path-based key for the metadata id cache.
+   *
+   * <p>Container objects end with the internal separator so prefix invalidation can remove both the
+   * container and entries under the same name path. Leaf objects include the type suffix to avoid
+   * collisions between objects that share the same name path.
+   */
+  static String metadataObjectKey(String metalake, MetadataObject metadataObject) {
+    if (metadataObject.type() == MetadataObject.Type.METALAKE) {
+      return metalake + SEPARATOR;
+    }
+
+    StringBuilder sb = new StringBuilder(metalake);
+    sb.append(SEPARATOR);
+    sb.append(String.join(SEPARATOR, metadataObject.fullName().split("\\.")));
+    if (isMetadataContainer(metadataObject.type())) {
+      sb.append(SEPARATOR);
+    } else {
+      sb.append(SEPARATOR);
+      sb.append(metadataObject.type().name());
+    }
+    return sb.toString();
+  }
+
+  static String userRoleKey(String metalake, String username) {
+    return "USER" + SEPARATOR + metalake + SEPARATOR + username;
+  }
+
+  static String groupRoleKey(String metalake, String groupname) {
+    return "GROUP" + SEPARATOR + metalake + SEPARATOR + groupname;
+  }
+
+  @VisibleForTesting
+  static boolean isMetadataContainer(MetadataObject.Type type) {
+    return type == MetadataObject.Type.METALAKE
+        || type == MetadataObject.Type.CATALOG
+        || type == MetadataObject.Type.SCHEMA;
+  }
+}

server-common/src/main/java/org/apache/gravitino/server/authorization/jcasbin/JcasbinAuthorizationLookups.java

@@ -18,7 +18,6 @@
  */
 package org.apache.gravitino.server.authorization.jcasbin;
 
-import com.google.common.annotations.VisibleForTesting;
 import java.util.Optional;
 import org.apache.gravitino.MetadataObject;
 import org.apache.gravitino.authorization.AuthorizationRequestContext;
@@ -43,17 +42,14 @@
  */
 public class JcasbinAuthorizationLookups {
 
-  /** Unit Separator for internal path-based cache keys. */
-  static final String KEY_SEP = "\u001F";
-
   private final GravitinoCache<String, Long> metadataIdCache;
   private final GravitinoCache<Long, Optional<OwnerInfo>> ownerRelCache;
 
   /**
    * Creates a new lookups facade around the supplied caches. The caches are owned by the caller and
    * remain accessible for invalidation by other components (poller, change hooks).
    *
-   * @param metadataIdCache path-based {@code metalake::catalog::schema::object::TYPE} → entity id
+   * @param metadataIdCache path-based metadata object key → entity id
    * @param ownerRelCache {@code metadataObjectId} → {@link Optional} of {@link OwnerInfo}
    */
   public JcasbinAuthorizationLookups(
@@ -72,7 +68,7 @@ public JcasbinAuthorizationLookups(
    */
   public Optional<Long> resolveMetadataId(
       MetadataObject metadataObject, String metalake, AuthorizationRequestContext requestContext) {
-    String cacheKey = buildCacheKey(metalake, metadataObject);
+    String cacheKey = JcasbinAuthorizationCacheKeys.metadataObjectKey(metalake, metadataObject);
     try {
       // Both cache tiers load atomically and forbid caching null, so a missing object is signalled
       // by throwing through the loaders and translated back to Optional.empty() here. This caches
@@ -116,52 +112,4 @@ public Optional<OwnerInfo> resolveOwnerId(
                   return ownerInfo == null ? Optional.empty() : Optional.of(ownerInfo);
                 }));
   }
-
-  /** Underlying metadata-id cache; exposed for invalidation by the change hooks and the poller. */
-  public GravitinoCache<String, Long> metadataIdCache() {
-    return metadataIdCache;
-  }
-
-  /** Underlying owner cache; exposed for invalidation by the change hooks and the poller. */
-  public GravitinoCache<Long, Optional<OwnerInfo>> ownerRelCache() {
-    return ownerRelCache;
-  }
-
-  /**
-   * Builds a path-based cache key for the metadataIdCache. Container objects end with the internal
-   * separator so a prefix invalidation can remove the container and all entries under the same name
-   * path.
-   *
-   * <p>Examples: {@code metalake<sep>}, {@code metalake<sep>catalog<sep>}, {@code
-   * metalake<sep>catalog<sep>schema<sep>}, {@code
-   * metalake<sep>catalog<sep>schema<sep>table<sep>TABLE}.
-   */
-  @VisibleForTesting
-  public static String buildCacheKey(String metalake, MetadataObject metadataObject) {
-    if (metadataObject.type() == MetadataObject.Type.METALAKE) {
-      return metalake + KEY_SEP;
-    }
-    StringBuilder sb = new StringBuilder(metalake);
-    sb.append(KEY_SEP);
-    // fullName uses '.' as separator, e.g. "catalog1.schema1.table1"
-    String[] parts = metadataObject.fullName().split("\\.");
-    sb.append(String.join(KEY_SEP, parts));
-    if (isContainerType(metadataObject.type())) {
-      // Trailing separator enables prefix-based invalidation.
-      sb.append(KEY_SEP);
-    } else {
-      // Leaf nodes get the type suffix to avoid collisions
-      sb.append(KEY_SEP);
-      sb.append(metadataObject.type().name());
-    }
-    return sb.toString();
-  }
-
-  /** Returns true for entity types that can contain children (metalake, catalog, schema). */
-  @VisibleForTesting
-  public static boolean isContainerType(MetadataObject.Type type) {
-    return type == MetadataObject.Type.METALAKE
-        || type == MetadataObject.Type.CATALOG
-        || type == MetadataObject.Type.SCHEMA;
-  }
 }