This is an automated email from the ASF dual-hosted git repository.
yuqi1129 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new 279ed981b0 [#13004] fix(core): Fence live endpoints for relation
writes (#13486)
279ed981b0 is described below
commit 279ed981b0b125177be7df46ee3290b20d59c8f9
Author: Qi Yu <[email protected]>
AuthorDate: Wed Sep 30 11:55:39 2026 +0800
[#13004] fix(core): Fence live endpoints for relation writes (#13486)
### What changes were proposed in this pull request?
Add a transaction-scoped endpoint fence for tag assignments and role
privileges:
- `LiveEndpointMapper` locks active metadata rows by ID with the
database's row-lock syntax. `LiveEndpointService` locks the observed
metalake, catalog, schema, and target in hierarchy order; columns use
their table row as the lock.
- After locking, the service clears the MyBatis session cache and
resolves the name again from the database. It compares the full ID chain
and rejects a deleted, renamed, moved, or same-name replacement endpoint
before the relation write commits.
- Tag assignments and role privilege writes use this fence in the same
transaction as their relation changes. Lock ordering remains metalake
before Role and Tag before Policy to avoid lock-order inversion.
### Why are the changes needed?
A relation writer can resolve an endpoint ID and then commit after
another server deletes or replaces that endpoint. The write can report
success even though the new relation is immediately invisible. Garbage
collection repairs orphan rows later but cannot provide a synchronous
result to the writer.
Fix: #13004
### Scope
Owner relations are already fenced by
`OwnerMetaService.lockOwnedObjectsForOwnerWrite`, which locks the owned
row by ID through `OccWriteSupport`, so they need nothing here.
Statistic writers take a schema lock of their own and adopt this fence
in #13177; `LiveEndpointService` says so rather than implying it already
covers them.
### Does this PR introduce _any_ user-facing change?
Concurrent assignments to deleted or replaced endpoints now fail
synchronously. No API or configuration fields change.
### How was this patch tested?
`TestLiveEndpointService` 12, `TestRoleMetaService` 63,
`TestTagMetaService` 87, `TestPolicyTagRelService` 36 — 198 tests, 0
failures, on H2, MySQL and PostgreSQL via `./gradlew :core:test
-PskipITs` with class filters. `-Werror` compile clean.
The concurrency tests drive their interleaving with latches rather than
sleeps. Two of them were mutation-checked, so they are known to fail
against the defect they describe rather than only passing today:
- `testRechecksNameAfterTransactionLocalLookup` primes the MyBatis
first-level cache with a resolver read, renames the row over a separate
connection, and asserts the fence still rejects it. Replacing
`session.clearCache()` with a no-op call on the same session makes it
fail on all three backends — the stale name-to-ID mapping is served from
the session cache and the rename goes undetected.
- `testEndpointLocksSortIntoTheGlobalLockOrder` pins the sort that keeps
`RoleMetaService`'s endpoint locks in the order `TagMetaService` and
`PolicyTagRelService` take by hand. It uses FUNCTION and MODEL
deliberately: both are declared *after* TAG and POLICY in `EntityType`
(FUNCTION 20, POLICY 15, MODEL 13, TAG 12), so sorting on the enum order
alone takes them after a tag or a policy and inverts the order. Dropping
the TAG/POLICY special cases from `lockOrder` makes the test fail; an
earlier version of it using only METALAKE/CATALOG/TAG/POLICY passed
under that same mutation, because those four are already in ordinal
order.
`EndpointLock` is package-private for that test and has a `toString`, so
a failed ordering assertion names the endpoints instead of printing
identity hashes.
`LiveEndpointService`'s hierarchy walk indexes `namespaceIds` by
position. No current entity type reaches the deeper reads with a chain
too short, but the walk now checks the depth it is about to rely on, so
a malformed chain names the endpoint instead of failing with an index
error.
---
.../relational/mapper/LiveEndpointMapper.java | 31 ++
.../relational/mapper/LiveEndpointSQLProvider.java | 136 +++++++
.../provider/DefaultMapperPackageProvider.java | 2 +
.../relational/service/LiveEndpointService.java | 131 +++++++
.../relational/service/PolicyTagRelService.java | 5 +
.../relational/service/RoleMetaService.java | 87 ++++-
.../storage/relational/service/TagMetaService.java | 18 +-
.../service/TestLiveEndpointService.java | 171 +++++++++
.../relational/service/TestRoleMetaService.java | 398 +++++++++++++++++++++
.../relational/service/TestTagMetaService.java | 71 ++++
10 files changed, 1046 insertions(+), 4 deletions(-)
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/LiveEndpointMapper.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/LiveEndpointMapper.java
new file mode 100644
index 0000000000..6ca22419a7
--- /dev/null
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/LiveEndpointMapper.java
@@ -0,0 +1,31 @@
+/*
+ * 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.storage.relational.mapper;
+
+import org.apache.gravitino.Entity;
+import org.apache.ibatis.annotations.Param;
+import org.apache.ibatis.annotations.SelectProvider;
+
+/** MyBatis mapper for shared locks on live metadata endpoints. */
+public interface LiveEndpointMapper {
+
+ /** Returns the ID of a live, locked endpoint, or null if it was deleted. */
+ @SelectProvider(type = LiveEndpointSQLProvider.class, method =
"lockLiveEndpoint")
+ Long lockLiveEndpoint(@Param("type") Entity.EntityType type, @Param("id")
long id);
+}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/LiveEndpointSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/LiveEndpointSQLProvider.java
new file mode 100644
index 0000000000..4f1b85e7d1
--- /dev/null
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/LiveEndpointSQLProvider.java
@@ -0,0 +1,136 @@
+/*
+ * 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.storage.relational.mapper;
+
+import org.apache.gravitino.Entity;
+import org.apache.gravitino.storage.relational.JDBCBackend;
+import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper;
+import org.apache.ibatis.annotations.Param;
+
+/** SQL for locking a live endpoint row with the backend's shared-lock syntax.
*/
+public class LiveEndpointSQLProvider {
+
+ /** Builds a locking read for a whitelisted metadata-object table. */
+ public static String lockLiveEndpoint(
+ @Param("type") Entity.EntityType type, @Param("id") long id) {
+ String table;
+ String idColumn;
+ switch (type) {
+ case METALAKE:
+ table = MetalakeMetaMapper.TABLE_NAME;
+ idColumn = "metalake_id";
+ break;
+ case CATALOG:
+ table = CatalogMetaMapper.TABLE_NAME;
+ idColumn = "catalog_id";
+ break;
+ case SCHEMA:
+ table = SchemaMetaMapper.TABLE_NAME;
+ idColumn = "schema_id";
+ break;
+ case TABLE:
+ table = TableMetaMapper.TABLE_NAME;
+ idColumn = "table_id";
+ break;
+ case VIEW:
+ table = ViewMetaMapper.TABLE_NAME;
+ idColumn = "view_id";
+ break;
+ case SEMANTIC_MODEL:
+ table = SemanticModelMetaMapper.TABLE_NAME;
+ idColumn = "semantic_model_id";
+ break;
+ case FILESET:
+ table = FilesetMetaMapper.META_TABLE_NAME;
+ idColumn = "fileset_id";
+ break;
+ case TOPIC:
+ table = TopicMetaMapper.TABLE_NAME;
+ idColumn = "topic_id";
+ break;
+ case MODEL:
+ table = ModelMetaMapper.TABLE_NAME;
+ idColumn = "model_id";
+ break;
+ case FUNCTION:
+ table = FunctionMetaMapper.TABLE_NAME;
+ idColumn = "function_id";
+ break;
+ case USER:
+ table = UserMetaMapper.USER_TABLE_NAME;
+ idColumn = "user_id";
+ break;
+ case GROUP:
+ table = GroupMetaMapper.GROUP_TABLE_NAME;
+ idColumn = "group_id";
+ break;
+ case ROLE:
+ table = RoleMetaMapper.ROLE_TABLE_NAME;
+ idColumn = "role_id";
+ break;
+ case TAG:
+ table = TagMetaMapper.TAG_TABLE_NAME;
+ idColumn = "tag_id";
+ break;
+ case POLICY:
+ table = PolicyMetaMapper.POLICY_META_TABLE_NAME;
+ idColumn = "policy_id";
+ break;
+ case JOB_TEMPLATE:
+ table = JobTemplateMetaMapper.TABLE_NAME;
+ idColumn = "job_template_id";
+ break;
+ case JOB:
+ table = JobMetaMapper.TABLE_NAME;
+ idColumn = "job_run_id";
+ break;
+ default:
+ throw new IllegalArgumentException("Unsupported live endpoint type: "
+ type);
+ }
+
+ String databaseId =
+ SqlSessionFactoryHelper.getInstance()
+ .getSqlSessionFactory()
+ .getConfiguration()
+ .getDatabaseId();
+ JDBCBackend.JDBCBackendType backend =
JDBCBackend.JDBCBackendType.fromString(databaseId);
+ String lockClause;
+ switch (backend) {
+ case H2:
+ lockClause = " FOR UPDATE";
+ break;
+ case MYSQL:
+ lockClause = " LOCK IN SHARE MODE";
+ break;
+ case POSTGRESQL:
+ lockClause = " FOR SHARE";
+ break;
+ default:
+ throw new IllegalArgumentException("Unsupported JDBC backend: " +
backend);
+ }
+ return "SELECT "
+ + idColumn
+ + " FROM "
+ + table
+ + " WHERE "
+ + idColumn
+ + " = #{id} AND deleted_at = 0"
+ + lockClause;
+ }
+}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/DefaultMapperPackageProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/DefaultMapperPackageProvider.java
index c05336d88c..ae8773fd75 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/DefaultMapperPackageProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/DefaultMapperPackageProvider.java
@@ -30,6 +30,7 @@ import
org.apache.gravitino.storage.relational.mapper.GroupMetaMapper;
import org.apache.gravitino.storage.relational.mapper.GroupRoleRelMapper;
import org.apache.gravitino.storage.relational.mapper.JobMetaMapper;
import org.apache.gravitino.storage.relational.mapper.JobTemplateMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.LiveEndpointMapper;
import org.apache.gravitino.storage.relational.mapper.MetalakeMetaMapper;
import org.apache.gravitino.storage.relational.mapper.ModelMetaMapper;
import
org.apache.gravitino.storage.relational.mapper.ModelVersionAliasRelMapper;
@@ -72,6 +73,7 @@ public class DefaultMapperPackageProvider implements
MapperPackageProvider {
GroupRoleRelMapper.class,
JobMetaMapper.class,
JobTemplateMetaMapper.class,
+ LiveEndpointMapper.class,
MetalakeMetaMapper.class,
ModelMetaMapper.class,
ModelVersionAliasRelMapper.class,
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/service/LiveEndpointService.java
b/core/src/main/java/org/apache/gravitino/storage/relational/service/LiveEndpointService.java
new file mode 100644
index 0000000000..9f90b60f59
--- /dev/null
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/service/LiveEndpointService.java
@@ -0,0 +1,131 @@
+/*
+ * 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.storage.relational.service;
+
+import com.google.common.base.Preconditions;
+import java.util.Arrays;
+import org.apache.gravitino.Entity;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.exceptions.NoSuchEntityException;
+import org.apache.gravitino.meta.NamespacedEntityId;
+import org.apache.gravitino.storage.relational.RelationalEntityStoreIdResolver;
+import org.apache.gravitino.storage.relational.mapper.LiveEndpointMapper;
+import org.apache.gravitino.storage.relational.session.SqlSessions;
+import org.apache.gravitino.storage.relational.utils.SessionUtils;
+import org.apache.ibatis.session.SqlSession;
+
+/**
+ * Fences metadata-object endpoints while a relation write is in progress.
+ *
+ * <p>Statistic writers take a schema lock of their own today and adopt this
fence in #13177, so
+ * nothing here covers them yet.
+ */
+public final class LiveEndpointService {
+
+ private LiveEndpointService() {}
+
+ /**
+ * Locks the observed endpoint and its ancestors in hierarchy order, then
verifies that the name
+ * still resolves to the same ID chain. The caller must keep the current
transaction open until
+ * its dependent write completes. For a column, the table lock also fences
column changes.
+ *
+ * @param identifier the originally observed endpoint name
+ * @param type the endpoint type
+ * @param observed the originally observed endpoint and namespace IDs
+ */
+ public static void lockLiveEndpoint(
+ NameIdentifier identifier, Entity.EntityType type, NamespacedEntityId
observed) {
+ Preconditions.checkState(
+ SessionUtils.isInTransaction(), "A transaction is required to lock a
live endpoint");
+ long[] namespaceIds = observed.namespaceIds();
+ if (type == Entity.EntityType.METALAKE) {
+ lock(identifier, type, observed.entityId());
+ } else {
+ checkNamespaceDepth(type, namespaceIds, 1);
+ lock(identifier, Entity.EntityType.METALAKE, namespaceIds[0]);
+ if (type == Entity.EntityType.CATALOG) {
+ lock(identifier, type, observed.entityId());
+ } else if (namespaceIds.length > 1) {
+ lock(identifier, Entity.EntityType.CATALOG, namespaceIds[1]);
+ if (type == Entity.EntityType.SCHEMA) {
+ lock(identifier, type, observed.entityId());
+ } else {
+ checkNamespaceDepth(type, namespaceIds, 3);
+ lock(identifier, Entity.EntityType.SCHEMA, namespaceIds[2]);
+ if (type == Entity.EntityType.COLUMN) {
+ checkNamespaceDepth(type, namespaceIds, 4);
+ lock(identifier, Entity.EntityType.TABLE, namespaceIds[3]);
+ } else {
+ lock(identifier, type, observed.entityId());
+ }
+ }
+ } else {
+ lock(identifier, type, observed.entityId());
+ }
+ }
+
+ // An earlier name lookup in this transaction may be in MyBatis's
first-level cache. The
+ // locking reads above must be followed by a database read of the current
name-to-ID chain.
+ SqlSession session = SqlSessions.getSqlSession();
+ try {
+ session.clearCache();
+ } finally {
+ SqlSessions.closeSqlSession();
+ }
+ NamespacedEntityId current =
+ new RelationalEntityStoreIdResolver().getEntityIds(identifier, type);
+ if (!Arrays.equals(current.fullIds(), observed.fullIds())) {
+ throw missing(identifier, type);
+ }
+ }
+
+ /**
+ * Rejects a namespace that is too shallow for the walk below it.
+ *
+ * <p>The walk indexes the namespace by position, so a chain shorter than
its type implies would
+ * otherwise leave the endpoint partly unlocked or fail with an index error
that names nothing.
+ *
+ * @param type the endpoint type being locked
+ * @param namespaceIds the observed namespace ID chain
+ * @param required the number of namespace IDs the walk is about to rely on
+ */
+ private static void checkNamespaceDepth(
+ Entity.EntityType type, long[] namespaceIds, int required) {
+ Preconditions.checkArgument(
+ namespaceIds.length >= required,
+ "A %s endpoint needs at least %s namespace IDs, got %s",
+ type,
+ required,
+ namespaceIds.length);
+ }
+
+ private static void lock(NameIdentifier identifier, Entity.EntityType type,
long id) {
+ Long locked =
+ SessionUtils.getWithoutCommit(
+ LiveEndpointMapper.class, mapper -> mapper.lockLiveEndpoint(type,
id));
+ if (locked == null) {
+ throw missing(identifier, type);
+ }
+ }
+
+ private static NoSuchEntityException missing(NameIdentifier identifier,
Entity.EntityType type) {
+ return new NoSuchEntityException(
+ NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
type.name().toLowerCase(), identifier);
+ }
+}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/service/PolicyTagRelService.java
b/core/src/main/java/org/apache/gravitino/storage/relational/service/PolicyTagRelService.java
index 886fd565e6..1e44b956f0 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/service/PolicyTagRelService.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/service/PolicyTagRelService.java
@@ -155,6 +155,11 @@ public class PolicyTagRelService {
RelationEdgeTarget[] targetsToRemove)
throws IOException {
String metalake = tagIdentifier.namespace().level(0);
+ NameIdentifier metalakeIdentifier = NameIdentifier.of(metalake);
+ LiveEndpointService.lockLiveEndpoint(
+ metalakeIdentifier,
+ Entity.EntityType.METALAKE,
+ EntityIdService.getEntityIds(metalakeIdentifier,
Entity.EntityType.METALAKE));
TagPO tagPO = lockTag(tagIdentifier);
long tagId = tagPO.getTagId();
Map<String, Long> policyIds = resolvePolicyIds(metalake, targetsToAdd,
targetsToRemove);
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/service/RoleMetaService.java
b/core/src/main/java/org/apache/gravitino/storage/relational/service/RoleMetaService.java
index 318ead56b6..040dcb1dfc 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/service/RoleMetaService.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/service/RoleMetaService.java
@@ -20,12 +20,14 @@ package org.apache.gravitino.storage.relational.service;
import static
org.apache.gravitino.metrics.source.MetricsSource.GRAVITINO_RELATIONAL_STORE_METRIC_NAME;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import java.io.IOException;
import java.util.Collection;
import java.util.Collections;
+import java.util.Comparator;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
@@ -45,6 +47,7 @@ import org.apache.gravitino.authorization.SecurableObject;
import org.apache.gravitino.exceptions.NoSuchEntityException;
import org.apache.gravitino.exceptions.NoSuchMetadataObjectException;
import org.apache.gravitino.exceptions.NoSuchRoleException;
+import org.apache.gravitino.meta.NamespacedEntityId;
import org.apache.gravitino.meta.RoleEntity;
import org.apache.gravitino.meta.UserEntity;
import org.apache.gravitino.metrics.Monitored;
@@ -173,15 +176,19 @@ public class RoleMetaService {
RolePO.Builder builder =
RolePO.builder().withMetalakeId(metalakePO.getMetalakeId());
RolePO rolePO = POConverters.initializeRolePOWithVersion(roleEntity,
builder);
List<SecurableObjectPO> securableObjectPOs = Lists.newArrayList();
+ List<EndpointLock> endpointLocks = Lists.newArrayList();
for (SecurableObject object : roleEntity.securableObjects()) {
SecurableObjectPO.Builder objectBuilder =
POConverters.initializeSecurablePOBuilderWithVersion(
roleEntity.id(), object, getType(object));
NameIdentifier identifier = MetadataObjectUtil.toEntityIdent(metalake,
object);
Entity.EntityType entityType =
MetadataObjectUtil.toEntityType(object.type());
-
objectBuilder.withMetadataObjectId(EntityIdService.getEntityId(identifier,
entityType));
+ NamespacedEntityId observed = EntityIdService.getEntityIds(identifier,
entityType);
+ objectBuilder.withMetadataObjectId(observed.entityId());
+ endpointLocks.add(new EndpointLock(identifier, entityType, observed));
securableObjectPOs.add(objectBuilder.build());
}
+ Collections.sort(endpointLocks);
// The role row is written before its securable objects. Concurrent
overwrites then contend
// on the role row first and replace the child rows only after they are
serialized. The role
@@ -198,6 +205,7 @@ public class RoleMetaService {
mapper.insertRoleMeta(rolePO);
}
}),
+ () -> endpointLocks.forEach(Runnable::run),
() ->
SessionUtils.doWithoutCommit(
SecurableObjectMapper.class,
@@ -251,10 +259,33 @@ public class RoleMetaService {
List<SecurableObjectPO> deleteSecurableObjectPOs =
toSecurableObjectPOs(deleteObjects, oldRoleEntity, metalake);
- List<SecurableObjectPO> insertSecurableObjectPOs =
- toSecurableObjectPOs(insertObjects, oldRoleEntity, metalake);
+ List<SecurableObjectPO> insertSecurableObjectPOs = Lists.newArrayList();
+ List<EndpointLock> endpointLocks = Lists.newArrayList();
+ for (SecurableObject object : insertObjects) {
+ NameIdentifier objectIdentifier =
MetadataObjectUtil.toEntityIdent(metalake, object);
+ Entity.EntityType objectType =
MetadataObjectUtil.toEntityType(object.type());
+ NamespacedEntityId observed;
+ try {
+ observed = EntityIdService.getEntityIds(objectIdentifier,
objectType);
+ } catch (NoSuchEntityException nse) {
+ throw new NoSuchMetadataObjectException(
+ nse, "Metadata object %s type %s doesn't exist",
object.fullName(), object.type());
+ }
+ insertSecurableObjectPOs.add(
+ POConverters.initializeSecurablePOBuilderWithVersion(
+ oldRoleEntity.id(), object, getType(object))
+ .withMetadataObjectId(observed.entityId())
+ .build());
+ endpointLocks.add(new EndpointLock(objectIdentifier, objectType,
observed));
+ }
+ Collections.sort(endpointLocks);
SessionUtils.doMultipleWithCommit(
+ () ->
+ LiveEndpointService.lockLiveEndpoint(
+ NameIdentifier.of(metalake),
+ Entity.EntityType.METALAKE,
+ new NamespacedEntityId(metalakeId)),
() -> {
int updated =
SessionUtils.getWithoutCommit(
@@ -266,6 +297,10 @@ public class RoleMetaService {
throw roleWriteFailure(identifier, rolePO);
}
},
+ // Target deletion locks the endpoint before cleaning its privilege
relations. Fence
+ // every added endpoint before touching old relations to keep the
same lock order when
+ // replacing the privileges on an existing endpoint.
+ () -> endpointLocks.forEach(Runnable::run),
() -> {
if (deleteSecurableObjectPOs.isEmpty()) {
return;
@@ -590,4 +625,50 @@ public class RoleMetaService {
private static String getType(SecurableObject securableObject) {
return securableObject.type().name();
}
+
+ @VisibleForTesting
+ static final class EndpointLock implements Runnable,
Comparable<EndpointLock> {
+ private static final Comparator<EndpointLock> ORDER =
+ Comparator.comparingInt((EndpointLock lock) -> lockOrder(lock.type))
+ .thenComparingLong(lock -> lock.observed.entityId());
+
+ private final NameIdentifier identifier;
+ private final Entity.EntityType type;
+ private final NamespacedEntityId observed;
+
+ EndpointLock(NameIdentifier identifier, Entity.EntityType type,
NamespacedEntityId observed) {
+ this.identifier = identifier;
+ this.type = type;
+ this.observed = observed;
+ }
+
+ @Override
+ public void run() {
+ LiveEndpointService.lockLiveEndpoint(identifier, type, observed);
+ }
+
+ @Override
+ public int compareTo(EndpointLock other) {
+ return ORDER.compare(this, other);
+ }
+
+ @Override
+ public String toString() {
+ // A failed ordering assertion is about which endpoint sits where, so
print that rather
+ // than an identity hash.
+ return type + "(" + observed.entityId() + ")";
+ }
+
+ private static int lockOrder(Entity.EntityType type) {
+ // Relation writers take ordinary metadata endpoints before tags and
policies, and
+ // PolicyTagRelService locks tags before policies.
+ if (type == Entity.EntityType.TAG) {
+ return Integer.MAX_VALUE - 1;
+ }
+ if (type == Entity.EntityType.POLICY) {
+ return Integer.MAX_VALUE;
+ }
+ return type.ordinal();
+ }
+ }
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/service/TagMetaService.java
b/core/src/main/java/org/apache/gravitino/storage/relational/service/TagMetaService.java
index cbf8b2e3a6..41d665307d 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/service/TagMetaService.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/service/TagMetaService.java
@@ -49,6 +49,7 @@ import org.apache.gravitino.exceptions.NoSuchEntityException;
import org.apache.gravitino.exceptions.NoSuchTagException;
import org.apache.gravitino.json.JsonUtils;
import org.apache.gravitino.meta.GenericEntity;
+import org.apache.gravitino.meta.NamespacedEntityId;
import org.apache.gravitino.meta.TagEntity;
import org.apache.gravitino.metrics.Monitored;
import org.apache.gravitino.storage.relational.mapper.MetalakeMetaMapper;
@@ -395,7 +396,8 @@ public class TagMetaService {
String metalake = objectIdent.namespace().level(0);
try {
- Long metadataObjectId = EntityIdService.getEntityId(objectIdent,
objectType);
+ NamespacedEntityId observedEndpoint =
EntityIdService.getEntityIds(objectIdent, objectType);
+ Long metadataObjectId = observedEndpoint.entityId();
List<TagValue> tagValuesToAdd = new
ArrayList<>(Arrays.asList(nullToEmpty(tagsToAdd)));
List<TagValue> tagValuesToRemove = new
ArrayList<>(Arrays.asList(nullToEmpty(tagsToRemove)));
Set<TagValue> commonTagValues = new LinkedHashSet<>(tagValuesToAdd);
@@ -403,12 +405,26 @@ public class TagMetaService {
tagValuesToAdd.removeAll(commonTagValues);
tagValuesToRemove.removeAll(commonTagValues);
+ NameIdentifier metalakeIdent = NameIdentifier.of(metalake);
+ long metalakeId =
+ objectType == Entity.EntityType.METALAKE
+ ? observedEndpoint.entityId()
+ : observedEndpoint.namespaceIds()[0];
+ LiveEndpointService.lockLiveEndpoint(
+ metalakeIdent, Entity.EntityType.METALAKE, new
NamespacedEntityId(metalakeId));
+ if (objectType != Entity.EntityType.POLICY) {
+ LiveEndpointService.lockLiveEndpoint(objectIdent, objectType,
observedEndpoint);
+ }
+
List<String> tagNamesToUpdate = tagNamesToUpdate(tagValuesToAdd,
tagValuesToRemove);
List<TagPO> tagPOsToUpdate =
tagNamesToUpdate.isEmpty()
? Collections.emptyList()
: getTagPOsByMetalakeAndNames(metalake, tagNamesToUpdate);
tagPOsToUpdate = lockTagsForAssignment(tagPOsToUpdate);
+ if (objectType == Entity.EntityType.POLICY) {
+ LiveEndpointService.lockLiveEndpoint(objectIdent, objectType,
observedEndpoint);
+ }
Map<String, TagPO> tagPOsByName = tagPOsByName(tagPOsToUpdate);
List<TagPO> currentTagPOs =
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestLiveEndpointService.java
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestLiveEndpointService.java
new file mode 100644
index 0000000000..27f758cea7
--- /dev/null
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestLiveEndpointService.java
@@ -0,0 +1,171 @@
+/*
+ * 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.storage.relational.service;
+
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.apache.gravitino.Entity;
+import org.apache.gravitino.Namespace;
+import org.apache.gravitino.exceptions.NoSuchEntityException;
+import org.apache.gravitino.meta.CatalogEntity;
+import org.apache.gravitino.meta.NamespacedEntityId;
+import org.apache.gravitino.meta.TableEntity;
+import org.apache.gravitino.storage.relational.RelationalEntityStoreIdResolver;
+import org.apache.gravitino.storage.relational.TestJDBCBackend;
+import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper;
+import org.apache.gravitino.storage.relational.utils.SessionUtils;
+import org.apache.ibatis.session.SqlSession;
+import org.junit.jupiter.api.TestTemplate;
+
+/** Tests database-backed liveness and identity checks for relation endpoints.
*/
+public class TestLiveEndpointService extends TestJDBCBackend {
+
+ @TestTemplate
+ public void testRejectsRecreatedCatalogWithSameName() throws IOException {
+ String metalake = "endpoint_fence_metalake";
+ String catalogName = "endpoint_fence_catalog";
+ createAndInsertMakeLake(metalake);
+ CatalogEntity oldCatalog = createAndInsertCatalog(metalake, catalogName);
+ NamespacedEntityId observed =
+ EntityIdService.getEntityIds(oldCatalog.nameIdentifier(),
Entity.EntityType.CATALOG);
+
+ assertThrows(
+ IllegalStateException.class,
+ () ->
+ LiveEndpointService.lockLiveEndpoint(
+ oldCatalog.nameIdentifier(), Entity.EntityType.CATALOG,
observed));
+ SessionUtils.doMultipleWithCommit(
+ () ->
+ LiveEndpointService.lockLiveEndpoint(
+ oldCatalog.nameIdentifier(), Entity.EntityType.CATALOG,
observed));
+
+
CatalogMetaService.getInstance().deleteCatalog(oldCatalog.nameIdentifier(),
true);
+ CatalogEntity replacement = createAndInsertCatalog(metalake, catalogName);
+ assertNotEquals(oldCatalog.id(), replacement.id());
+ assertThrows(
+ NoSuchEntityException.class,
+ () ->
+ SessionUtils.doMultipleWithCommit(
+ () ->
+ LiveEndpointService.lockLiveEndpoint(
+ oldCatalog.nameIdentifier(),
Entity.EntityType.CATALOG, observed)));
+ }
+
+ @TestTemplate
+ public void testLocksSchemaScopedTarget() throws IOException {
+ String metalake = "endpoint_fence_table_metalake";
+ String catalog = "endpoint_fence_table_catalog";
+ String schema = "endpoint_fence_table_schema";
+ createAndInsertMakeLake(metalake);
+ createAndInsertCatalog(metalake, catalog);
+ createAndInsertSchema(metalake, catalog, schema);
+ TableEntity table =
+ createAndInsertTableEntity(Namespace.of(metalake, catalog, schema),
"endpoint_table");
+ NamespacedEntityId observed =
+ EntityIdService.getEntityIds(table.nameIdentifier(),
Entity.EntityType.TABLE);
+
+ SessionUtils.doMultipleWithCommit(
+ () ->
+ LiveEndpointService.lockLiveEndpoint(
+ table.nameIdentifier(), Entity.EntityType.TABLE, observed));
+ }
+
+ @TestTemplate
+ public void testRejectsTargetDeletedAfterIdResolution() throws Exception {
+ String metalake = "endpoint_race_metalake";
+ createAndInsertMakeLake(metalake);
+ CatalogEntity catalog = createAndInsertCatalog(metalake,
"endpoint_race_catalog");
+ CountDownLatch resolved = new CountDownLatch(1);
+ CountDownLatch continueWrite = new CountDownLatch(1);
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+ Future<Throwable> write =
+ executor.submit(
+ () -> {
+ NamespacedEntityId observed =
+ EntityIdService.getEntityIds(
+ catalog.nameIdentifier(), Entity.EntityType.CATALOG);
+ resolved.countDown();
+ if (!continueWrite.await(30, TimeUnit.SECONDS)) {
+ throw new IllegalStateException("Timed out waiting for
endpoint deletion");
+ }
+ try {
+ SessionUtils.doMultipleWithCommit(
+ () ->
+ LiveEndpointService.lockLiveEndpoint(
+ catalog.nameIdentifier(),
Entity.EntityType.CATALOG, observed));
+ return null;
+ } catch (Throwable failure) {
+ return failure;
+ }
+ });
+ assertTrue(resolved.await(30, TimeUnit.SECONDS));
+ CatalogMetaService.getInstance().deleteCatalog(catalog.nameIdentifier(),
true);
+ continueWrite.countDown();
+ assertInstanceOf(NoSuchEntityException.class, write.get(30,
TimeUnit.SECONDS));
+ } finally {
+ continueWrite.countDown();
+ executor.shutdownNow();
+ }
+ }
+
+ @TestTemplate
+ public void testRechecksNameAfterTransactionLocalLookup() throws Exception {
+ String metalake = "endpoint_rename_metalake";
+ createAndInsertMakeLake(metalake);
+ CatalogEntity catalog = createAndInsertCatalog(metalake,
"endpoint_rename_catalog");
+ NamespacedEntityId observed =
+ EntityIdService.getEntityIds(catalog.nameIdentifier(),
Entity.EntityType.CATALOG);
+
+ SessionUtils.doMultipleWithCommit(
+ () -> {
+ new RelationalEntityStoreIdResolver()
+ .getEntityIds(catalog.nameIdentifier(),
Entity.EntityType.CATALOG);
+ try (SqlSession otherSession =
+
SqlSessionFactoryHelper.getInstance().getSqlSessionFactory().openSession(true);
+ Connection connection = otherSession.getConnection();
+ PreparedStatement rename =
+ connection.prepareStatement(
+ "UPDATE catalog_meta SET catalog_name = ? WHERE
catalog_id = ?")) {
+ rename.setString(1, "endpoint_renamed_catalog");
+ rename.setLong(2, catalog.id());
+ assertTrue(rename.executeUpdate() == 1);
+ } catch (SQLException e) {
+ throw new RuntimeException(e);
+ }
+ assertThrows(
+ NoSuchEntityException.class,
+ () ->
+ LiveEndpointService.lockLiveEndpoint(
+ catalog.nameIdentifier(), Entity.EntityType.CATALOG,
observed));
+ });
+ }
+}
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestRoleMetaService.java
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestRoleMetaService.java
index f955bdd336..e140139045 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestRoleMetaService.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestRoleMetaService.java
@@ -27,6 +27,7 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;
import java.io.IOException;
+import java.io.UncheckedIOException;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
@@ -37,10 +38,20 @@ import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import java.util.function.Function;
import org.apache.gravitino.Entity;
import org.apache.gravitino.EntityAlreadyExistsException;
+import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.Namespace;
+import org.apache.gravitino.RelationEdgeTarget;
+import org.apache.gravitino.RelationUpdate;
+import org.apache.gravitino.SupportsRelationOperations;
import org.apache.gravitino.authorization.AuthorizationUtils;
import org.apache.gravitino.authorization.Privileges;
import org.apache.gravitino.authorization.SecurableObject;
@@ -53,31 +64,382 @@ import org.apache.gravitino.meta.BaseMetalake;
import org.apache.gravitino.meta.CatalogEntity;
import org.apache.gravitino.meta.FilesetEntity;
import org.apache.gravitino.meta.GroupEntity;
+import org.apache.gravitino.meta.NamespacedEntityId;
+import org.apache.gravitino.meta.PolicyEntity;
import org.apache.gravitino.meta.RoleEntity;
import org.apache.gravitino.meta.TableEntity;
+import org.apache.gravitino.meta.TagEntity;
import org.apache.gravitino.meta.TopicEntity;
import org.apache.gravitino.meta.UserEntity;
import org.apache.gravitino.storage.RandomIdGenerator;
import org.apache.gravitino.storage.relational.TestJDBCBackend;
+import org.apache.gravitino.storage.relational.mapper.CatalogMetaMapper;
import org.apache.gravitino.storage.relational.mapper.GroupMetaMapper;
import org.apache.gravitino.storage.relational.mapper.MetalakeMetaMapper;
import org.apache.gravitino.storage.relational.mapper.RoleMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.TagMetaMapper;
import org.apache.gravitino.storage.relational.mapper.UserMetaMapper;
+import org.apache.gravitino.storage.relational.po.CatalogPO;
import org.apache.gravitino.storage.relational.po.GroupPO;
import org.apache.gravitino.storage.relational.po.MetalakePO;
import org.apache.gravitino.storage.relational.po.RolePO;
+import org.apache.gravitino.storage.relational.po.TagPO;
import org.apache.gravitino.storage.relational.po.UserPO;
import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper;
import org.apache.gravitino.storage.relational.utils.SessionUtils;
import org.apache.gravitino.utils.CollectionUtils;
+import org.apache.gravitino.utils.NamespaceUtil;
import org.apache.ibatis.session.SqlSession;
import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Assumptions;
import org.junit.jupiter.api.TestTemplate;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
class TestRoleMetaService extends TestJDBCBackend {
private static final String METALAKE_NAME = "metalake_for_role_test";
+ /** Verifies privilege replacement locks its endpoint before deleting the
old relation. */
+ @TestTemplate
+ public void testPrivilegeUpdateDoesNotDeadlockWithTargetDelete() throws
Exception {
+ createAndInsertMakeLake(METALAKE_NAME);
+ CatalogEntity catalog = createAndInsertCatalog(METALAKE_NAME,
"role_update_delete_catalog");
+ RoleEntity role =
+ createRoleEntity(
+ RandomIdGenerator.INSTANCE.nextId(),
+ AuthorizationUtils.ofRoleNamespace(METALAKE_NAME),
+ "role_update_delete",
+ AUDIT_INFO,
+ catalog.name());
+ RoleMetaService service = RoleMetaService.getInstance();
+ service.insertRole(role, false);
+ RolePO beforeUpdate = getRolePO(role.name());
+ CountDownLatch targetDeleted = new CountDownLatch(1);
+ CountDownLatch endpointLockRequested = new CountDownLatch(1);
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+ try {
+ Future<Boolean> deletion =
+ executor.submit(
+ () -> {
+ // Pause the real delete after its catalog CAS, before
relation cleanup. Static
+ // mocks are thread-local; every intercepted operation still
executes its SQL.
+ try (MockedStatic<OccWriteSupport> ignored =
+ Mockito.mockStatic(
+ OccWriteSupport.class,
+ invocation -> {
+ Object result = invocation.callRealMethod();
+ if
(invocation.getMethod().getName().equals("deleteWithVersion")) {
+ targetDeleted.countDown();
+ assertTrue(endpointLockRequested.await(30,
TimeUnit.SECONDS));
+ }
+ return result;
+ })) {
+ return CatalogMetaService.getInstance()
+ .deleteCatalog(catalog.nameIdentifier(), false);
+ }
+ });
+ assertTrue(targetDeleted.await(30, TimeUnit.SECONDS));
+ Future<Throwable> update =
+ executor.submit(
+ () -> {
+ try (MockedStatic<LiveEndpointService> ignored =
+ Mockito.mockStatic(
+ LiveEndpointService.class,
+ invocation -> {
+ if
(invocation.getMethod().getName().equals("lockLiveEndpoint")
+ && invocation.getArgument(1) ==
Entity.EntityType.CATALOG) {
+ // The old implementation has already locked the
old relation here.
+ // Cleanup then waits for that relation while the
fence waits for the
+ // catalog. The fixed implementation has not
touched the relation yet.
+ endpointLockRequested.countDown();
+ }
+ return invocation.callRealMethod();
+ })) {
+ service.updateRole(
+ role.nameIdentifier(),
+ (RoleEntity current) ->
+ createRoleEntity(
+ current.id(),
+ current.namespace(),
+ current.name(),
+ current.auditInfo(),
+ List.of(
+ SecurableObjects.ofCatalog(
+ catalog.name(),
+ List.of(
+ Privileges.UseCatalog.allow(),
+ Privileges.CreateSchema.allow()))),
+ ImmutableMap.of("update", "must-roll-back")));
+ return null;
+ } catch (Throwable failure) {
+ return failure;
+ }
+ });
+ assertTrue(deletion.get(30, TimeUnit.SECONDS));
+ Throwable failure = update.get(30, TimeUnit.SECONDS);
+ Assertions.assertInstanceOf(NoSuchEntityException.class, failure,
String.valueOf(failure));
+ } finally {
+ endpointLockRequested.countDown();
+ executor.shutdownNow();
+ assertTrue(executor.awaitTermination(30, TimeUnit.SECONDS));
+ }
+
+ // The failed grant must roll back the role CAS and properties. Deletion
must still remove
+ // the old relation, and no replacement relation may survive for the
deleted endpoint.
+ assertEquals(beforeUpdate, getRolePO(role.name()));
+
assertTrue(RoleMetaService.listSecurableObjectsByRoleId(role.id()).isEmpty());
+
assertTrue(service.getRoleByIdentifier(role.nameIdentifier()).securableObjects().isEmpty());
+ Assertions.assertThrows(
+ NoSuchEntityException.class,
+ () ->
CatalogMetaService.getInstance().getCatalogByIdentifier(catalog.nameIdentifier()));
+ }
+
+ /** Verifies role privilege writes share the tag-before-policy order of
policy relations. */
+ @TestTemplate
+ public void testRoleAndPolicyTagWritesUseSameEndpointLockOrder() throws
Exception {
+ createAndInsertMakeLake(METALAKE_NAME);
+ TagEntity tag = createAndInsertTagEntity("lock_order_tag", "",
METALAKE_NAME);
+ PolicyEntity policy =
+ createPolicy(
+ RandomIdGenerator.INSTANCE.nextId(),
+ NamespaceUtil.ofPolicy(METALAKE_NAME),
+ "lock_order_policy",
+ AUDIT_INFO);
+ backend.insert(policy, false);
+ RoleEntity role =
+ createRoleEntity(
+ RandomIdGenerator.INSTANCE.nextId(),
+ AuthorizationUtils.ofRoleNamespace(METALAKE_NAME),
+ "lock_order_role",
+ AUDIT_INFO,
+ List.of(
+ SecurableObjects.ofPolicy(policy.name(),
List.of(Privileges.ApplyPolicy.allow())),
+ SecurableObjects.ofTag(tag.name(),
List.of(Privileges.ApplyTag.allow()))),
+ Collections.emptyMap());
+ TagPO observedTag =
+ SessionUtils.getWithoutCommit(
+ TagMetaMapper.class,
+ mapper -> mapper.selectTagMetaByMetalakeAndName(METALAKE_NAME,
tag.name()));
+ CountDownLatch tagLocked = new CountDownLatch(1);
+ CountDownLatch allowRelationWrite = new CountDownLatch(1);
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+ Future<Throwable> relation =
+ executor.submit(
+ () -> {
+ try {
+ SessionUtils.doMultipleWithCommit(
+ () -> {
+ NameIdentifier metalakeIdentifier =
NameIdentifier.of(METALAKE_NAME);
+ LiveEndpointService.lockLiveEndpoint(
+ metalakeIdentifier,
+ Entity.EntityType.METALAKE,
+ EntityIdService.getEntityIds(
+ metalakeIdentifier, Entity.EntityType.METALAKE));
+ TagMetaService.lockTags(List.of(observedTag));
+ tagLocked.countDown();
+ try {
+ assertTrue(allowRelationWrite.await(30,
TimeUnit.SECONDS));
+ backend.updateEntityRelations(
+ RelationUpdate.of(
+ SupportsRelationOperations.Type.POLICY_TAG_REL,
+ tag.nameIdentifier(),
+ Entity.EntityType.TAG,
+ new RelationEdgeTarget[] {
+ RelationEdgeTarget.of(
+ policy.nameIdentifier(),
Entity.EntityType.POLICY, null)
+ },
+ new RelationEdgeTarget[0]));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ });
+ return null;
+ } catch (Throwable failure) {
+ return failure;
+ }
+ });
+ try {
+ assertTrue(tagLocked.await(30, TimeUnit.SECONDS));
+ Future<Throwable> insertion =
+ executor.submit(
+ () -> {
+ try {
+ RoleMetaService.getInstance().insertRole(role, false);
+ return null;
+ } catch (Throwable failure) {
+ return failure;
+ }
+ });
+ Assertions.assertThrows(
+ TimeoutException.class, () -> insertion.get(500,
TimeUnit.MILLISECONDS));
+ allowRelationWrite.countDown();
+ Assertions.assertNull(relation.get(30, TimeUnit.SECONDS));
+ Assertions.assertNull(insertion.get(30, TimeUnit.SECONDS));
+ assertEquals(role,
RoleMetaService.getInstance().getRoleByIdentifier(role.nameIdentifier()));
+ } finally {
+ allowRelationWrite.countDown();
+ executor.shutdownNow();
+ }
+ }
+
+ /** Verifies role updates lock the metalake before the role row on H2. */
+ @TestTemplate
+ public void testRoleUpdateLocksMetalakeBeforeRoleOnH2() throws Exception {
+ Assumptions.assumeTrue("h2".equalsIgnoreCase(backendType));
+ createAndInsertMakeLake(METALAKE_NAME);
+ createAndInsertCatalog(METALAKE_NAME, "role_overwrite_catalog");
+ TagEntity tag = createAndInsertTagEntity("role_update_tag", "",
METALAKE_NAME);
+ RoleEntity role =
+ createRoleEntity(
+ RandomIdGenerator.INSTANCE.nextId(),
+ AuthorizationUtils.ofRoleNamespace(METALAKE_NAME),
+ "role_overwrite_order",
+ AUDIT_INFO,
+ "role_overwrite_catalog");
+ RoleMetaService.getInstance().insertRole(role, false);
+ NameIdentifier metalakeIdentifier = NameIdentifier.of(METALAKE_NAME);
+ CountDownLatch metalakeLocked = new CountDownLatch(1);
+ CountDownLatch allowRoleLock = new CountDownLatch(1);
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+ Future<Throwable> competingWriter =
+ executor.submit(
+ () -> {
+ try {
+ SessionUtils.doMultipleWithCommit(
+ () -> {
+ LiveEndpointService.lockLiveEndpoint(
+ metalakeIdentifier,
+ Entity.EntityType.METALAKE,
+ EntityIdService.getEntityIds(
+ metalakeIdentifier, Entity.EntityType.METALAKE));
+ metalakeLocked.countDown();
+ try {
+ assertTrue(allowRoleLock.await(30, TimeUnit.SECONDS));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ assertTrue(
+ SessionUtils.getWithoutCommit(
+ RoleMetaMapper.class,
+ mapper ->
mapper.selectRoleMetaByIdForUpdate(role.id()))
+ != null);
+ });
+ return null;
+ } catch (Throwable failure) {
+ return failure;
+ }
+ });
+ try {
+ assertTrue(metalakeLocked.await(30, TimeUnit.SECONDS));
+ Future<Throwable> update =
+ executor.submit(
+ () -> {
+ try {
+ RoleMetaService.getInstance()
+ .updateRole(
+ role.nameIdentifier(),
+ (RoleEntity current) ->
+ createRoleEntity(
+ current.id(),
+ current.namespace(),
+ current.name(),
+ current.auditInfo(),
+ List.of(
+ SecurableObjects.ofCatalog(
+ "role_overwrite_catalog",
+
List.of(Privileges.UseCatalog.allow())),
+ SecurableObjects.ofTag(
+ tag.name(),
List.of(Privileges.ApplyTag.allow()))),
+ Collections.emptyMap()));
+ return null;
+ } catch (Throwable failure) {
+ return failure;
+ }
+ });
+ Assertions.assertThrows(TimeoutException.class, () -> update.get(500,
TimeUnit.MILLISECONDS));
+ allowRoleLock.countDown();
+ Assertions.assertNull(competingWriter.get(30, TimeUnit.SECONDS));
+ Assertions.assertNull(update.get(30, TimeUnit.SECONDS));
+ } finally {
+ allowRoleLock.countDown();
+ executor.shutdownNow();
+ }
+ }
+
+ @TestTemplate
+ public void testPrivilegeInsertRejectsConcurrentTargetDelete() throws
Exception {
+ createAndInsertMakeLake(METALAKE_NAME);
+ CatalogEntity catalog = createAndInsertCatalog(METALAKE_NAME,
"role_target_delete_catalog");
+ RoleEntity role =
+ createRoleEntity(
+ RandomIdGenerator.INSTANCE.nextId(),
+ AuthorizationUtils.ofRoleNamespace(METALAKE_NAME),
+ "role_target_delete",
+ AUDIT_INFO,
+ catalog.name());
+ CatalogPO observed =
+ SessionUtils.getWithoutCommit(
+ CatalogMetaMapper.class, mapper ->
mapper.selectCatalogMetaById(catalog.id()));
+ CountDownLatch deleteWritten = new CountDownLatch(1);
+ CountDownLatch allowDeleteCommit = new CountDownLatch(1);
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+ Future<Throwable> deletion =
+ executor.submit(
+ () -> {
+ try {
+ SessionUtils.doMultipleWithCommit(
+ () -> {
+ int updated =
+ SessionUtils.getWithoutCommit(
+ CatalogMetaMapper.class,
+ mapper ->
+ mapper.softDeleteCatalogMetasByCatalogId(
+ catalog.id(),
observed.getCurrentVersion()));
+ assertEquals(1, updated);
+ deleteWritten.countDown();
+ try {
+ assertTrue(allowDeleteCommit.await(30,
TimeUnit.SECONDS));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ });
+ return null;
+ } catch (Throwable failure) {
+ return failure;
+ }
+ });
+ try {
+ assertTrue(deleteWritten.await(30, TimeUnit.SECONDS));
+ Future<Throwable> insertion =
+ executor.submit(
+ () -> {
+ try {
+ RoleMetaService.getInstance().insertRole(role, false);
+ return null;
+ } catch (Throwable failure) {
+ return failure;
+ }
+ });
+ Assertions.assertThrows(
+ TimeoutException.class, () -> insertion.get(500,
TimeUnit.MILLISECONDS));
+ allowDeleteCommit.countDown();
+ Assertions.assertNull(deletion.get(30, TimeUnit.SECONDS));
+ Assertions.assertInstanceOf(NoSuchEntityException.class,
insertion.get(30, TimeUnit.SECONDS));
+ } finally {
+ allowDeleteCommit.countDown();
+ executor.shutdownNow();
+ }
+ Assertions.assertThrows(
+ NoSuchEntityException.class,
+ () ->
RoleMetaService.getInstance().getRoleByIdentifier(role.nameIdentifier()));
+ }
+
private long queryRoleUpdatedAt(long roleId) {
try (SqlSession sqlSession =
SqlSessionFactoryHelper.getInstance().getSqlSessionFactory().openSession(true))
{
@@ -1404,4 +1766,40 @@ class TestRoleMetaService extends TestJDBCBackend {
throw new RuntimeException("Advance role version failed", e);
}
}
+
+ @TestTemplate
+ public void testEndpointLocksSortIntoTheGlobalLockOrder() {
+ // Every relation writer takes its endpoint locks in one order, or two of
them can wait on each
+ // other: ordinary metadata objects first, then tags, then policies.
TagMetaService and
+ // PolicyTagRelService encode the tail of that order by hand, so this pins
the sort that
+ // RoleMetaService relies on to agree with them.
+ // FUNCTION and MODEL are declared after TAG and POLICY in EntityType, so
sorting on the enum
+ // order alone would take them after a tag or a policy and invert the
order the other two
+ // services use. Those two types are what make this sort more than a
formality.
+ RoleMetaService.EndpointLock policy = endpointLock("a_policy",
Entity.EntityType.POLICY, 10L);
+ RoleMetaService.EndpointLock tag = endpointLock("z_tag",
Entity.EntityType.TAG, 20L);
+ RoleMetaService.EndpointLock function =
+ endpointLock("a_function", Entity.EntityType.FUNCTION, 30L);
+ RoleMetaService.EndpointLock model = endpointLock("a_model",
Entity.EntityType.MODEL, 31L);
+ RoleMetaService.EndpointLock catalogHigherId =
+ endpointLock("c2", Entity.EntityType.CATALOG, 41L);
+ RoleMetaService.EndpointLock catalogLowerId =
+ endpointLock("c1", Entity.EntityType.CATALOG, 40L);
+
+ List<RoleMetaService.EndpointLock> locks =
+ Lists.newArrayList(policy, tag, function, model, catalogHigherId,
catalogLowerId);
+ Collections.sort(locks);
+
+ assertEquals(
+ Lists.newArrayList(catalogLowerId, catalogHigherId, model, function,
tag, policy),
+ locks,
+ "endpoint locks must sort ordinary objects before tags and policies,
and by ID within a"
+ + " type");
+ }
+
+ private static RoleMetaService.EndpointLock endpointLock(
+ String name, Entity.EntityType type, long id) {
+ return new RoleMetaService.EndpointLock(
+ NameIdentifier.of("metalake", name), type, new NamespacedEntityId(id,
1L));
+ }
}
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestTagMetaService.java
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestTagMetaService.java
index 51c0bda9c1..1d79cefa67 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestTagMetaService.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestTagMetaService.java
@@ -72,8 +72,10 @@ import org.apache.gravitino.policy.PolicyContents;
import org.apache.gravitino.rel.types.Types;
import org.apache.gravitino.storage.RandomIdGenerator;
import org.apache.gravitino.storage.relational.TestJDBCBackend;
+import org.apache.gravitino.storage.relational.mapper.CatalogMetaMapper;
import org.apache.gravitino.storage.relational.mapper.MetalakeMetaMapper;
import org.apache.gravitino.storage.relational.mapper.TagMetaMapper;
+import org.apache.gravitino.storage.relational.po.CatalogPO;
import org.apache.gravitino.storage.relational.po.MetalakePO;
import org.apache.gravitino.storage.relational.po.TagPO;
import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper;
@@ -754,6 +756,75 @@ public class TestTagMetaService extends TestJDBCBackend {
tag.id(),
tagMetaService.getTagByIdentifier(renamed.nameIdentifier()).id());
}
+ @TestTemplate
+ public void testTagAssignmentRejectsConcurrentTargetDelete() throws
Exception {
+ createAndInsertMakeLake(METALAKE_NAME);
+ CatalogEntity catalog = createAndInsertCatalog(METALAKE_NAME,
"catalog_tag_target_delete");
+ TagEntity tag =
+ TagEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName("tag_target_delete")
+ .withNamespace(NamespaceUtil.ofTag(METALAKE_NAME))
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ TagMetaService.getInstance().insertTag(tag, false);
+ CatalogPO observed =
+ SessionUtils.getWithoutCommit(
+ CatalogMetaMapper.class, mapper ->
mapper.selectCatalogMetaById(catalog.id()));
+ CountDownLatch deleteWritten = new CountDownLatch(1);
+ CountDownLatch allowDeleteCommit = new CountDownLatch(1);
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+ Future<Throwable> deletion =
+ executor.submit(
+ () -> {
+ try {
+ SessionUtils.doMultipleWithCommit(
+ () -> {
+ assertEquals(
+ Integer.valueOf(1),
+ SessionUtils.getWithoutCommit(
+ CatalogMetaMapper.class,
+ mapper ->
+ mapper.softDeleteCatalogMetasByCatalogId(
+ catalog.id(),
observed.getCurrentVersion())));
+ deleteWritten.countDown();
+ await(allowDeleteCommit);
+ });
+ return null;
+ } catch (Throwable failure) {
+ return failure;
+ }
+ });
+ try {
+ assertTrue(deleteWritten.await(30, TimeUnit.SECONDS));
+ Future<Throwable> assignment =
+ executor.submit(
+ () -> {
+ try {
+ TagMetaService.getInstance()
+ .associateTagsWithMetadataObject(
+ catalog.nameIdentifier(),
+ catalog.type(),
+ new NameIdentifier[] {tag.nameIdentifier()},
+ new NameIdentifier[0]);
+ return null;
+ } catch (Throwable failure) {
+ return failure;
+ }
+ });
+ Assertions.assertThrows(
+ TimeoutException.class, () -> assignment.get(500,
TimeUnit.MILLISECONDS));
+ allowDeleteCommit.countDown();
+ Assertions.assertNull(deletion.get(30, TimeUnit.SECONDS));
+ Assertions.assertInstanceOf(
+ NoSuchEntityException.class, assignment.get(30, TimeUnit.SECONDS));
+ } finally {
+ allowDeleteCommit.countDown();
+ executor.shutdownNow();
+ }
+ assertEquals(0, countActiveTagRel(tag.id()));
+ }
+
@TestTemplate
public void testTagAlterReportsOptimisticLockConflict() throws IOException {
createAndInsertMakeLake(METALAKE_NAME);
