Copilot commented on code in PR #13792:
URL: https://github.com/apache/gravitino/pull/13792#discussion_r4248609802
##########
core/src/main/java/org/apache/gravitino/job/JobManager.java:
##########
@@ -600,6 +600,27 @@ public JobEntity runJob(String metalake, String
jobTemplateName, Map<String, Str
"Job template with name %s under metalake %s does not exist",
jobTemplateName,
metalake);
+ } catch (OptimisticLockException e) {
+ // The registration conflicted with a concurrent writer, e.g. a lock
wait timeout, and the
+ // conflict tells the caller to retry. A retry submits the job again, so
cancel this
+ // execution rather than leave it running untracked.
+ LOG.error(
+ "Job {} was submitted as execution {} but could not be registered
because of a "
+ + "concurrent operation, cancelling the execution",
+ jobEntity.name(),
+ jobExecutionId,
+ e);
Review Comment:
This logs an expected retryable conflict (HTTP 409 semantics) at ERROR
level, which can create noisy error logs during normal concurrent activity and
obscure real failures. Consider logging this as WARN (or INFO if appropriate
for your operational posture), reserving ERROR for non-retryable or unexpected
job-registration failures.
##########
core/src/test/java/org/apache/gravitino/storage/relational/TestRelationalEntityStoreLockWaitTimeout.java:
##########
@@ -0,0 +1,494 @@
+/*
+ * 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;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.time.Instant;
+import java.util.List;
+import org.apache.commons.io.FileUtils;
+import org.apache.commons.lang3.reflect.FieldUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.gravitino.Catalog;
+import org.apache.gravitino.Config;
+import org.apache.gravitino.Configs;
+import org.apache.gravitino.Entity;
+import org.apache.gravitino.GravitinoEnv;
+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.cache.EntityCache;
+import org.apache.gravitino.exceptions.OptimisticLockException;
+import org.apache.gravitino.meta.AuditInfo;
+import org.apache.gravitino.meta.BaseMetalake;
+import org.apache.gravitino.meta.CatalogEntity;
+import org.apache.gravitino.meta.SchemaEntity;
+import org.apache.gravitino.meta.SchemaVersion;
+import org.apache.gravitino.meta.TableEntity;
+import org.apache.gravitino.meta.TagEntity;
+import org.apache.gravitino.storage.RandomIdGenerator;
+import org.apache.gravitino.utils.NamespaceUtil;
+import org.apache.ibatis.exceptions.PersistenceException;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.function.Executable;
+import org.mockito.Mockito;
+
+/**
+ * Verifies that {@link RelationalEntityStore} reports a write that fails on a
lock wait timeout as
+ * an {@link OptimisticLockException}, and leaves every other failure
unchanged.
+ */
+public class TestRelationalEntityStoreLockWaitTimeout {
+
+ private static final String METALAKE = "metalake_lwt";
+ private static final String CATALOG = "catalog_lwt";
+ private static final String SCHEMA = "schema_lwt";
+ private static final String USER = "root";
+ private static final String PASSWORD = "123456";
+
+ private static final AuditInfo AUDIT_INFO =
+
AuditInfo.builder().withCreator("creator").withCreateTime(Instant.now()).build();
+
+ private static final NameIdentifier TABLE_IDENT =
+ NameIdentifier.of(METALAKE, CATALOG, SCHEMA, "t1");
+
+ private RelationalEntityStore store;
+ private String dbPath;
+ private Object previousConfig;
+ private Object previousIdGenerator;
+
+ @AfterEach
+ void tearDown() throws Exception {
+ // Only the H2 store owns resources; a store over a failing mock backend
is just dropped.
+ if (dbPath != null) {
+ store.close();
+ FileUtils.deleteQuietly(new File(dbPath));
Review Comment:
The test cleanup deletes `dbPath` (a temp directory), but the H2 URL uses
`jdbc:h2:file:%s` where `%s` is the directory path. H2 typically creates
database files adjacent to that path (e.g., `<dbPath>.mv.db`,
`<dbPath>.trace.db`) rather than inside the directory, so `deleteQuietly(new
File(dbPath))` may leave the actual DB files behind. To avoid leaking temp
files across test runs, consider either (a) using a file base name inside the
directory (e.g., `...file:%s/db...`) so deleting the directory removes all
artifacts, or (b) explicitly deleting the known H2 file suffixes for the chosen
base path.
##########
core/src/main/java/org/apache/gravitino/storage/relational/utils/ExceptionUtils.java:
##########
@@ -69,4 +94,103 @@ public static OptimisticLockException
concurrentChildModification(
parentType.name().toLowerCase(Locale.ROOT),
parentIdentifier);
}
+
+ /**
+ * Throws a write that failed on a lock wait timeout as a conflict, and
returns otherwise. A
+ * service calls this before {@link #checkSQLException} when the type and
name it reports for
+ * other SQL errors do not describe what was written, for example the
metadata object of a tag
+ * association.
+ *
+ * @param re The failure of the write.
+ * @param target What was being written, as the subject of the message.
+ * @throws OptimisticLockException If the failure is a lock wait timeout.
+ * @see #lockWaitTimeoutAsConflict(RuntimeException, String)
+ */
+ public static void throwIfLockWaitTimeout(RuntimeException re, String
target) {
+ RuntimeException conflict = lockWaitTimeoutAsConflict(re, target);
+ if (conflict != re) {
+ throw conflict;
+ }
+ }
+
+ /**
+ * Returns whether a failure was caused by a statement that waited on a row
lock longer than the
+ * backend's lock wait timeout. MySQL reports error code 1205, H2 error code
50200 (SQLState
+ * HYT00), and PostgreSQL SQLState 55P03 when lock_timeout is set. The MySQL
and H2 vendor codes
+ * are matched rather than their SQLState, because the MySQL driver reports
a lock wait timeout as
+ * SQLState 40001 or HY000 depending on its version, and 40001 is also used
for deadlocks.
+ *
+ * <p>Deadlocks are deliberately not included: a deadlock on a write path
points at an
+ * inconsistent lock order, which should be fixed on that path rather than
reported as a conflict.
+ *
+ * @param failure The failure to inspect, including its causes.
+ * @return {@code true} if a lock wait timeout is found in the cause chain.
+ */
+ public static boolean isLockWaitTimeout(Throwable failure) {
+ return hasSqlException(
+ failure,
+ Set.of(POSTGRESQL_LOCK_NOT_AVAILABLE),
+ Set.of(MYSQL_LOCK_WAIT_TIMEOUT, H2_LOCK_TIMEOUT));
+ }
+
+ /**
+ * Returns whether the cause chain of a failure contains a {@link
SQLException} with one of the
+ * given SQLStates or vendor error codes.
+ *
+ * @param failure The failure to inspect, including its causes.
+ * @param sqlStates The SQLStates to match.
+ * @param vendorErrorCodes The vendor error codes to match.
+ * @return {@code true} if a matching {@link SQLException} is found.
+ */
+ public static boolean hasSqlException(
+ Throwable failure, Set<String> sqlStates, Set<Integer> vendorErrorCodes)
{
+ for (Throwable cause = failure; cause != null; cause = cause.getCause()) {
+ if (cause instanceof SQLException) {
+ SQLException sql = (SQLException) cause;
+ if (sqlStates.contains(sql.getSQLState())
+ || vendorErrorCodes.contains(sql.getErrorCode())) {
+ return true;
+ }
+ }
+ }
+ return false;
+ }
Review Comment:
`hasSqlException` only walks the `Throwable.getCause()` chain. JDBC drivers
frequently attach additional `SQLException`s via
`SQLException#getNextException()` (especially for batch/statement failures),
which this implementation will miss—causing lock-wait timeouts (or other
classified SQL errors) to be misreported. Consider iterating `SQLException`'s
`getNextException()` chain for each encountered `SQLException` (while still
walking the outer cause chain) so classification works reliably across drivers
and batch scenarios.
##########
core/src/main/java/org/apache/gravitino/storage/relational/utils/ExceptionUtils.java:
##########
@@ -69,4 +94,103 @@ public static OptimisticLockException
concurrentChildModification(
parentType.name().toLowerCase(Locale.ROOT),
parentIdentifier);
}
+
+ /**
+ * Throws a write that failed on a lock wait timeout as a conflict, and
returns otherwise. A
+ * service calls this before {@link #checkSQLException} when the type and
name it reports for
+ * other SQL errors do not describe what was written, for example the
metadata object of a tag
+ * association.
+ *
+ * @param re The failure of the write.
+ * @param target What was being written, as the subject of the message.
+ * @throws OptimisticLockException If the failure is a lock wait timeout.
+ * @see #lockWaitTimeoutAsConflict(RuntimeException, String)
+ */
+ public static void throwIfLockWaitTimeout(RuntimeException re, String
target) {
+ RuntimeException conflict = lockWaitTimeoutAsConflict(re, target);
+ if (conflict != re) {
+ throw conflict;
+ }
+ }
+
+ /**
+ * Returns whether a failure was caused by a statement that waited on a row
lock longer than the
+ * backend's lock wait timeout. MySQL reports error code 1205, H2 error code
50200 (SQLState
+ * HYT00), and PostgreSQL SQLState 55P03 when lock_timeout is set. The MySQL
and H2 vendor codes
+ * are matched rather than their SQLState, because the MySQL driver reports
a lock wait timeout as
+ * SQLState 40001 or HY000 depending on its version, and 40001 is also used
for deadlocks.
+ *
+ * <p>Deadlocks are deliberately not included: a deadlock on a write path
points at an
+ * inconsistent lock order, which should be fixed on that path rather than
reported as a conflict.
+ *
+ * @param failure The failure to inspect, including its causes.
+ * @return {@code true} if a lock wait timeout is found in the cause chain.
+ */
+ public static boolean isLockWaitTimeout(Throwable failure) {
+ return hasSqlException(
+ failure,
+ Set.of(POSTGRESQL_LOCK_NOT_AVAILABLE),
+ Set.of(MYSQL_LOCK_WAIT_TIMEOUT, H2_LOCK_TIMEOUT));
+ }
Review Comment:
`isLockWaitTimeout` allocates new `Set` instances on every call
(`Set.of(...)`). If this method is called on hot paths (multiple write
failures, retries), these allocations are avoidable. Consider hoisting these
into `private static final` sets (e.g., `LOCK_WAIT_TIMEOUT_SQLSTATES` /
`LOCK_WAIT_TIMEOUT_VENDOR_CODES`) and reuse them.
##########
core/src/test/java/org/apache/gravitino/storage/relational/TestRelationalEntityStoreLockWaitTimeout.java:
##########
@@ -0,0 +1,494 @@
+/*
+ * 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;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.time.Instant;
+import java.util.List;
+import org.apache.commons.io.FileUtils;
+import org.apache.commons.lang3.reflect.FieldUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.gravitino.Catalog;
+import org.apache.gravitino.Config;
+import org.apache.gravitino.Configs;
+import org.apache.gravitino.Entity;
+import org.apache.gravitino.GravitinoEnv;
+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.cache.EntityCache;
+import org.apache.gravitino.exceptions.OptimisticLockException;
+import org.apache.gravitino.meta.AuditInfo;
+import org.apache.gravitino.meta.BaseMetalake;
+import org.apache.gravitino.meta.CatalogEntity;
+import org.apache.gravitino.meta.SchemaEntity;
+import org.apache.gravitino.meta.SchemaVersion;
+import org.apache.gravitino.meta.TableEntity;
+import org.apache.gravitino.meta.TagEntity;
+import org.apache.gravitino.storage.RandomIdGenerator;
+import org.apache.gravitino.utils.NamespaceUtil;
+import org.apache.ibatis.exceptions.PersistenceException;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.function.Executable;
+import org.mockito.Mockito;
+
+/**
+ * Verifies that {@link RelationalEntityStore} reports a write that fails on a
lock wait timeout as
+ * an {@link OptimisticLockException}, and leaves every other failure
unchanged.
+ */
+public class TestRelationalEntityStoreLockWaitTimeout {
+
+ private static final String METALAKE = "metalake_lwt";
+ private static final String CATALOG = "catalog_lwt";
+ private static final String SCHEMA = "schema_lwt";
+ private static final String USER = "root";
+ private static final String PASSWORD = "123456";
+
+ private static final AuditInfo AUDIT_INFO =
+
AuditInfo.builder().withCreator("creator").withCreateTime(Instant.now()).build();
+
+ private static final NameIdentifier TABLE_IDENT =
+ NameIdentifier.of(METALAKE, CATALOG, SCHEMA, "t1");
+
+ private RelationalEntityStore store;
+ private String dbPath;
+ private Object previousConfig;
+ private Object previousIdGenerator;
+
+ @AfterEach
+ void tearDown() throws Exception {
+ // Only the H2 store owns resources; a store over a failing mock backend
is just dropped.
+ if (dbPath != null) {
+ store.close();
+ FileUtils.deleteQuietly(new File(dbPath));
+ dbPath = null;
+ FieldUtils.writeField(GravitinoEnv.getInstance(), "config",
previousConfig, true);
+ FieldUtils.writeField(GravitinoEnv.getInstance(), "idGenerator",
previousIdGenerator, true);
+ }
+ store = null;
+ }
+
+ /**
+ * Holds the row of a table from another session, as a concurrent
table-scoped write does, so that
+ * a drop of the table waits on it until H2 aborts the statement.
+ */
+ @Test
+ void testDropWaitingOnHeldTableRowIsConflict() throws Exception {
+ String jdbcUrl = initH2Store();
+ TableEntity table = createTable();
+
+ OptimisticLockException conflict =
+ writeWhileRowHeld(
+ jdbcUrl,
+ "table_meta",
+ "table_id",
+ table.id(),
+ () -> store.delete(TABLE_IDENT, Entity.EntityType.TABLE));
+ Assertions.assertTrue(
+ conflict.getMessage().startsWith("The table " + TABLE_IDENT + " could
not be changed"),
+ conflict.getMessage());
+
+ // The failed drop was rolled back, so the table is still there and a
retry succeeds.
+ Assertions.assertTrue(store.exists(TABLE_IDENT, Entity.EntityType.TABLE));
+ Assertions.assertTrue(store.delete(TABLE_IDENT, Entity.EntityType.TABLE));
+ Assertions.assertFalse(store.exists(TABLE_IDENT, Entity.EntityType.TABLE));
+ }
+
+ /** An alter goes through the service's own SQL error conversion, which must
keep the conflict. */
+ @Test
+ void testAlterWaitingOnHeldTableRowIsConflict() throws Exception {
+ String jdbcUrl = initH2Store();
+ TableEntity table = createTable();
+ NameIdentifier renamed = NameIdentifier.of(TABLE_IDENT.namespace(),
"t1_renamed");
+
+ OptimisticLockException conflict =
+ writeWhileRowHeld(
+ jdbcUrl,
+ "table_meta",
+ "table_id",
+ table.id(),
+ () ->
+ store.update(
+ TABLE_IDENT, TableEntity.class, Entity.EntityType.TABLE,
this::rename));
+ // The service names the entity it writes, which is the new name of a
rename.
+ Assertions.assertTrue(
+ conflict.getMessage().startsWith("The table " + renamed + " could not
be changed"),
+ conflict.getMessage());
+
+ Assertions.assertTrue(store.exists(TABLE_IDENT, Entity.EntityType.TABLE));
+ Assertions.assertFalse(store.exists(renamed, Entity.EntityType.TABLE));
+ store.update(TABLE_IDENT, TableEntity.class, Entity.EntityType.TABLE,
this::rename);
+ Assertions.assertTrue(store.exists(renamed, Entity.EntityType.TABLE));
+ }
+
+ /** A create locks its parent row first, so it waits behind a write that
holds the schema. */
+ @Test
+ void testCreateWaitingOnHeldSchemaRowIsConflict() throws Exception {
+ String jdbcUrl = initH2Store();
+ store.put(metalake(), false);
+ store.put(catalog(), false);
+ SchemaEntity schema = schema();
+ store.put(schema, false);
+ TableEntity table = table();
+
+ OptimisticLockException conflict =
+ writeWhileRowHeld(
+ jdbcUrl, "schema_meta", "schema_id", schema.id(), () ->
store.put(table, false));
+ Assertions.assertTrue(
+ conflict.getMessage().startsWith("The table " + TABLE_IDENT + " could
not be changed"),
+ conflict.getMessage());
+
+ Assertions.assertFalse(store.exists(TABLE_IDENT, Entity.EntityType.TABLE));
+ store.put(table, false);
+ Assertions.assertTrue(store.exists(TABLE_IDENT, Entity.EntityType.TABLE));
+ }
+
+ /**
+ * A tag association locks the table row first, as the issue's example of a
table-scoped write.
+ */
+ @Test
+ void testTagAssociationWaitingOnHeldTableRowIsConflict() throws Exception {
+ String jdbcUrl = initH2Store();
+ TableEntity table = createTable();
+ TagEntity tag =
+ TagEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName("tag1")
+ .withNamespace(NamespaceUtil.ofTag(METALAKE))
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ store.put(tag, false);
+ NameIdentifier[] tags = {tag.nameIdentifier()};
+
+ OptimisticLockException conflict =
+ writeWhileRowHeld(jdbcUrl, "table_meta", "table_id", table.id(), () ->
associateTags(tags));
+ Assertions.assertTrue(
+ conflict.getMessage().startsWith("The tags of table " + TABLE_IDENT +
" could not"),
+ conflict.getMessage());
+
+ Assertions.assertTrue(
+ store
+ .listEntitiesByRelation(
+ SupportsRelationOperations.Type.TAG_METADATA_OBJECT_REL,
+ TABLE_IDENT,
+ Entity.EntityType.TABLE)
+ .isEmpty());
+ Assertions.assertEquals(1, associateTags(tags).size());
+ }
+
+ /** Verifies every write of the store maps a lock wait timeout of the
backend to a conflict. */
+ @Test
+ void testEveryWriteMapsLockWaitTimeout() throws Exception {
+ for (SQLException timeout :
+ List.of(
+ new SQLException("MySQL lock wait timeout", "40001", 1205),
+ new SQLException("MySQL lock wait timeout", "HY000", 1205),
+ new SQLException("H2 lock wait timeout", "HYT00", 50200),
+ new SQLException("PostgreSQL lock timeout", "55P03"))) {
+ PersistenceException failure = new PersistenceException(timeout);
+ initFailingStore(failure);
+ for (Pair<String, Executable> write : writes()) {
+ OptimisticLockException conflict =
+ Assertions.assertThrows(
+ OptimisticLockException.class, write.getRight(),
write.getLeft());
+ Assertions.assertSame(failure, conflict.getCause(), write.getLeft());
+ Assertions.assertTrue(
+ conflict.getMessage().contains("lock wait timeout"),
conflict.getMessage());
+ }
+ }
+
+ // The message names what was written.
+ initFailingStore(new PersistenceException(new SQLException("timeout",
"HYT00", 50200)));
+ OptimisticLockException conflict =
+ Assertions.assertThrows(
+ OptimisticLockException.class,
+ () -> store.delete(TABLE_IDENT, Entity.EntityType.TABLE));
+ Assertions.assertTrue(
+ conflict.getMessage().startsWith("The table " + TABLE_IDENT + " could
not be changed"),
+ conflict.getMessage());
+ }
+
+ /** Verifies a failure that is not a lock wait timeout reaches the caller
unchanged. */
+ @Test
+ void testOtherFailuresAreUnchanged() throws Exception {
+ for (RuntimeException failure :
+ List.of(
+ // A deadlock points at an inconsistent lock order and is not
reported as a conflict.
+ new PersistenceException(new SQLException("MySQL deadlock",
"40001", 1213)),
+ new PersistenceException(new SQLException("PostgreSQL deadlock",
"40P01")),
+ new PersistenceException(new SQLException("connection lost",
"08006")),
+ new IllegalStateException("not a SQL failure"),
+ // A conflict reported by the backend keeps its own message and is
not wrapped again.
+ new OptimisticLockException(
+ new PersistenceException(new SQLException("timeout", "HYT00",
50200)),
+ "already a conflict"))) {
+ initFailingStore(failure);
+ for (Pair<String, Executable> write : writes()) {
+ Throwable thrown = Assertions.assertThrows(Throwable.class,
write.getRight());
+ Assertions.assertSame(failure, thrown, write.getLeft());
+ }
+ }
+ }
+
+ private List<Pair<String, Executable>> writes() {
+ TableEntity table = table();
+ NameIdentifier tagIdent = NameIdentifier.of(METALAKE, "tag1");
+ return List.of(
+ Pair.of("put", () -> store.put(table, false)),
+ Pair.of(
+ "update",
+ () ->
+ store.update(
+ TABLE_IDENT, TableEntity.class, Entity.EntityType.TABLE,
entity -> entity)),
+ Pair.of("delete", () -> store.delete(TABLE_IDENT,
Entity.EntityType.TABLE)),
+ Pair.of(
+ "deleteAndGet",
+ () ->
+ store.deleteAndGet(
+ TABLE_IDENT, Entity.EntityType.TABLE, TableEntity.class,
entity -> {})),
+ Pair.of(
+ "insertRelation",
+ () ->
+ store.insertRelation(
+ SupportsRelationOperations.Type.OWNER_REL,
+ TABLE_IDENT,
+ Entity.EntityType.TABLE,
+ NameIdentifier.of(METALAKE, "user1"),
+ Entity.EntityType.USER,
+ true)),
+ Pair.of(
+ "batchInsertRelations",
+ () ->
+ store.batchInsertRelations(
+ SupportsRelationOperations.Type.TAG_METADATA_OBJECT_REL,
+ List.of(TABLE_IDENT),
+ Entity.EntityType.TABLE,
+ tagIdent,
+ Entity.EntityType.TAG,
+ false)),
+ Pair.of(
+ "updateEntityRelations",
+ () ->
+ store.updateEntityRelations(
+ SupportsRelationOperations.Type.TAG_METADATA_OBJECT_REL,
+ TABLE_IDENT,
+ Entity.EntityType.TABLE,
+ new NameIdentifier[] {tagIdent},
+ new NameIdentifier[0])),
+ Pair.of(
+ "updateEntityRelations(RelationUpdate)",
+ () ->
+ store.updateEntityRelations(
+ RelationUpdate.of(
+ SupportsRelationOperations.Type.POLICY_TAG_REL,
+ tagIdent,
+ Entity.EntityType.TAG,
+ new RelationEdgeTarget[] {
+ RelationEdgeTarget.of(
+ NameIdentifier.of(METALAKE, "policy1"),
+ Entity.EntityType.POLICY,
+ null)
+ },
+ new RelationEdgeTarget[0]))),
+ Pair.of(
+ "batchDelete",
+ () -> store.batchDelete(List.of(Pair.of(TABLE_IDENT,
Entity.EntityType.TABLE)), false)),
+ Pair.of("batchPut", () -> store.batchPut(List.of(table), true)));
+ }
+
+ private List<TagEntity> associateTags(NameIdentifier[] tags) throws
IOException {
+ return store.updateEntityRelations(
+ SupportsRelationOperations.Type.TAG_METADATA_OBJECT_REL,
+ TABLE_IDENT,
+ Entity.EntityType.TABLE,
+ tags,
+ new NameIdentifier[0]);
+ }
+
+ private TableEntity rename(TableEntity table) {
+ return TableEntity.builder()
+ .withId(table.id())
+ .withName("t1_renamed")
+ .withNamespace(table.namespace())
+ .withColumns(table.columns())
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ }
+
+ private TableEntity createTable() throws IOException {
+ store.put(metalake(), false);
+ store.put(catalog(), false);
+ store.put(schema(), false);
+ TableEntity table = table();
+ store.put(table, false);
+ return table;
+ }
+
+ /**
+ * Runs a write while another session holds a row with {@code SELECT ... FOR
UPDATE}, and checks
+ * that the write fails as a conflict caused by H2's lock wait timeout.
+ */
+ private OptimisticLockException writeWhileRowHeld(
+ String jdbcUrl, String table, String idColumn, long id, Executable write)
+ throws SQLException {
+ try (Connection holder = DriverManager.getConnection(jdbcUrl, USER,
PASSWORD)) {
+ holder.setAutoCommit(false);
+ try {
+ try (PreparedStatement statement =
+ holder.prepareStatement(
+ String.format(
+ "SELECT %s FROM %s WHERE %s = ? FOR UPDATE", idColumn,
table, idColumn))) {
+ statement.setLong(1, id);
+ try (ResultSet rs = statement.executeQuery()) {
+ Assertions.assertTrue(rs.next(), "the row must exist to be held");
+ }
+ }
+
+ OptimisticLockException conflict =
+ Assertions.assertThrows(OptimisticLockException.class, write);
+ Assertions.assertTrue(
+ conflict.getMessage().contains("lock wait timeout; retry the
operation"),
+ conflict.getMessage());
+ SQLException timeout = findSqlException(conflict);
+ Assertions.assertNotNull(timeout, "the lock wait timeout stays
attached for diagnosis");
+ Assertions.assertEquals(50200, timeout.getErrorCode());
+ return conflict;
+ } finally {
+ holder.rollback();
+ }
+ }
+ }
+
+ /** Creates a store whose backend fails every call with the given failure. */
+ private void initFailingStore(RuntimeException failure) throws
IllegalAccessException {
+ RelationalBackend backend =
+ Mockito.mock(
+ RelationalBackend.class,
+ invocation -> {
+ throw failure;
+ });
+ store = new RelationalEntityStore();
+ FieldUtils.writeField(store, "backend", backend, true);
+ FieldUtils.writeField(store, "cache", Mockito.mock(EntityCache.class),
true);
+ }
+
+ private String initH2Store() throws IOException, IllegalAccessException {
+ dbPath =
Files.createTempDirectory("gravitino_lock_wait_timeout_").toString();
+ String jdbcUrl =
String.format("jdbc:h2:file:%s;DB_CLOSE_DELAY=-1;MODE=MYSQL", dbPath);
Review Comment:
The test cleanup deletes `dbPath` (a temp directory), but the H2 URL uses
`jdbc:h2:file:%s` where `%s` is the directory path. H2 typically creates
database files adjacent to that path (e.g., `<dbPath>.mv.db`,
`<dbPath>.trace.db`) rather than inside the directory, so `deleteQuietly(new
File(dbPath))` may leave the actual DB files behind. To avoid leaking temp
files across test runs, consider either (a) using a file base name inside the
directory (e.g., `...file:%s/db...`) so deleting the directory removes all
artifacts, or (b) explicitly deleting the known H2 file suffixes for the chosen
base path.
--
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]
