Copilot commented on code in PR #10996:
URL: https://github.com/apache/gravitino/pull/10996#discussion_r3245717214
##########
server-common/src/main/java/org/apache/gravitino/server/authorization/jcasbin/JcasbinAuthorizer.java:
##########
@@ -490,86 +630,240 @@ private boolean authorizeByJcasbin(
}
}
- private static UserEntity getUserEntity(String username, String metalake)
throws IOException {
+ //
---------------------------------------------------------------------------
+ // Metadata ID / user info / owner resolution — two-tier (request +
Caffeine)
+ //
---------------------------------------------------------------------------
+
+ /**
+ * Two-tier lookup: the per-request map in {@code requestContext} dedups
calls within the same
+ * HTTP request; on a miss, we consult the long-lived Caffeine cache, and
finally fall back to a
+ * DB query via {@link MetadataIdConverter}.
+ */
+ private Long resolveMetadataId(
+ MetadataObject metadataObject, String metalake,
AuthorizationRequestContext requestContext) {
+ String cacheKey = buildCacheKey(metalake, metadataObject);
+ return requestContext.computeMetadataIdIfAbsent(
+ cacheKey,
+ k -> {
+ Optional<Long> cached = metadataIdCache.getIfPresent(k);
+ if (cached.isPresent()) {
+ return cached.get();
+ }
+ Long id = MetadataIdConverter.getID(metadataObject, metalake);
+ metadataIdCache.put(k, id);
+ return id;
+ });
+ }
+
+ /**
+ * Per-request {@link UserUpdatedAt} lookup. The underlying {@code
user_meta} query is issued at
+ * most once per (metalake, username) within a single request.
+ */
+ private Optional<UserUpdatedAt> loadUserInfo(
+ String metalake, String username, AuthorizationRequestContext
requestContext) {
+ String cacheKey = metalake + KEY_SEP + username;
+ return requestContext.computeUserInfoIfAbsent(
+ cacheKey,
+ k ->
+ Optional.ofNullable(
+ SessionUtils.getWithoutCommit(
+ UserMetaMapper.class, m -> m.getUserUpdatedAt(metalake,
username))));
+ }
+
+ /**
+ * Two-tier owner lookup: request-level dedup first, then the shared {@code
ownerRelCache}, and
+ * finally a single {@code owner_meta} query. A successful DB fetch
populates both tiers so
+ * subsequent {@code isOwner} calls — in this request and later ones — hit
the cache.
+ */
+ private Optional<OwnerInfo> resolveOwnerId(
+ Long metadataId,
+ MetadataObject.Type metadataType,
+ AuthorizationRequestContext requestContext) {
+ return requestContext.computeOwnerIfAbsent(
+ metadataId,
+ id -> {
+ Optional<Optional<OwnerInfo>> cached =
ownerRelCache.getIfPresent(id);
+ if (cached.isPresent()) {
+ return cached.get();
+ }
+ OwnerInfo ownerInfo =
+ SessionUtils.getWithoutCommit(
+ OwnerMetaMapper.class,
+ m -> m.selectOwnerByMetadataObjectIdAndType(id,
metadataType.name()));
+ Optional<OwnerInfo> owner = ownerInfo == null ? Optional.empty() :
Optional.of(ownerInfo);
+ ownerRelCache.put(id, owner);
+ return owner;
+ });
+ }
+
+ /**
+ * Returns true when the cached owner type and ID match the current user or
one of the user's
+ * groups.
+ */
+ private boolean ownerMatchesUserOrGroups(
+ Optional<OwnerInfo> owner, long userId, String metalake) {
+ if (!owner.isPresent()) {
+ return false;
+ }
+ OwnerInfo ownerInfo = owner.get();
+ if
(Entity.EntityType.USER.name().equalsIgnoreCase(ownerInfo.getOwnerType())) {
+ return ownerInfo.getOwnerId() == userId;
+ }
+ if
(!Entity.EntityType.GROUP.name().equalsIgnoreCase(ownerInfo.getOwnerType())) {
+ return false;
+ }
EntityStore entityStore = GravitinoEnv.getInstance().entityStore();
- UserEntity userEntity =
- entityStore.get(
- NameIdentifierUtil.ofUser(metalake, username),
- Entity.EntityType.USER,
- UserEntity.class);
- return userEntity;
+ for (GroupEntity groupEntity : resolveCurrentUserGroups(metalake,
entityStore)) {
+ if (Objects.equals(groupEntity.id(), ownerInfo.getOwnerId())) {
+ return true;
+ }
+ }
+ return false;
}
+ //
---------------------------------------------------------------------------
+ // 4-step role loading with version validation
+ //
---------------------------------------------------------------------------
+
private void loadRolePrivilege(
- String metalake, String username, Long userId,
AuthorizationRequestContext requestContext) {
+ String metalake,
+ String username,
+ long userId,
+ UserUpdatedAt userInfo,
+ AuthorizationRequestContext requestContext) {
requestContext.loadRole(
() -> {
- EntityStore entityStore = GravitinoEnv.getInstance().entityStore();
- NameIdentifier userNameIdentifier =
NameIdentifierUtil.ofUser(metalake, username);
- List<RoleEntity> entities;
- try {
- entities =
- entityStore
- .relationOperations()
- .listEntitiesByRelation(
- SupportsRelationOperations.Type.ROLE_USER_REL,
- userNameIdentifier,
- Entity.EntityType.USER);
- List<CompletableFuture<Void>> loadRoleFutures = new ArrayList<>();
- Set<String> desiredRoleIds = new HashSet<>();
- for (RoleEntity role : entities) {
- desiredRoleIds.add(String.valueOf(role.id()));
- addRoleForUserAndLoadPolicies(
- userId, metalake, role.id(), role.name(), loadRoleFutures,
entityStore);
- }
-
- // Load roles inherited from the user's groups.
- for (GroupEntity groupEntity : resolveCurrentUserGroups(metalake,
entityStore)) {
- List<Long> roleIds = groupEntity.roleIds();
- List<String> roleNames = groupEntity.roleNames();
- if (roleIds == null || roleNames == null) {
- continue;
- }
- if (roleIds.size() != roleNames.size()) {
- LOG.warn(
- "Group {} has mismatched roleIds ({}) and roleNames ({})
-- skipping",
- groupEntity.name(),
- roleIds.size(),
- roleNames.size());
- continue;
- }
- for (int i = 0; i < roleIds.size(); i++) {
- desiredRoleIds.add(String.valueOf(roleIds.get(i)));
- addRoleForUserAndLoadPolicies(
- userId,
- metalake,
- roleIds.get(i),
- roleNames.get(i),
- loadRoleFutures,
- entityStore);
- }
- }
+ // Step 1a: version-validated user-direct roles via cache.
+ List<Long> userDirectRoleIds = loadUserRoles(metalake, username,
userId, userInfo);
- CompletableFuture.allOf(loadRoleFutures.toArray(new
CompletableFuture[0])).join();
+ // Step 1b: version-validated group-inherited roles via cache. Group
membership comes
+ // from the IdP-pushed UserPrincipal; for each group we load its
roles via the same
+ // version-validated path as users (group_meta.updated_at as the
staleness sentinel).
+ List<Long> groupInheritedRoleIds = new ArrayList<>();
+ for (String groupname : currentPrincipalGroupNames()) {
+ groupInheritedRoleIds.addAll(
+ loadGroupRoles(metalake, groupname, userId, requestContext));
+ }
- // Prune stale g-rows: remove role mappings that are no longer
valid
- // (e.g. user was removed from a group at the IdP level).
- String userIdStr = String.valueOf(userId);
- for (String currentRole :
allowEnforcer.getRolesForUser(userIdStr)) {
- if (!desiredRoleIds.contains(currentRole)) {
- allowEnforcer.deleteRoleForUser(userIdStr, currentRole);
- denyEnforcer.deleteRoleForUser(userIdStr, currentRole);
- }
+ // Prune stale g-rows: any role currently bound but no longer in the
desired
+ // set (e.g. user removed from a group at the IdP, or role
unassigned).
+ Set<String> desiredRoleIds = new HashSet<>();
+ for (Long id : userDirectRoleIds) {
+ desiredRoleIds.add(String.valueOf(id));
+ }
+ for (Long id : groupInheritedRoleIds) {
+ desiredRoleIds.add(String.valueOf(id));
+ }
+ String userIdStr = String.valueOf(userId);
+ for (String currentRole : allowEnforcer.getRolesForUser(userIdStr)) {
+ if (!desiredRoleIds.contains(currentRole)) {
+ allowEnforcer.deleteRoleForUser(userIdStr, currentRole);
+ denyEnforcer.deleteRoleForUser(userIdStr, currentRole);
}
- } catch (IOException e) {
- throw new RuntimeException(e);
+ }
+
+ // Step 3: batch version-check all role IDs (direct +
group-inherited),
+ // load stale ones (1 query for the version probe).
+ List<Long> allRoleIds = new ArrayList<>(userDirectRoleIds);
+ allRoleIds.addAll(groupInheritedRoleIds);
+ if (!allRoleIds.isEmpty()) {
+ versionCheckAndLoadRoles(metalake, allRoleIds, requestContext);
}
});
}
+ private List<Long> loadUserRoles(
+ String metalake, String username, long userId, UserUpdatedAt userInfo) {
+ String userCacheKey = metalake + KEY_SEP + username;
+ Optional<CachedUserRoles> cachedOpt =
userRoleCache.getIfPresent(userCacheKey);
+
+ if (cachedOpt.isPresent() && cachedOpt.get().getUpdatedAt() >=
userInfo.getUpdatedAt()) {
+ // Cache is still valid
+ CachedUserRoles cached = cachedOpt.get();
+ bindUserRoles(userId, cached.getRoleIds());
+ return cached.getRoleIds();
+ }
+
+ // Cache miss or stale — reload from DB
+ List<RolePO> rolePOs =
+ SessionUtils.getWithoutCommit(RoleMetaMapper.class, m ->
m.listRolesByUserId(userId));
+ List<Long> roleIds =
rolePOs.stream().map(RolePO::getRoleId).collect(Collectors.toList());
+
+ userRoleCache.put(userCacheKey, new CachedUserRoles(userId,
userInfo.getUpdatedAt(), roleIds));
+ bindUserRoles(userId, roleIds);
+ return roleIds;
+ }
+
+ /**
+ * Per-request {@link GroupUpdatedAt} lookup, mirroring {@link
#loadUserInfo}. The {@code
+ * group_meta} probe runs at most once per (metalake, groupname) within a
single request.
+ */
+ private Optional<GroupUpdatedAt> loadGroupInfo(
+ String metalake, String groupname, AuthorizationRequestContext
requestContext) {
+ String cacheKey = metalake + KEY_SEP + groupname;
+ return requestContext.computeGroupInfoIfAbsent(
+ cacheKey,
+ k ->
+ Optional.ofNullable(
+ SessionUtils.getWithoutCommit(
+ GroupMetaMapper.class, m -> m.getGroupUpdatedAt(metalake,
groupname))));
+ }
+
+ /**
+ * Version-validated group-role load, mirroring {@link #loadUserRoles}. Uses
{@code
+ * group_meta.updated_at} as the staleness sentinel: if the cached snapshot
is at least as fresh
+ * as the DB version, we reuse it; otherwise we reload from {@code
role_meta}. In both cases the
+ * resulting role IDs are bound to the user's jcasbin g-rows so that the
enforcer sees inherited
+ * privileges. Groups missing from the DB return an empty list.
+ */
+ private List<Long> loadGroupRoles(
+ String metalake, String groupname, long userId,
AuthorizationRequestContext requestContext) {
+ Optional<GroupUpdatedAt> groupInfoOpt = loadGroupInfo(metalake, groupname,
requestContext);
+ if (!groupInfoOpt.isPresent()) {
+ return new ArrayList<>();
+ }
+ GroupUpdatedAt groupInfo = groupInfoOpt.get();
+ long groupId = groupInfo.getGroupId();
+ String groupCacheKey = metalake + KEY_SEP + groupname;
+ Optional<CachedGroupRoles> cachedOpt =
groupRoleCache.getIfPresent(groupCacheKey);
+
+ if (cachedOpt.isPresent() && cachedOpt.get().getUpdatedAt() >=
groupInfo.getUpdatedAt()) {
+ CachedGroupRoles cached = cachedOpt.get();
+ bindUserRoles(userId, cached.getRoleIds());
+ return cached.getRoleIds();
Review Comment:
This group-role cache validation has the same timestamp-collision problem as
the user-role cache: `group_meta.updated_at` is generated with millisecond
precision, so multiple group role changes in one millisecond can compare equal
and incorrectly reuse stale inherited roles. Use a strictly monotonic version
value, or otherwise force a reload when equality cannot prove freshness.
##########
server-common/src/main/java/org/apache/gravitino/server/authorization/jcasbin/JcasbinAuthorizer.java:
##########
@@ -587,178 +881,302 @@ private List<GroupEntity>
resolveCurrentUserGroups(String metalake, EntityStore
return entityStore.batchGet(groupIdents, Entity.EntityType.GROUP,
GroupEntity.class);
}
- /**
- * Adds a role mapping for the given user in both enforcers and
asynchronously loads the role's
- * policies if they are not already cached. When a role needs loading, the
resulting {@link
- * CompletableFuture} is appended to {@code loadRoleFutures} so the caller
can join all futures
- * after processing both direct and group-inherited roles.
- */
- private void addRoleForUserAndLoadPolicies(
- Long userId,
- String metalake,
- Long roleId,
- String roleName,
- List<CompletableFuture<Void>> loadRoleFutures,
- EntityStore entityStore) {
- allowEnforcer.addRoleForUser(String.valueOf(userId),
String.valueOf(roleId));
- denyEnforcer.addRoleForUser(String.valueOf(userId),
String.valueOf(roleId));
- if (loadedRoles.getIfPresent(roleId) != null) {
- return;
- }
- CompletableFuture<Void> loadRoleFuture =
- CompletableFuture.supplyAsync(
- () -> {
- try {
- return entityStore.get(
- NameIdentifierUtil.ofRole(metalake, roleName),
- Entity.EntityType.ROLE,
- RoleEntity.class);
- } catch (Exception e) {
- throw new RuntimeException("Failed to load role: " +
roleName, e);
- }
- },
- executor)
- .thenAcceptAsync(
- roleEntity -> {
- loadPolicyByRoleEntity(roleEntity);
- loadedRoles.put(roleId, true);
- },
- executor);
- loadRoleFutures.add(loadRoleFuture);
- }
-
- private void loadOwnerPolicy(String metalake, MetadataObject metadataObject,
Long metadataId) {
- if (ownerRel.getIfPresent(metadataId) != null) {
- LOG.debug("Metadata {} OWNER has been loaded.", metadataId);
- return;
- }
- try {
- NameIdentifier entityIdent = MetadataObjectUtil.toEntityIdent(metalake,
metadataObject);
- EntityStore entityStore = GravitinoEnv.getInstance().entityStore();
- List<? extends Entity> owners =
- entityStore
- .relationOperations()
- .listEntitiesByRelation(
- SupportsRelationOperations.Type.OWNER_REL,
- entityIdent,
- Entity.EntityType.valueOf(metadataObject.type().name()));
- if (owners.isEmpty()) {
- ownerRel.put(metadataId, Optional.empty());
- } else {
- for (Entity ownerEntity : owners) {
- if (ownerEntity instanceof UserEntity) {
- UserEntity user = (UserEntity) ownerEntity;
- ownerRel.put(
- metadataId,
- Optional.of(new OwnerInfo(user.id(), Entity.EntityType.USER,
user.name())));
- } else if (ownerEntity instanceof GroupEntity) {
- GroupEntity group = (GroupEntity) ownerEntity;
- ownerRel.put(
- metadataId,
- Optional.of(new OwnerInfo(group.id(), Entity.EntityType.GROUP,
group.name())));
- }
- }
+ private void versionCheckAndLoadRoles(
+ String metalake, List<Long> roleIds, AuthorizationRequestContext
requestContext) {
+ // Step 3: batch fetch (roleId, roleName, updated_at) for all role IDs — 1
query
+ List<Long> uniqueRoleIds =
roleIds.stream().distinct().collect(Collectors.toList());
+ List<RoleUpdatedAt> roleVersions =
+ SessionUtils.getWithoutCommit(
+ RoleMetaMapper.class, m -> m.batchGetRoleUpdatedAt(uniqueRoleIds));
+
+ for (RoleUpdatedAt rv : roleVersions) {
+ long roleId = rv.getRoleId();
+ long dbUpdatedAt = rv.getUpdatedAt();
+ Optional<Long> cachedUpdatedAt = loadedRoles.getIfPresent(roleId);
+
+ if (cachedUpdatedAt.isPresent() && cachedUpdatedAt.get() >= dbUpdatedAt)
{
+ // Role policies are still current
+ continue;
}
- } catch (IOException e) {
- LOG.warn("Can not load metadata owner", e);
+
+ // Stale or missing — evict old policies and reload
+ if (cachedUpdatedAt.isPresent()) {
+ allowEnforcer.deleteRole(String.valueOf(roleId));
+ denyEnforcer.deleteRole(String.valueOf(roleId));
+ }
+
+ // Load full role entity using roleName from the batch query (no extra
DB scan)
+ try {
+ EntityStore entityStore = GravitinoEnv.getInstance().entityStore();
+ RoleEntity roleEntity =
+ entityStore.get(
+ NameIdentifierUtil.ofRole(metalake, rv.getRoleName()),
+ Entity.EntityType.ROLE,
+ RoleEntity.class);
+ loadPolicyByRoleEntity(roleEntity, requestContext);
+ } catch (Exception e) {
+ LOG.warn("Failed to load role policies for roleId {}", roleId, e);
+ continue;
+ }
+
+ loadedRoles.put(roleId, dbUpdatedAt);
Review Comment:
When a cached role is stale, this reload path leaves the old `loadedRoles`
entry in the cache and then calls `put` with the new version. Because
`LoadedRolesCache` has a removal listener that calls `deleteRole` on
replacements, the `put` can delete the policies that were just loaded, causing
updated roles to authorize as if they had no privileges. Invalidate/remove the
old cache entry before reloading, or avoid firing the eviction listener on this
replacement path.
##########
core/src/main/java/org/apache/gravitino/Configs.java:
##########
@@ -330,6 +333,24 @@ private Configs() {}
.longConf()
.createWithDefault(DEFAULT_GRAVITINO_AUTHORIZATION_OWNER_CACHE_SIZE);
+ public static final long
DEFAULT_GRAVITINO_AUTHORIZATION_METADATA_ID_CACHE_SIZE = 100000L;
+
+ public static final ConfigEntry<Long>
GRAVITINO_AUTHORIZATION_METADATA_ID_CACHE_SIZE =
+ new ConfigBuilder("gravitino.authorization.jcasbin.metadataIdCacheSize")
+ .doc("The maximum size of the metadata-id cache for authorization")
+ .version(ConfigConstants.VERSION_1_3_0)
+ .longConf()
+
.createWithDefault(DEFAULT_GRAVITINO_AUTHORIZATION_METADATA_ID_CACHE_SIZE);
+
+ public static final long
DEFAULT_GRAVITINO_AUTHORIZATION_CHANGE_POLL_INTERVAL_SECS = 3L;
+
+ public static final ConfigEntry<Long>
GRAVITINO_AUTHORIZATION_CHANGE_POLL_INTERVAL_SECS =
+ new
ConfigBuilder("gravitino.authorization.jcasbin.changePollIntervalSecs")
+ .doc("The interval in seconds for polling entity and owner changes")
+ .version(ConfigConstants.VERSION_1_3_0)
+ .longConf()
+
.createWithDefault(DEFAULT_GRAVITINO_AUTHORIZATION_CHANGE_POLL_INTERVAL_SECS);
Review Comment:
These new user-facing authorization cache settings are added to `Configs`,
but the existing authorization configuration docs still only list
`cacheExpirationSecs`, `roleCacheSize`, and `ownerCacheSize`. Please document
`metadataIdCacheSize` and `changePollIntervalSecs` alongside the other JCasbin
cache options so operators know they can tune the new caches/poller.
##########
server-common/build.gradle.kts:
##########
@@ -25,6 +25,9 @@ plugins {
}
dependencies {
+ annotationProcessor(libs.lombok)
+ compileOnly(libs.lombok)
Review Comment:
This module did not previously depend on Lombok; the new dependency is only
needed for two simple cache snapshot classes. The project guideline discourages
adding dependencies without an explicit need, so prefer writing the
constructor/getters directly and avoid adding Lombok to `server-common`.
##########
server-common/src/main/java/org/apache/gravitino/server/authorization/jcasbin/JcasbinAuthorizer.java:
##########
@@ -490,86 +630,240 @@ private boolean authorizeByJcasbin(
}
}
- private static UserEntity getUserEntity(String username, String metalake)
throws IOException {
+ //
---------------------------------------------------------------------------
+ // Metadata ID / user info / owner resolution — two-tier (request +
Caffeine)
+ //
---------------------------------------------------------------------------
+
+ /**
+ * Two-tier lookup: the per-request map in {@code requestContext} dedups
calls within the same
+ * HTTP request; on a miss, we consult the long-lived Caffeine cache, and
finally fall back to a
+ * DB query via {@link MetadataIdConverter}.
+ */
+ private Long resolveMetadataId(
+ MetadataObject metadataObject, String metalake,
AuthorizationRequestContext requestContext) {
+ String cacheKey = buildCacheKey(metalake, metadataObject);
+ return requestContext.computeMetadataIdIfAbsent(
+ cacheKey,
+ k -> {
+ Optional<Long> cached = metadataIdCache.getIfPresent(k);
+ if (cached.isPresent()) {
+ return cached.get();
+ }
+ Long id = MetadataIdConverter.getID(metadataObject, metalake);
+ metadataIdCache.put(k, id);
+ return id;
+ });
+ }
+
+ /**
+ * Per-request {@link UserUpdatedAt} lookup. The underlying {@code
user_meta} query is issued at
+ * most once per (metalake, username) within a single request.
+ */
+ private Optional<UserUpdatedAt> loadUserInfo(
+ String metalake, String username, AuthorizationRequestContext
requestContext) {
+ String cacheKey = metalake + KEY_SEP + username;
+ return requestContext.computeUserInfoIfAbsent(
+ cacheKey,
+ k ->
+ Optional.ofNullable(
+ SessionUtils.getWithoutCommit(
+ UserMetaMapper.class, m -> m.getUserUpdatedAt(metalake,
username))));
+ }
+
+ /**
+ * Two-tier owner lookup: request-level dedup first, then the shared {@code
ownerRelCache}, and
+ * finally a single {@code owner_meta} query. A successful DB fetch
populates both tiers so
+ * subsequent {@code isOwner} calls — in this request and later ones — hit
the cache.
+ */
+ private Optional<OwnerInfo> resolveOwnerId(
+ Long metadataId,
+ MetadataObject.Type metadataType,
+ AuthorizationRequestContext requestContext) {
+ return requestContext.computeOwnerIfAbsent(
+ metadataId,
+ id -> {
+ Optional<Optional<OwnerInfo>> cached =
ownerRelCache.getIfPresent(id);
+ if (cached.isPresent()) {
+ return cached.get();
+ }
+ OwnerInfo ownerInfo =
+ SessionUtils.getWithoutCommit(
+ OwnerMetaMapper.class,
+ m -> m.selectOwnerByMetadataObjectIdAndType(id,
metadataType.name()));
+ Optional<OwnerInfo> owner = ownerInfo == null ? Optional.empty() :
Optional.of(ownerInfo);
+ ownerRelCache.put(id, owner);
+ return owner;
+ });
+ }
+
+ /**
+ * Returns true when the cached owner type and ID match the current user or
one of the user's
+ * groups.
+ */
+ private boolean ownerMatchesUserOrGroups(
+ Optional<OwnerInfo> owner, long userId, String metalake) {
+ if (!owner.isPresent()) {
+ return false;
+ }
+ OwnerInfo ownerInfo = owner.get();
+ if
(Entity.EntityType.USER.name().equalsIgnoreCase(ownerInfo.getOwnerType())) {
+ return ownerInfo.getOwnerId() == userId;
+ }
+ if
(!Entity.EntityType.GROUP.name().equalsIgnoreCase(ownerInfo.getOwnerType())) {
+ return false;
+ }
EntityStore entityStore = GravitinoEnv.getInstance().entityStore();
- UserEntity userEntity =
- entityStore.get(
- NameIdentifierUtil.ofUser(metalake, username),
- Entity.EntityType.USER,
- UserEntity.class);
- return userEntity;
+ for (GroupEntity groupEntity : resolveCurrentUserGroups(metalake,
entityStore)) {
+ if (Objects.equals(groupEntity.id(), ownerInfo.getOwnerId())) {
+ return true;
+ }
+ }
+ return false;
}
+ //
---------------------------------------------------------------------------
+ // 4-step role loading with version validation
+ //
---------------------------------------------------------------------------
+
private void loadRolePrivilege(
- String metalake, String username, Long userId,
AuthorizationRequestContext requestContext) {
+ String metalake,
+ String username,
+ long userId,
+ UserUpdatedAt userInfo,
+ AuthorizationRequestContext requestContext) {
requestContext.loadRole(
() -> {
- EntityStore entityStore = GravitinoEnv.getInstance().entityStore();
- NameIdentifier userNameIdentifier =
NameIdentifierUtil.ofUser(metalake, username);
- List<RoleEntity> entities;
- try {
- entities =
- entityStore
- .relationOperations()
- .listEntitiesByRelation(
- SupportsRelationOperations.Type.ROLE_USER_REL,
- userNameIdentifier,
- Entity.EntityType.USER);
- List<CompletableFuture<Void>> loadRoleFutures = new ArrayList<>();
- Set<String> desiredRoleIds = new HashSet<>();
- for (RoleEntity role : entities) {
- desiredRoleIds.add(String.valueOf(role.id()));
- addRoleForUserAndLoadPolicies(
- userId, metalake, role.id(), role.name(), loadRoleFutures,
entityStore);
- }
-
- // Load roles inherited from the user's groups.
- for (GroupEntity groupEntity : resolveCurrentUserGroups(metalake,
entityStore)) {
- List<Long> roleIds = groupEntity.roleIds();
- List<String> roleNames = groupEntity.roleNames();
- if (roleIds == null || roleNames == null) {
- continue;
- }
- if (roleIds.size() != roleNames.size()) {
- LOG.warn(
- "Group {} has mismatched roleIds ({}) and roleNames ({})
-- skipping",
- groupEntity.name(),
- roleIds.size(),
- roleNames.size());
- continue;
- }
- for (int i = 0; i < roleIds.size(); i++) {
- desiredRoleIds.add(String.valueOf(roleIds.get(i)));
- addRoleForUserAndLoadPolicies(
- userId,
- metalake,
- roleIds.get(i),
- roleNames.get(i),
- loadRoleFutures,
- entityStore);
- }
- }
+ // Step 1a: version-validated user-direct roles via cache.
+ List<Long> userDirectRoleIds = loadUserRoles(metalake, username,
userId, userInfo);
- CompletableFuture.allOf(loadRoleFutures.toArray(new
CompletableFuture[0])).join();
+ // Step 1b: version-validated group-inherited roles via cache. Group
membership comes
+ // from the IdP-pushed UserPrincipal; for each group we load its
roles via the same
+ // version-validated path as users (group_meta.updated_at as the
staleness sentinel).
+ List<Long> groupInheritedRoleIds = new ArrayList<>();
+ for (String groupname : currentPrincipalGroupNames()) {
+ groupInheritedRoleIds.addAll(
+ loadGroupRoles(metalake, groupname, userId, requestContext));
+ }
- // Prune stale g-rows: remove role mappings that are no longer
valid
- // (e.g. user was removed from a group at the IdP level).
- String userIdStr = String.valueOf(userId);
- for (String currentRole :
allowEnforcer.getRolesForUser(userIdStr)) {
- if (!desiredRoleIds.contains(currentRole)) {
- allowEnforcer.deleteRoleForUser(userIdStr, currentRole);
- denyEnforcer.deleteRoleForUser(userIdStr, currentRole);
- }
+ // Prune stale g-rows: any role currently bound but no longer in the
desired
+ // set (e.g. user removed from a group at the IdP, or role
unassigned).
+ Set<String> desiredRoleIds = new HashSet<>();
+ for (Long id : userDirectRoleIds) {
+ desiredRoleIds.add(String.valueOf(id));
+ }
+ for (Long id : groupInheritedRoleIds) {
+ desiredRoleIds.add(String.valueOf(id));
+ }
+ String userIdStr = String.valueOf(userId);
+ for (String currentRole : allowEnforcer.getRolesForUser(userIdStr)) {
+ if (!desiredRoleIds.contains(currentRole)) {
+ allowEnforcer.deleteRoleForUser(userIdStr, currentRole);
+ denyEnforcer.deleteRoleForUser(userIdStr, currentRole);
}
- } catch (IOException e) {
- throw new RuntimeException(e);
+ }
+
+ // Step 3: batch version-check all role IDs (direct +
group-inherited),
+ // load stale ones (1 query for the version probe).
+ List<Long> allRoleIds = new ArrayList<>(userDirectRoleIds);
+ allRoleIds.addAll(groupInheritedRoleIds);
+ if (!allRoleIds.isEmpty()) {
+ versionCheckAndLoadRoles(metalake, allRoleIds, requestContext);
}
});
}
+ private List<Long> loadUserRoles(
+ String metalake, String username, long userId, UserUpdatedAt userInfo) {
+ String userCacheKey = metalake + KEY_SEP + username;
+ Optional<CachedUserRoles> cachedOpt =
userRoleCache.getIfPresent(userCacheKey);
+
+ if (cachedOpt.isPresent() && cachedOpt.get().getUpdatedAt() >=
userInfo.getUpdatedAt()) {
+ // Cache is still valid
+ CachedUserRoles cached = cachedOpt.get();
+ bindUserRoles(userId, cached.getRoleIds());
+ return cached.getRoleIds();
+ }
+
+ // Cache miss or stale — reload from DB
Review Comment:
This treats the cached direct-role snapshot as valid when the database
`user_meta.updated_at` is equal to the cached timestamp. The writer-side
`touchUserUpdatedAt` uses millisecond-resolution timestamps, so two user-role
changes in the same millisecond can leave `updated_at` unchanged and keep
serving stale role assignments until TTL eviction. The version sentinel needs
to be strictly monotonic (for example, incrementing when the clock does not
advance) before it can be used for strong validation.
##########
server-common/src/main/java/org/apache/gravitino/server/authorization/jcasbin/JcasbinAuthorizer.java:
##########
@@ -587,178 +881,302 @@ private List<GroupEntity>
resolveCurrentUserGroups(String metalake, EntityStore
return entityStore.batchGet(groupIdents, Entity.EntityType.GROUP,
GroupEntity.class);
}
- /**
- * Adds a role mapping for the given user in both enforcers and
asynchronously loads the role's
- * policies if they are not already cached. When a role needs loading, the
resulting {@link
- * CompletableFuture} is appended to {@code loadRoleFutures} so the caller
can join all futures
- * after processing both direct and group-inherited roles.
- */
- private void addRoleForUserAndLoadPolicies(
- Long userId,
- String metalake,
- Long roleId,
- String roleName,
- List<CompletableFuture<Void>> loadRoleFutures,
- EntityStore entityStore) {
- allowEnforcer.addRoleForUser(String.valueOf(userId),
String.valueOf(roleId));
- denyEnforcer.addRoleForUser(String.valueOf(userId),
String.valueOf(roleId));
- if (loadedRoles.getIfPresent(roleId) != null) {
- return;
- }
- CompletableFuture<Void> loadRoleFuture =
- CompletableFuture.supplyAsync(
- () -> {
- try {
- return entityStore.get(
- NameIdentifierUtil.ofRole(metalake, roleName),
- Entity.EntityType.ROLE,
- RoleEntity.class);
- } catch (Exception e) {
- throw new RuntimeException("Failed to load role: " +
roleName, e);
- }
- },
- executor)
- .thenAcceptAsync(
- roleEntity -> {
- loadPolicyByRoleEntity(roleEntity);
- loadedRoles.put(roleId, true);
- },
- executor);
- loadRoleFutures.add(loadRoleFuture);
- }
-
- private void loadOwnerPolicy(String metalake, MetadataObject metadataObject,
Long metadataId) {
- if (ownerRel.getIfPresent(metadataId) != null) {
- LOG.debug("Metadata {} OWNER has been loaded.", metadataId);
- return;
- }
- try {
- NameIdentifier entityIdent = MetadataObjectUtil.toEntityIdent(metalake,
metadataObject);
- EntityStore entityStore = GravitinoEnv.getInstance().entityStore();
- List<? extends Entity> owners =
- entityStore
- .relationOperations()
- .listEntitiesByRelation(
- SupportsRelationOperations.Type.OWNER_REL,
- entityIdent,
- Entity.EntityType.valueOf(metadataObject.type().name()));
- if (owners.isEmpty()) {
- ownerRel.put(metadataId, Optional.empty());
- } else {
- for (Entity ownerEntity : owners) {
- if (ownerEntity instanceof UserEntity) {
- UserEntity user = (UserEntity) ownerEntity;
- ownerRel.put(
- metadataId,
- Optional.of(new OwnerInfo(user.id(), Entity.EntityType.USER,
user.name())));
- } else if (ownerEntity instanceof GroupEntity) {
- GroupEntity group = (GroupEntity) ownerEntity;
- ownerRel.put(
- metadataId,
- Optional.of(new OwnerInfo(group.id(), Entity.EntityType.GROUP,
group.name())));
- }
- }
+ private void versionCheckAndLoadRoles(
+ String metalake, List<Long> roleIds, AuthorizationRequestContext
requestContext) {
+ // Step 3: batch fetch (roleId, roleName, updated_at) for all role IDs — 1
query
+ List<Long> uniqueRoleIds =
roleIds.stream().distinct().collect(Collectors.toList());
+ List<RoleUpdatedAt> roleVersions =
+ SessionUtils.getWithoutCommit(
+ RoleMetaMapper.class, m -> m.batchGetRoleUpdatedAt(uniqueRoleIds));
+
+ for (RoleUpdatedAt rv : roleVersions) {
+ long roleId = rv.getRoleId();
+ long dbUpdatedAt = rv.getUpdatedAt();
+ Optional<Long> cachedUpdatedAt = loadedRoles.getIfPresent(roleId);
+
+ if (cachedUpdatedAt.isPresent() && cachedUpdatedAt.get() >= dbUpdatedAt)
{
+ // Role policies are still current
+ continue;
}
- } catch (IOException e) {
- LOG.warn("Can not load metadata owner", e);
+
+ // Stale or missing — evict old policies and reload
+ if (cachedUpdatedAt.isPresent()) {
+ allowEnforcer.deleteRole(String.valueOf(roleId));
+ denyEnforcer.deleteRole(String.valueOf(roleId));
Review Comment:
`deleteRole` removes the role from users as well as clearing its policies.
Because `bindUserRoles` already added the current user's g-row before this
stale-role branch runs, this call removes the freshly bound user-role
relationship and the subsequent policy reload does not add it back. A role
whose privileges are refreshed can therefore stop applying to the user until a
later request rebinds it.
##########
server-common/src/main/java/org/apache/gravitino/server/authorization/jcasbin/JcasbinAuthorizer.java:
##########
@@ -587,178 +881,302 @@ private List<GroupEntity>
resolveCurrentUserGroups(String metalake, EntityStore
return entityStore.batchGet(groupIdents, Entity.EntityType.GROUP,
GroupEntity.class);
}
- /**
- * Adds a role mapping for the given user in both enforcers and
asynchronously loads the role's
- * policies if they are not already cached. When a role needs loading, the
resulting {@link
- * CompletableFuture} is appended to {@code loadRoleFutures} so the caller
can join all futures
- * after processing both direct and group-inherited roles.
- */
- private void addRoleForUserAndLoadPolicies(
- Long userId,
- String metalake,
- Long roleId,
- String roleName,
- List<CompletableFuture<Void>> loadRoleFutures,
- EntityStore entityStore) {
- allowEnforcer.addRoleForUser(String.valueOf(userId),
String.valueOf(roleId));
- denyEnforcer.addRoleForUser(String.valueOf(userId),
String.valueOf(roleId));
- if (loadedRoles.getIfPresent(roleId) != null) {
- return;
- }
- CompletableFuture<Void> loadRoleFuture =
- CompletableFuture.supplyAsync(
- () -> {
- try {
- return entityStore.get(
- NameIdentifierUtil.ofRole(metalake, roleName),
- Entity.EntityType.ROLE,
- RoleEntity.class);
- } catch (Exception e) {
- throw new RuntimeException("Failed to load role: " +
roleName, e);
- }
- },
- executor)
- .thenAcceptAsync(
- roleEntity -> {
- loadPolicyByRoleEntity(roleEntity);
- loadedRoles.put(roleId, true);
- },
- executor);
- loadRoleFutures.add(loadRoleFuture);
- }
-
- private void loadOwnerPolicy(String metalake, MetadataObject metadataObject,
Long metadataId) {
- if (ownerRel.getIfPresent(metadataId) != null) {
- LOG.debug("Metadata {} OWNER has been loaded.", metadataId);
- return;
- }
- try {
- NameIdentifier entityIdent = MetadataObjectUtil.toEntityIdent(metalake,
metadataObject);
- EntityStore entityStore = GravitinoEnv.getInstance().entityStore();
- List<? extends Entity> owners =
- entityStore
- .relationOperations()
- .listEntitiesByRelation(
- SupportsRelationOperations.Type.OWNER_REL,
- entityIdent,
- Entity.EntityType.valueOf(metadataObject.type().name()));
- if (owners.isEmpty()) {
- ownerRel.put(metadataId, Optional.empty());
- } else {
- for (Entity ownerEntity : owners) {
- if (ownerEntity instanceof UserEntity) {
- UserEntity user = (UserEntity) ownerEntity;
- ownerRel.put(
- metadataId,
- Optional.of(new OwnerInfo(user.id(), Entity.EntityType.USER,
user.name())));
- } else if (ownerEntity instanceof GroupEntity) {
- GroupEntity group = (GroupEntity) ownerEntity;
- ownerRel.put(
- metadataId,
- Optional.of(new OwnerInfo(group.id(), Entity.EntityType.GROUP,
group.name())));
- }
- }
+ private void versionCheckAndLoadRoles(
+ String metalake, List<Long> roleIds, AuthorizationRequestContext
requestContext) {
+ // Step 3: batch fetch (roleId, roleName, updated_at) for all role IDs — 1
query
+ List<Long> uniqueRoleIds =
roleIds.stream().distinct().collect(Collectors.toList());
+ List<RoleUpdatedAt> roleVersions =
+ SessionUtils.getWithoutCommit(
+ RoleMetaMapper.class, m -> m.batchGetRoleUpdatedAt(uniqueRoleIds));
+
+ for (RoleUpdatedAt rv : roleVersions) {
+ long roleId = rv.getRoleId();
+ long dbUpdatedAt = rv.getUpdatedAt();
+ Optional<Long> cachedUpdatedAt = loadedRoles.getIfPresent(roleId);
+
+ if (cachedUpdatedAt.isPresent() && cachedUpdatedAt.get() >= dbUpdatedAt)
{
+ // Role policies are still current
+ continue;
Review Comment:
Role policy freshness also relies on millisecond-resolution
`role_meta.updated_at`. If two privilege edits land in the same millisecond,
the cached value can compare equal to the DB value and this path skips
reloading, leaving stale allow/deny policies active until eviction. The role
version used here needs to be strictly monotonic for version-based
authorization invalidation to be reliable.
##########
server-common/build.gradle.kts:
##########
@@ -55,6 +58,9 @@ dependencies {
implementation(libs.prometheus.servlet)
implementation(libs.nimbus.jose.jwt)
+ testAnnotationProcessor(libs.lombok)
+ testCompileOnly(libs.lombok)
Review Comment:
There are no Lombok annotations in `server-common/src/test/java`, so these
test-scoped Lombok dependencies are unused. Please remove them to keep the
module dependency graph minimal.
##########
core/src/main/java/org/apache/gravitino/Configs.java:
##########
@@ -330,6 +333,24 @@ private Configs() {}
.longConf()
.createWithDefault(DEFAULT_GRAVITINO_AUTHORIZATION_OWNER_CACHE_SIZE);
+ public static final long
DEFAULT_GRAVITINO_AUTHORIZATION_METADATA_ID_CACHE_SIZE = 100000L;
+
+ public static final ConfigEntry<Long>
GRAVITINO_AUTHORIZATION_METADATA_ID_CACHE_SIZE =
+ new ConfigBuilder("gravitino.authorization.jcasbin.metadataIdCacheSize")
+ .doc("The maximum size of the metadata-id cache for authorization")
+ .version(ConfigConstants.VERSION_1_3_0)
+ .longConf()
+
.createWithDefault(DEFAULT_GRAVITINO_AUTHORIZATION_METADATA_ID_CACHE_SIZE);
+
+ public static final long
DEFAULT_GRAVITINO_AUTHORIZATION_CHANGE_POLL_INTERVAL_SECS = 3L;
+
+ public static final ConfigEntry<Long>
GRAVITINO_AUTHORIZATION_CHANGE_POLL_INTERVAL_SECS =
+ new
ConfigBuilder("gravitino.authorization.jcasbin.changePollIntervalSecs")
+ .doc("The interval in seconds for polling entity and owner changes")
+ .version(ConfigConstants.VERSION_1_3_0)
+ .longConf()
Review Comment:
`changePollIntervalSecs` is not validated. If it is set to 0 or a negative
value, `scheduleWithFixedDelay(..., pollIntervalSecs, pollIntervalSecs, ...)`
will throw during authorizer initialization and prevent startup. Add a
positive-value check to fail configuration validation with a clear message.
##########
server-common/src/main/java/org/apache/gravitino/server/authorization/jcasbin/JcasbinAuthorizer.java:
##########
@@ -107,41 +184,62 @@ public void initialize() {
GravitinoEnv.getInstance().config().get(Configs.GRAVITINO_AUTHORIZATION_ROLE_CACHE_SIZE);
long ownerCacheSize =
GravitinoEnv.getInstance().config().get(Configs.GRAVITINO_AUTHORIZATION_OWNER_CACHE_SIZE);
+ long metadataIdCacheSize =
+ GravitinoEnv.getInstance()
+ .config()
+ .get(Configs.GRAVITINO_AUTHORIZATION_METADATA_ID_CACHE_SIZE);
+ long pollIntervalSecs =
+ GravitinoEnv.getInstance()
+ .config()
+ .get(Configs.GRAVITINO_AUTHORIZATION_CHANGE_POLL_INTERVAL_SECS);
+
+ long ttlMs = cacheExpirationSecs * 1000L;
- // Initialize enforcers before the caches that reference them in removal
listeners
+ // Initialize enforcers before caches that reference them in removal
listeners
allowEnforcer = new SyncedEnforcer(getModel("/jcasbin_model.conf"), new
GravitinoAdapter());
allowInternalAuthorizer = new InternalAuthorizer(allowEnforcer);
denyEnforcer = new SyncedEnforcer(getModel("/jcasbin_model.conf"), new
GravitinoAdapter());
denyInternalAuthorizer = new InternalAuthorizer(denyEnforcer);
- loadedRoles =
- Caffeine.newBuilder()
- .expireAfterAccess(cacheExpirationSecs, TimeUnit.SECONDS)
- .maximumSize(roleCacheSize)
- .executor(Runnable::run)
- .removalListener(
- (roleId, value, cause) -> {
- if (roleId != null) {
- allowEnforcer.deleteRole(String.valueOf(roleId));
- denyEnforcer.deleteRole(String.valueOf(roleId));
- }
- })
- .build();
- ownerRel =
- Caffeine.newBuilder()
- .expireAfterAccess(cacheExpirationSecs, TimeUnit.SECONDS)
- .maximumSize(ownerCacheSize)
- .build();
- executor =
- Executors.newFixedThreadPool(
- GravitinoEnv.getInstance()
- .config()
- .get(Configs.GRAVITINO_AUTHORIZATION_THREAD_POOL_SIZE),
- runnable -> {
- Thread thread = new Thread(runnable);
- thread.setName("GravitinoAuthorizer-ThreadPool-" +
thread.getId());
- return thread;
+ // loadedRoles: roleId -> updated_at.
+ // When evicted, we must clean up the corresponding JCasbin policies.
+ loadedRoles = new LoadedRolesCache(ttlMs, roleCacheSize, allowEnforcer,
denyEnforcer);
+
+ userRoleCache = new CaffeineGravitinoCache<>(ttlMs, roleCacheSize);
+ groupRoleCache = new CaffeineGravitinoCache<>(ttlMs, roleCacheSize);
+ metadataIdCache = new CaffeineGravitinoCache<>(ttlMs, metadataIdCacheSize);
+ ownerRelCache = new CaffeineGravitinoCache<>(ttlMs, ownerCacheSize);
+
+ // Initialize id cursors to the current DB tail so startup does not scan
historical changes.
+ //
+ // Known trade-off: an id-based high-water mark can miss rows whose id is
allocated before
+ // the cursor snapshot but whose commit lands after it. Concretely, if
writer A holds id=N-1
+ // uncommitted while writer B commits id=N, selectMaxChangeId() returns N
and the next poll
+ // queries `id > N` — A's row is never consumed. In that case the affected
cache entry stays
+ // stale until either (a) the version-validated path catches it on the
next request, or
+ // (b) TTL eviction. Acceptable for the eventual-consistency caches
(metadataIdCache /
+ // ownerRelCache) targeted by these pollers; revisit if we ever route
strong-consistency
+ // data through here.
+ ownerPollHighWaterId =
+ nullToZero(
+ SessionUtils.getWithoutCommit(
+ OwnerMetaMapper.class, OwnerMetaMapper::selectMaxChangeId));
+ entityPollHighWaterId =
+ nullToZero(
+ SessionUtils.getWithoutCommit(
+ EntityChangeLogMapper.class,
EntityChangeLogMapper::selectMaxChangeId));
Review Comment:
Initializing the poll cursor to the current max id can miss a change that
was allocated before startup but commits after this snapshot, and the affected
`metadataIdCache`/`ownerRelCache` entries are not version-validated. A request
served during that window can cache the pre-change id/owner and keep it until
TTL eviction, which violates the intended poll-interval HA invalidation
guarantee. Consider starting from a conservative cursor (or using
commit-time/created_at ordering) so in-flight changes are still consumed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
