This is an automated email from the ASF dual-hosted git repository.
jerryshao pushed a commit to branch branch-1.3
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/branch-1.3 by this push:
new 4a66d8cce0 [Cherry-pick to branch-1.3] [#13206] improvement(core):
lock the entity node instead of the schema when creating tables, topics, views
and models (#13207) (#13210)
4a66d8cce0 is described below
commit 4a66d8cce0c6a8756d5c96425828b1b1221e763b
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Wed Sep 16 20:38:33 2026 +0800
[Cherry-pick to branch-1.3] [#13206] improvement(core): lock the entity
node instead of the schema when creating tables, topics, views and models
(#13207) (#13210)
**Cherry-pick Information:**
- Original commit: 2928e0f0f31d765360298f379f0439c182c5f02f
- Target branch: `branch-1.3`
- Status: Conflicts resolved
Resolved the model dispatcher test conflict by retaining the new
tree-lock tests and excluding an unrelated main-branch property
assertion helper.
Validation: `spotlessApply` and all 88 tests across the table, topic,
view, and model operation dispatcher test suites passed.
---------
Co-authored-by: Qi Yu <[email protected]>
---
.../catalog/ModelOperationDispatcher.java | 4 +-
.../catalog/TableOperationDispatcher.java | 8 +-
.../catalog/TopicOperationDispatcher.java | 6 +-
.../gravitino/catalog/ViewOperationDispatcher.java | 4 +-
.../catalog/TestModelOperationDispatcher.java | 67 +++++++
.../catalog/TestTableOperationDispatcher.java | 68 +++++++
.../catalog/TestTopicOperationDispatcher.java | 68 +++++++
.../catalog/TestViewOperationDispatcher.java | 77 ++++++++
.../gravitino/catalog/TreeLockTestSupport.java | 214 +++++++++++++++++++++
9 files changed, 510 insertions(+), 6 deletions(-)
diff --git
a/core/src/main/java/org/apache/gravitino/catalog/ModelOperationDispatcher.java
b/core/src/main/java/org/apache/gravitino/catalog/ModelOperationDispatcher.java
index f3d3389098..55a90f6351 100644
---
a/core/src/main/java/org/apache/gravitino/catalog/ModelOperationDispatcher.java
+++
b/core/src/main/java/org/apache/gravitino/catalog/ModelOperationDispatcher.java
@@ -93,9 +93,11 @@ public class ModelOperationDispatcher extends
OperationDispatcher implements Mod
checkAndUpdateProperties(
catalogIdent, properties,
HasPropertyMetadata::modelPropertiesMetadata);
+ // Lock the model node, not the schema, so models in the same schema can
be registered
+ // concurrently. See TableOperationDispatcher#createTable for the
reasoning and trade-off.
Model registeredModel =
TreeLockUtils.doWithTreeLock(
- NameIdentifier.of(ident.namespace().levels()),
+ ident,
LockType.WRITE,
() ->
doWithCatalog(
diff --git
a/core/src/main/java/org/apache/gravitino/catalog/TableOperationDispatcher.java
b/core/src/main/java/org/apache/gravitino/catalog/TableOperationDispatcher.java
index 147c2da544..3cdcd77b43 100644
---
a/core/src/main/java/org/apache/gravitino/catalog/TableOperationDispatcher.java
+++
b/core/src/main/java/org/apache/gravitino/catalog/TableOperationDispatcher.java
@@ -208,8 +208,14 @@ public class TableOperationDispatcher extends
OperationDispatcher implements Tab
NameIdentifier schemaIdent = NameIdentifier.of(ident.namespace().levels());
schemaDispatcher.loadSchema(schemaIdent);
+ // Lock the table node, not the schema, so tables in the same schema can
be created
+ // concurrently. The connector call and the entity-store write otherwise
run under one
+ // schema-wide lock, which serialized all creates in a schema. Ancestors
are still read-locked,
+ // so create keeps excluding dropSchema/createSchema (catalog WRITE) and
rename/drop/import
+ // (schema WRITE); a same-name create, load or alter contends on the table
node.
+ // Trade-off: listTables may briefly observe a table whose creation has
not committed yet.
return TreeLockUtils.doWithTreeLock(
- NameIdentifier.of(ident.namespace().levels()),
+ ident,
LockType.WRITE,
() ->
internalCreateTable(
diff --git
a/core/src/main/java/org/apache/gravitino/catalog/TopicOperationDispatcher.java
b/core/src/main/java/org/apache/gravitino/catalog/TopicOperationDispatcher.java
index 3c43860701..77e136969d 100644
---
a/core/src/main/java/org/apache/gravitino/catalog/TopicOperationDispatcher.java
+++
b/core/src/main/java/org/apache/gravitino/catalog/TopicOperationDispatcher.java
@@ -138,10 +138,10 @@ public class TopicOperationDispatcher extends
OperationDispatcher implements Top
NameIdentifier schemaIdent = NameIdentifier.of(ident.namespace().levels());
schemaDispatcher.loadSchema(schemaIdent);
+ // Lock the topic node, not the schema, so topics in the same schema can
be created
+ // concurrently. See TableOperationDispatcher#createTable for the
reasoning and trade-off.
return TreeLockUtils.doWithTreeLock(
- NameIdentifier.of(ident.namespace().levels()),
- LockType.WRITE,
- () -> internalCreateTopic(ident, comment, dataLayout, properties));
+ ident, LockType.WRITE, () -> internalCreateTopic(ident, comment,
dataLayout, properties));
}
/**
diff --git
a/core/src/main/java/org/apache/gravitino/catalog/ViewOperationDispatcher.java
b/core/src/main/java/org/apache/gravitino/catalog/ViewOperationDispatcher.java
index 844a5e5f34..557819b4dc 100644
---
a/core/src/main/java/org/apache/gravitino/catalog/ViewOperationDispatcher.java
+++
b/core/src/main/java/org/apache/gravitino/catalog/ViewOperationDispatcher.java
@@ -178,8 +178,10 @@ public class ViewOperationDispatcher extends
OperationDispatcher implements View
NameIdentifier schemaIdent = NameIdentifier.of(ident.namespace().levels());
schemaDispatcher.loadSchema(schemaIdent);
+ // Lock the view node, not the schema, so views in the same schema can be
created
+ // concurrently. See TableOperationDispatcher#createTable for the
reasoning and trade-off.
return TreeLockUtils.doWithTreeLock(
- schemaIdent,
+ ident,
LockType.WRITE,
() ->
internalCreateView(
diff --git
a/core/src/test/java/org/apache/gravitino/catalog/TestModelOperationDispatcher.java
b/core/src/test/java/org/apache/gravitino/catalog/TestModelOperationDispatcher.java
index c9f2cb069a..27a4639360 100644
---
a/core/src/test/java/org/apache/gravitino/catalog/TestModelOperationDispatcher.java
+++
b/core/src/test/java/org/apache/gravitino/catalog/TestModelOperationDispatcher.java
@@ -40,6 +40,7 @@ import org.apache.gravitino.exceptions.NoSuchModelException;
import org.apache.gravitino.exceptions.NoSuchModelVersionException;
import org.apache.gravitino.exceptions.NoSuchModelVersionURINameException;
import org.apache.gravitino.lock.LockManager;
+import org.apache.gravitino.lock.LockType;
import org.apache.gravitino.model.Model;
import org.apache.gravitino.model.ModelChange;
import org.apache.gravitino.model.ModelVersion;
@@ -1287,4 +1288,70 @@ public class TestModelOperationDispatcher extends
TestOperationDispatcher {
private String randomModelName() {
return "model_" + UUID.randomUUID().toString().replace("-", "");
}
+
+ @Test
+ public void
testRegisterModelRunsConcurrentlyWithRegisterModelOfAnotherModel() throws
Exception {
+ NameIdentifier schemaIdent = NameIdentifier.of(metalake, catalog,
"schema_model_lock_1");
+ createSchemaForLockTest(schemaIdent);
+
+ // Another in-flight create holds the WRITE lock on its own model node.
+ try (TreeLockTestSupport.HeldLock inFlightCreate =
+ TreeLockTestSupport.HeldLock.acquire(
+ NameIdentifier.of(metalake, catalog, "schema_model_lock_1",
"other_model"),
+ LockType.WRITE)) {
+ TreeLockTestSupport.assertRunsConcurrentlyWith(
+ inFlightCreate,
+ () ->
+ registerModelForLockTest(
+ NameIdentifier.of(metalake, catalog, "schema_model_lock_1",
"model1")));
+ }
+ }
+
+ @Test
+ public void testRegisterModelWaitsForRegisterModelOfSameName() throws
Exception {
+ NameIdentifier schemaIdent = NameIdentifier.of(metalake, catalog,
"schema_model_lock_2");
+ createSchemaForLockTest(schemaIdent);
+ NameIdentifier ident = NameIdentifier.of(metalake, catalog,
"schema_model_lock_2", "model1");
+
+ TreeLockTestSupport.HeldLock sameNameCreate =
+ TreeLockTestSupport.HeldLock.acquire(ident, LockType.WRITE);
+ TreeLockTestSupport.assertWaitsFor(sameNameCreate, () ->
registerModelForLockTest(ident));
+ }
+
+ @Test
+ public void testRegisterModelWaitsForSchemaWriteLock() throws Exception {
+ NameIdentifier schemaIdent = NameIdentifier.of(metalake, catalog,
"schema_model_lock_3");
+ createSchemaForLockTest(schemaIdent);
+
+ TreeLockTestSupport.HeldLock schemaWriter =
+ TreeLockTestSupport.HeldLock.acquire(schemaIdent, LockType.WRITE);
+ TreeLockTestSupport.assertWaitsFor(
+ schemaWriter,
+ () ->
+ registerModelForLockTest(
+ NameIdentifier.of(metalake, catalog, "schema_model_lock_3",
"model1")));
+ }
+
+ @Test
+ public void testRegisterModelWaitsForCatalogWriteLock() throws Exception {
+ NameIdentifier schemaIdent = NameIdentifier.of(metalake, catalog,
"schema_model_lock_4");
+ createSchemaForLockTest(schemaIdent);
+
+ TreeLockTestSupport.HeldLock catalogWriter =
+ TreeLockTestSupport.HeldLock.acquire(NameIdentifier.of(metalake,
catalog), LockType.WRITE);
+ TreeLockTestSupport.assertWaitsFor(
+ catalogWriter,
+ () ->
+ registerModelForLockTest(
+ NameIdentifier.of(metalake, catalog, "schema_model_lock_4",
"model1")));
+ }
+
+ private static void createSchemaForLockTest(NameIdentifier schemaIdent) {
+ schemaOperationDispatcher.createSchema(schemaIdent, "comment", null);
+ }
+
+ private static Model registerModelForLockTest(NameIdentifier ident) {
+ return modelOperationDispatcher.registerModel(
+ ident, "comment", ImmutableMap.of("k1", "v1", "k2", "v2"));
+ }
}
diff --git
a/core/src/test/java/org/apache/gravitino/catalog/TestTableOperationDispatcher.java
b/core/src/test/java/org/apache/gravitino/catalog/TestTableOperationDispatcher.java
index 94b490f0a7..9977b132a1 100644
---
a/core/src/test/java/org/apache/gravitino/catalog/TestTableOperationDispatcher.java
+++
b/core/src/test/java/org/apache/gravitino/catalog/TestTableOperationDispatcher.java
@@ -60,6 +60,7 @@ import org.apache.gravitino.connector.TestCatalogOperations;
import org.apache.gravitino.dto.util.DTOConverters;
import org.apache.gravitino.exceptions.NoSuchEntityException;
import org.apache.gravitino.lock.LockManager;
+import org.apache.gravitino.lock.LockType;
import org.apache.gravitino.meta.AuditInfo;
import org.apache.gravitino.meta.ColumnEntity;
import org.apache.gravitino.meta.SchemaEntity;
@@ -1175,6 +1176,73 @@ public class TestTableOperationDispatcher extends
TestOperationDispatcher {
return tableOperationDispatcher;
}
+ @Test
+ public void testCreateTableRunsConcurrentlyWithCreateOfAnotherTable() throws
Exception {
+ Namespace tableNs = Namespace.of(metalake, catalog,
"schema_create_lock_1");
+ schemaOperationDispatcher.createSchema(
+ NameIdentifier.of(tableNs.levels()), "comment", ImmutableMap.of("k1",
"v1", "k2", "v2"));
+
+ // Another in-flight create holds the WRITE lock on its own table node.
+ try (TreeLockTestSupport.HeldLock inFlightCreate =
+ TreeLockTestSupport.HeldLock.acquire(
+ NameIdentifier.of(tableNs, "other_table"), LockType.WRITE)) {
+ TreeLockTestSupport.assertRunsConcurrentlyWith(
+ inFlightCreate, () -> createTable(NameIdentifier.of(tableNs,
"table1")));
+ }
+ }
+
+ @Test
+ public void testCreateTableWaitsForCreateOfSameName() throws Exception {
+ Namespace tableNs = Namespace.of(metalake, catalog,
"schema_create_lock_2");
+ schemaOperationDispatcher.createSchema(
+ NameIdentifier.of(tableNs.levels()), "comment", ImmutableMap.of("k1",
"v1", "k2", "v2"));
+ NameIdentifier tableIdent = NameIdentifier.of(tableNs, "table1");
+
+ TreeLockTestSupport.HeldLock sameNameCreate =
+ TreeLockTestSupport.HeldLock.acquire(tableIdent, LockType.WRITE);
+ TreeLockTestSupport.assertWaitsFor(sameNameCreate, () ->
createTable(tableIdent));
+ }
+
+ @Test
+ public void testCreateTableWaitsForSchemaWriteLock() throws Exception {
+ // Rename, drop and import of tables in the schema take the schema WRITE
lock.
+ Namespace tableNs = Namespace.of(metalake, catalog,
"schema_create_lock_3");
+ NameIdentifier schemaIdent = NameIdentifier.of(tableNs.levels());
+ schemaOperationDispatcher.createSchema(
+ schemaIdent, "comment", ImmutableMap.of("k1", "v1", "k2", "v2"));
+
+ TreeLockTestSupport.HeldLock schemaWriter =
+ TreeLockTestSupport.HeldLock.acquire(schemaIdent, LockType.WRITE);
+ TreeLockTestSupport.assertWaitsFor(
+ schemaWriter, () -> createTable(NameIdentifier.of(tableNs, "table1")));
+ }
+
+ @Test
+ public void testCreateTableWaitsForCatalogWriteLock() throws Exception {
+ // dropSchema and createSchema take the catalog WRITE lock.
+ Namespace tableNs = Namespace.of(metalake, catalog,
"schema_create_lock_4");
+ schemaOperationDispatcher.createSchema(
+ NameIdentifier.of(tableNs.levels()), "comment", ImmutableMap.of("k1",
"v1", "k2", "v2"));
+
+ TreeLockTestSupport.HeldLock catalogWriter =
+ TreeLockTestSupport.HeldLock.acquire(NameIdentifier.of(metalake,
catalog), LockType.WRITE);
+ TreeLockTestSupport.assertWaitsFor(
+ catalogWriter, () -> createTable(NameIdentifier.of(tableNs,
"table1")));
+ }
+
+ private static Table createTable(NameIdentifier ident) {
+ Column[] columns =
+ new Column[] {
+ TestColumn.builder()
+ .withName("col1")
+ .withPosition(0)
+ .withType(Types.StringType.get())
+ .build()
+ };
+ return tableOperationDispatcher.createTable(
+ ident, columns, "comment", ImmutableMap.of("k1", "v1", "k2", "v2"),
new Transform[0]);
+ }
+
public static SchemaOperationDispatcher getSchemaOperationDispatcher() {
return schemaOperationDispatcher;
}
diff --git
a/core/src/test/java/org/apache/gravitino/catalog/TestTopicOperationDispatcher.java
b/core/src/test/java/org/apache/gravitino/catalog/TestTopicOperationDispatcher.java
index b8b979e7e5..c043e8f0b0 100644
---
a/core/src/test/java/org/apache/gravitino/catalog/TestTopicOperationDispatcher.java
+++
b/core/src/test/java/org/apache/gravitino/catalog/TestTopicOperationDispatcher.java
@@ -50,6 +50,7 @@ import org.apache.gravitino.auth.AuthConstants;
import org.apache.gravitino.connector.TestCatalogOperations;
import org.apache.gravitino.exceptions.NoSuchEntityException;
import org.apache.gravitino.lock.LockManager;
+import org.apache.gravitino.lock.LockType;
import org.apache.gravitino.messaging.Topic;
import org.apache.gravitino.messaging.TopicChange;
import org.apache.gravitino.meta.AuditInfo;
@@ -282,4 +283,71 @@ public class TestTopicOperationDispatcher extends
TestOperationDispatcher {
public static TopicOperationDispatcher getTopicOperationDispatcher() {
return topicOperationDispatcher;
}
+
+ @Test
+ public void testCreateTopicRunsConcurrentlyWithCreateTopicOfAnotherTopic()
throws Exception {
+ NameIdentifier schemaIdent = NameIdentifier.of(metalake, catalog,
"schema_topic_lock_1");
+ createSchemaForLockTest(schemaIdent);
+
+ // Another in-flight create holds the WRITE lock on its own topic node.
+ try (TreeLockTestSupport.HeldLock inFlightCreate =
+ TreeLockTestSupport.HeldLock.acquire(
+ NameIdentifier.of(metalake, catalog, "schema_topic_lock_1",
"other_topic"),
+ LockType.WRITE)) {
+ TreeLockTestSupport.assertRunsConcurrentlyWith(
+ inFlightCreate,
+ () ->
+ createTopicForLockTest(
+ NameIdentifier.of(metalake, catalog, "schema_topic_lock_1",
"topic1")));
+ }
+ }
+
+ @Test
+ public void testCreateTopicWaitsForCreateTopicOfSameName() throws Exception {
+ NameIdentifier schemaIdent = NameIdentifier.of(metalake, catalog,
"schema_topic_lock_2");
+ createSchemaForLockTest(schemaIdent);
+ NameIdentifier ident = NameIdentifier.of(metalake, catalog,
"schema_topic_lock_2", "topic1");
+
+ TreeLockTestSupport.HeldLock sameNameCreate =
+ TreeLockTestSupport.HeldLock.acquire(ident, LockType.WRITE);
+ TreeLockTestSupport.assertWaitsFor(sameNameCreate, () ->
createTopicForLockTest(ident));
+ }
+
+ @Test
+ public void testCreateTopicWaitsForSchemaWriteLock() throws Exception {
+ NameIdentifier schemaIdent = NameIdentifier.of(metalake, catalog,
"schema_topic_lock_3");
+ createSchemaForLockTest(schemaIdent);
+
+ TreeLockTestSupport.HeldLock schemaWriter =
+ TreeLockTestSupport.HeldLock.acquire(schemaIdent, LockType.WRITE);
+ TreeLockTestSupport.assertWaitsFor(
+ schemaWriter,
+ () ->
+ createTopicForLockTest(
+ NameIdentifier.of(metalake, catalog, "schema_topic_lock_3",
"topic1")));
+ }
+
+ @Test
+ public void testCreateTopicWaitsForCatalogWriteLock() throws Exception {
+ NameIdentifier schemaIdent = NameIdentifier.of(metalake, catalog,
"schema_topic_lock_4");
+ createSchemaForLockTest(schemaIdent);
+
+ TreeLockTestSupport.HeldLock catalogWriter =
+ TreeLockTestSupport.HeldLock.acquire(NameIdentifier.of(metalake,
catalog), LockType.WRITE);
+ TreeLockTestSupport.assertWaitsFor(
+ catalogWriter,
+ () ->
+ createTopicForLockTest(
+ NameIdentifier.of(metalake, catalog, "schema_topic_lock_4",
"topic1")));
+ }
+
+ private static void createSchemaForLockTest(NameIdentifier schemaIdent) {
+ schemaOperationDispatcher.createSchema(
+ schemaIdent, "comment", ImmutableMap.of("k1", "v1", "k2", "v2"));
+ }
+
+ private static Topic createTopicForLockTest(NameIdentifier ident) {
+ return topicOperationDispatcher.createTopic(
+ ident, "comment", null, ImmutableMap.of("k1", "v1", "k2", "v2"));
+ }
}
diff --git
a/core/src/test/java/org/apache/gravitino/catalog/TestViewOperationDispatcher.java
b/core/src/test/java/org/apache/gravitino/catalog/TestViewOperationDispatcher.java
index e98e0a2804..4505de380f 100644
---
a/core/src/test/java/org/apache/gravitino/catalog/TestViewOperationDispatcher.java
+++
b/core/src/test/java/org/apache/gravitino/catalog/TestViewOperationDispatcher.java
@@ -53,6 +53,7 @@ import org.apache.gravitino.connector.TestCatalogOperations;
import org.apache.gravitino.exceptions.NoSuchEntityException;
import org.apache.gravitino.exceptions.NoSuchViewException;
import org.apache.gravitino.lock.LockManager;
+import org.apache.gravitino.lock.LockType;
import org.apache.gravitino.meta.AuditInfo;
import org.apache.gravitino.meta.SchemaEntity;
import org.apache.gravitino.meta.ViewEntity;
@@ -699,4 +700,80 @@ public class TestViewOperationDispatcher extends
TestOperationDispatcher {
.build();
entityStore.put(entity, true);
}
+
+ @Test
+ public void testCreateViewRunsConcurrentlyWithCreateViewOfAnotherView()
throws Exception {
+ NameIdentifier schemaIdent = NameIdentifier.of(metalake, catalog,
"schema_view_lock_1");
+ createSchemaForLockTest(schemaIdent);
+
+ // Another in-flight create holds the WRITE lock on its own view node.
+ try (TreeLockTestSupport.HeldLock inFlightCreate =
+ TreeLockTestSupport.HeldLock.acquire(
+ NameIdentifier.of(metalake, catalog, "schema_view_lock_1",
"other_view"),
+ LockType.WRITE)) {
+ TreeLockTestSupport.assertRunsConcurrentlyWith(
+ inFlightCreate,
+ () ->
+ createViewForLockTest(
+ NameIdentifier.of(metalake, catalog, "schema_view_lock_1",
"view1")));
+ }
+ }
+
+ @Test
+ public void testCreateViewWaitsForCreateViewOfSameName() throws Exception {
+ NameIdentifier schemaIdent = NameIdentifier.of(metalake, catalog,
"schema_view_lock_2");
+ createSchemaForLockTest(schemaIdent);
+ NameIdentifier ident = NameIdentifier.of(metalake, catalog,
"schema_view_lock_2", "view1");
+
+ TreeLockTestSupport.HeldLock sameNameCreate =
+ TreeLockTestSupport.HeldLock.acquire(ident, LockType.WRITE);
+ TreeLockTestSupport.assertWaitsFor(sameNameCreate, () ->
createViewForLockTest(ident));
+ }
+
+ @Test
+ public void testCreateViewWaitsForSchemaWriteLock() throws Exception {
+ NameIdentifier schemaIdent = NameIdentifier.of(metalake, catalog,
"schema_view_lock_3");
+ createSchemaForLockTest(schemaIdent);
+
+ TreeLockTestSupport.HeldLock schemaWriter =
+ TreeLockTestSupport.HeldLock.acquire(schemaIdent, LockType.WRITE);
+ TreeLockTestSupport.assertWaitsFor(
+ schemaWriter,
+ () ->
+ createViewForLockTest(
+ NameIdentifier.of(metalake, catalog, "schema_view_lock_3",
"view1")));
+ }
+
+ @Test
+ public void testCreateViewWaitsForCatalogWriteLock() throws Exception {
+ NameIdentifier schemaIdent = NameIdentifier.of(metalake, catalog,
"schema_view_lock_4");
+ createSchemaForLockTest(schemaIdent);
+
+ TreeLockTestSupport.HeldLock catalogWriter =
+ TreeLockTestSupport.HeldLock.acquire(NameIdentifier.of(metalake,
catalog), LockType.WRITE);
+ TreeLockTestSupport.assertWaitsFor(
+ catalogWriter,
+ () ->
+ createViewForLockTest(
+ NameIdentifier.of(metalake, catalog, "schema_view_lock_4",
"view1")));
+ }
+
+ private static void createSchemaForLockTest(NameIdentifier schemaIdent) {
+ schemaOperationDispatcher.createSchema(
+ schemaIdent, "comment", ImmutableMap.of("k1", "v1", "k2", "v2"));
+ }
+
+ private static View createViewForLockTest(NameIdentifier ident) {
+ Representation[] representations = {
+ SQLRepresentation.builder().withDialect("spark").withSql("SELECT
1").build()
+ };
+ return viewOperationDispatcher.createView(
+ ident,
+ "comment",
+ new Column[0],
+ representations,
+ null,
+ null,
+ ImmutableMap.of("k1", "v1", "p1", "pv1"));
+ }
}
diff --git
a/core/src/test/java/org/apache/gravitino/catalog/TreeLockTestSupport.java
b/core/src/test/java/org/apache/gravitino/catalog/TreeLockTestSupport.java
new file mode 100644
index 0000000000..160191f838
--- /dev/null
+++ b/core/src/test/java/org/apache/gravitino/catalog/TreeLockTestSupport.java
@@ -0,0 +1,214 @@
+/*
+ * 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.catalog;
+
+import java.time.Duration;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+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 org.apache.gravitino.GravitinoEnv;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.lock.LockType;
+import org.apache.gravitino.lock.TreeLock;
+import org.apache.gravitino.lock.TreeLockNode;
+import org.junit.jupiter.api.Assertions;
+
+/**
+ * Helpers for asserting how a dispatcher operation interacts with tree locks
held by other threads.
+ * A {@link HeldLock} pins a lock on a helper thread so a test can check
whether an operation runs
+ * concurrently with it or waits for it. Waiting is detected by observing the
operation thread
+ * parked inside {@link TreeLockNode}, not by sleeping, so the assertions do
not depend on timing.
+ */
+final class TreeLockTestSupport {
+
+ private static final Duration TIMEOUT = Duration.ofSeconds(30);
+
+ private TreeLockTestSupport() {}
+
+ /**
+ * Asserts that {@code operation} completes while {@code held} is still
held, i.e. the operation
+ * does not contend with that lock.
+ */
+ static void assertRunsConcurrentlyWith(HeldLock held, Runnable operation)
throws Exception {
+ OperationRun run = OperationRun.start(operation);
+ try {
+ if (run.awaitDoneOrParked() == Outcome.PARKED) {
+ held.release();
+ Assertions.fail(
+ "Operation parked on a tree lock while " + held + " was held; it
contends with it");
+ }
+ run.join();
+ } finally {
+ run.shutdown();
+ }
+ }
+
+ /**
+ * Asserts that {@code operation} parks on a tree lock while {@code held} is
held and completes
+ * once the lock is released.
+ */
+ static void assertWaitsFor(HeldLock held, Runnable operation) throws
Exception {
+ OperationRun run = OperationRun.start(operation);
+ try {
+ Assertions.assertEquals(
+ Outcome.PARKED,
+ run.awaitDoneOrParked(),
+ "Operation finished without waiting for " + held);
+ held.release();
+ run.join();
+ } finally {
+ run.shutdown();
+ }
+ }
+
+ private enum Outcome {
+ DONE,
+ PARKED
+ }
+
+ /** An operation running on its own thread whose progress can be observed. */
+ private static final class OperationRun {
+ private final ExecutorService executor;
+ private final Thread[] worker = new Thread[1];
+ private Future<?> future;
+
+ private OperationRun() {
+ this.executor =
+ Executors.newSingleThreadExecutor(
+ r -> {
+ worker[0] = new Thread(r, "tree-lock-test-operation");
+ return worker[0];
+ });
+ }
+
+ static OperationRun start(Runnable operation) {
+ OperationRun run = new OperationRun();
+ run.future = run.executor.submit(operation);
+ return run;
+ }
+
+ /** Waits until the operation either completes or blocks inside {@link
TreeLockNode#lock}. */
+ Outcome awaitDoneOrParked() throws Exception {
+ long deadline = System.nanoTime() + TIMEOUT.toNanos();
+ while (System.nanoTime() < deadline) {
+ if (future.isDone()) {
+ return Outcome.DONE;
+ }
+ Thread thread = worker[0];
+ if (thread != null && isParkedOnTreeLock(thread)) {
+ return Outcome.PARKED;
+ }
+ Thread.sleep(5);
+ }
+ throw new TimeoutException("Operation neither finished nor parked on a
tree lock");
+ }
+
+ void join() throws Exception {
+ try {
+ future.get(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
+ } catch (ExecutionException e) {
+ throw new AssertionError("Operation failed", e.getCause());
+ }
+ }
+
+ void shutdown() {
+ executor.shutdownNow();
+ }
+
+ private static boolean isParkedOnTreeLock(Thread thread) {
+ Thread.State state = thread.getState();
+ if (state != Thread.State.WAITING && state !=
Thread.State.TIMED_WAITING) {
+ return false;
+ }
+ for (StackTraceElement frame : thread.getStackTrace()) {
+ if (TreeLockNode.class.getName().equals(frame.getClassName())
+ && "lock".equals(frame.getMethodName())) {
+ return true;
+ }
+ }
+ return false;
+ }
+ }
+
+ /** A tree lock held by a dedicated thread until {@link #release()} or
{@link #close()}. */
+ static final class HeldLock implements AutoCloseable {
+ private final NameIdentifier identifier;
+ private final LockType lockType;
+ private final CountDownLatch acquired = new CountDownLatch(1);
+ private final CountDownLatch releaseSignal = new CountDownLatch(1);
+ private final Thread holder;
+ private volatile Throwable failure;
+
+ private HeldLock(NameIdentifier identifier, LockType lockType) {
+ this.identifier = identifier;
+ this.lockType = lockType;
+ this.holder = new Thread(this::hold, "tree-lock-test-holder");
+ }
+
+ /** Acquires the lock on a helper thread and returns once it is held. */
+ static HeldLock acquire(NameIdentifier identifier, LockType lockType)
throws Exception {
+ HeldLock held = new HeldLock(identifier, lockType);
+ held.holder.start();
+ held.acquired.await(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
+ if (held.failure != null) {
+ throw new AssertionError("Failed to acquire " + held, held.failure);
+ }
+ return held;
+ }
+
+ private void hold() {
+ TreeLock lock =
GravitinoEnv.getInstance().lockManager().createTreeLock(identifier);
+ try {
+ lock.lock(lockType);
+ } catch (Throwable t) {
+ failure = t;
+ acquired.countDown();
+ return;
+ }
+ acquired.countDown();
+ try {
+ releaseSignal.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ /** Releases the lock and waits for the holder thread to exit. */
+ void release() throws InterruptedException {
+ releaseSignal.countDown();
+ holder.join(TIMEOUT.toMillis());
+ }
+
+ @Override
+ public void close() throws InterruptedException {
+ release();
+ }
+
+ @Override
+ public String toString() {
+ return lockType + " lock on " + identifier;
+ }
+ }
+}
