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 19d88d7bad [#13776] improvement(core): Report entity-store lock-wait
timeouts as conflicts (#13792)
19d88d7bad is described below
commit 19d88d7bad614f45d50b68aecd32824cbb4bdb30
Author: Qi Yu <[email protected]>
AuthorDate: Thu Oct 15 18:07:09 2026 +0800
[#13776] improvement(core): Report entity-store lock-wait timeouts as
conflicts (#13792)
### What changes were proposed in this pull request?
A write to the relational entity store that waits on a row lock longer
than the backend's lock-wait timeout is now classified as
`OptimisticLockException`. The main REST API reports HTTP 409 (code
1012) where the caller can retry; a job that was already submitted
remains HTTP 500 because execution side effects cannot be undone by
cancellation.
- `ExceptionUtils.isLockWaitTimeout` recognises the timeout through
causes and chained JDBC exceptions, with cycle protection: MySQL error
code 1205, H2 error code 50200 (SQLState HYT00), and PostgreSQL SQLState
55P03. The vendor codes are matched for MySQL and H2, because the MySQL
driver reports 1205 as SQLState 40001 or HY000, and 40001 is also a
deadlock. Deadlocks are not mapped.
- `ExceptionUtils.checkSQLException` gives duplicate-key violations
precedence across the complete exception chain, then classifies a lock
timeout before the backend converter, which would otherwise turn it into
an `IOException`. Mixed duplicate-key/timeout chains retain the
non-retryable already-exists error and original cause; the tag/model
association pre-classification helpers also preserve that precedence.
This covers create, alter and the relation writes of every service. Tag
associations and model versions name what was written via
`throwIfLockWaitTimeout`.
- `RelationalEntityStore` maps the timeout on every write (put, update,
delete, deleteAndGet, the relation writes, batchPut, batchDelete), for
paths such as drop that don't go through `checkSQLException`.
- `StatisticMetaService` reuses the classification and keeps its own
deadlock handling for batch statements, whose row lock order follows the
database's plan.
- Lance metadata repair and empty-version registration keep their
existing bounded CAS retries, but immediately propagate a lock-wait
timeout without retrying or backing off, so one load cannot multiply the
backend lock-wait budget by five. Job cleanup/status logs describe
concurrent operations rather than claiming the row changed.
- The conflict reaches the caller instead of being wrapped into a
`RuntimeException` (500) on the external schema, topic and view drops,
on the schema, table, topic and view import paths, on the schema drops
of a managed catalog force drop, and in Lance table create, drop and
purge. A Lance create that conflicts attempts to drop the dataset it
just created, as it already does when the table exists; cleanup
failures, including checked exceptions thrown by JNI, are logged and
retained as suppressed exceptions on the original registration error.
The same protection applies when metadata already exists.
- `JobManager.runJob` attempts to cancel a submitted execution when
registration conflicts. Because cancellation is asynchronous and cannot
undo completed work, this post-submission failure stays HTTP 500,
preserves the conflict as its cause, and identifies the execution to
check before retrying.
### Why are the changes needed?
With the tree lock removed (#13737), writers on one server queue on rows
inside the database. For example, a table drop waits behind a statistic
update or a tag association that holds the table row. If the holder
takes longer than the lock-wait timeout, the waiting statement is
aborted and the request returned 500. The store rolls its transaction
back, so the store write can be retried. Callers must account for any
preceding external side effects before replaying the whole operation.
H2, the default backend, has a lock timeout of about 1 to 2 seconds;
MySQL's default `innodb_lock_wait_timeout` is 50 seconds; PostgreSQL
waits by default (`lock_timeout` = 0).
Fix: #13776
### Does this PR introduce _any_ user-facing change?
Yes. A write that times out waiting for a row lock now fails with 409
(`OptimisticLockException`, "... could not be changed because a
concurrent operation held its lock longer than the backend's lock wait
timeout; retry the operation") instead of 500, except for the
post-submission job-registration failure described above. No API or
property changes.
Known limits:
- If cleaning up a newly created Lance dataset fails, the request still
returns 409 and the dataset may remain. Check and clean up that dataset
before retrying the create.
- If an external table, view or schema was already dropped in the
catalog, or an Iceberg REST change was already committed, and the store
write then times out, a retry does not redo the store write: the stale
Gravitino registration stays, as for any other store failure after the
external change.
- Loading an object that exists only in the external catalog imports it,
so a load can return 409 if that import times out; the load can be
retried.
- For a rename, the message names the new name, because services report
the entity they write.
- PostgreSQL only reports 55P03 when `lock_timeout` is set; by default
it waits.
- When a job registration conflicts, the execution keeps its staging
directory, because cancellation is asynchronous and the execution may
still be writing to it.
- The `EntityStore` interface still does not document
`OptimisticLockException`; that is #13622.
- Bounded server-side retries are deferred from this classification fix.
A safe retry must own the entire metadata transaction, roll it back and
close it, reread/recompute state in a fresh transaction, and replay only
audited callbacks without external side effects. Some backend calls join
an existing transaction, and update accepts a caller-provided function,
so wrapping every store write is not a safe retry boundary. A deadline
checked only between attempts cannot bound a running database lock wait
(MySQL defaults to 50 seconds); the design also needs per-attempt
timeout budgeting, capped backoff with jitter, and interruption
handling. Retry coverage should include contention that clears,
exhausted budgets, non-timeout errors, nested transactions, and no
replay of external work. This PR retains immediate conflict propagation
until those guarantees are implemented.
- Not covered: `JdbcPartitionStatisticStorage` (separate data source),
and the Lance and Iceberg REST services, which still map the conflict to
500. For Iceberg REST that is deliberate, since the Iceberg change is
already committed and a client retry is not safe; Lance REST can map it
to its 409 in a follow-up.
### How was this patch tested?
- `TestRelationalEntityStoreLockWaitTimeout` (new): real H2 lock-wait
timeouts with another session holding the row `FOR UPDATE`, for a table
drop, an alter, a create waiting on its schema row, and a tag
association; the failed write is rolled back and a retry succeeds. Every
store write is also checked with a failing backend, and deadlocks and
other failures are checked to stay unchanged.
- `TestRelationalEntityStoreBackendLockWaitTimeout`: real held-row
timeouts through the store on H2, MySQL and PostgreSQL, for drop and
create. Checks the real driver code, closed writer session, unchanged
data and a successful retry; the writer owns its normal transaction,
without a test-managed rollback. Tagged `gravitino-docker-test`.
- `TestExceptionUtils`: each backend code, chained batch exceptions,
null SQLStates, shared causes and cycles; deadlocks remain unmatched and
conflict conversion preserves the cause.
- `TestSchemaOperationDispatcher`, `TestTableOperationDispatcher`,
`TestTopicOperationDispatcher`, `TestViewOperationDispatcher`: conflicts
on drop and import reach the caller.
- `TestLanceTableOperations`: create drops the dataset and drop/purge
propagate the conflict. Native cleanup `IOException`s preserve the
original conflict or already-exists error and attach the cleanup failure
as suppressed evidence.
- `TestCatalogManager`: a managed catalog force drop propagates a schema
conflict. `TestJobManager`: registration conflicts attempt cancellation
and retain diagnostic details without returning a retryable conflict,
including cancellation failures. `TestExceptionHandlers` verifies that
the post-submission failure remains HTTP 500.
- Existing `TestStatisticMetaService`, `TestTagMetaService`,
`TestModelVersionMetaService`, `TestTagManager` pass.
- Mutation checks: next-exception traversal and post-submission job
semantics were checked in the preceding round. In this round, disabling
the timeout codes makes all three real-backend drop tests fail;
restricting cleanup to `RuntimeException` makes both native-cleanup
tests fail. All mutations were reverted.
- Earlier validation (before the latest review fixes): 466 tests passed
(326 filtered core, 36 server, all 104 non-Docker lakehouse module
cases), plus 6 real held-row timeout cases across H2, MySQL and
PostgreSQL. Scoped Spotless checks passed; the new real-backend fixture
closes pools/engines and leaves none of its H2 files.
- Real held-row timeout and retry tests now cover H2, MySQL and
PostgreSQL for drop/create. Dispatcher tests simulate metadata failures;
external-system end-to-end recovery remains outside this PR.
Latest review-fix validation:
- 452 tests passed, with 2 backend-specific statistics cases skipped as
designed: 342 filtered core cases plus all 110 lakehouse unit cases.
Storage-service tests and real held-row timeout cases ran on H2, MySQL
and PostgreSQL.
- The six new Lance cases exercise both load repair paths and all three
timeout codes, asserting one store update, no backoff, and the original
conflict/cause. The mixed-chain classifier test checks both exception
orders for all three backend codes and the association
pre-classification helper. Mixed chains are constructed in the
classifier test; this round did not reproduce a mixed
duplicate-key/timeout chain through a production batch path.
- Mutation validation removed the timeout retry guard and duplicate-key
precedence: all six new Lance cases and the mixed-chain classifier case
failed. Sources were restored and 196 relevant core/lakehouse cases
passed again. Scoped Spotless checks passed.
---------
Co-authored-by: yuqi <[email protected]>
---
.../lakehouse/lance/LanceTableOperations.java | 34 +-
.../lakehouse/lance/TestLanceTableOperations.java | 146 ++++++
.../apache/gravitino/catalog/CatalogManager.java | 4 +
.../catalog/SchemaOperationDispatcher.java | 6 +
.../catalog/TableOperationDispatcher.java | 3 +
.../catalog/TopicOperationDispatcher.java | 6 +
.../gravitino/catalog/ViewOperationDispatcher.java | 6 +
.../java/org/apache/gravitino/job/JobManager.java | 36 +-
.../storage/relational/RelationalEntityStore.java | 107 ++++-
.../service/ModelVersionMetaService.java | 2 +
.../relational/service/StatisticMetaService.java | 48 +-
.../storage/relational/service/TagMetaService.java | 6 +
.../storage/relational/utils/ExceptionUtils.java | 186 ++++++++
.../java/org/apache/gravitino/tag/TagManager.java | 4 +-
.../gravitino/catalog/TestCatalogManager.java | 45 ++
.../catalog/TestSchemaOperationDispatcher.java | 40 ++
.../catalog/TestTableOperationDispatcher.java | 31 ++
.../catalog/TestTopicOperationDispatcher.java | 43 ++
.../catalog/TestViewOperationDispatcher.java | 50 ++
.../org/apache/gravitino/job/TestJobManager.java | 35 ++
...elationalEntityStoreBackendLockWaitTimeout.java | 206 ++++++++
.../TestRelationalEntityStoreLockWaitTimeout.java | 518 +++++++++++++++++++++
.../service/TestStatisticMetaService.java | 48 +-
.../relational/utils/TestExceptionUtils.java | 204 ++++++++
.../server/web/rest/TestExceptionHandlers.java | 17 +
25 files changed, 1752 insertions(+), 79 deletions(-)
diff --git
a/catalogs/catalog-lakehouse-generic/src/main/java/org/apache/gravitino/catalog/lakehouse/lance/LanceTableOperations.java
b/catalogs/catalog-lakehouse-generic/src/main/java/org/apache/gravitino/catalog/lakehouse/lance/LanceTableOperations.java
index 1ed701113b..48d7d955d9 100644
---
a/catalogs/catalog-lakehouse-generic/src/main/java/org/apache/gravitino/catalog/lakehouse/lance/LanceTableOperations.java
+++
b/catalogs/catalog-lakehouse-generic/src/main/java/org/apache/gravitino/catalog/lakehouse/lance/LanceTableOperations.java
@@ -68,6 +68,7 @@ import org.apache.gravitino.rel.expressions.sorts.SortOrder;
import org.apache.gravitino.rel.expressions.transforms.Transform;
import org.apache.gravitino.rel.indexes.Index;
import org.apache.gravitino.storage.IdGenerator;
+import org.apache.gravitino.storage.relational.utils.ExceptionUtils;
import org.apache.gravitino.utils.ExceptionMessages;
import org.apache.gravitino.utils.PrincipalUtils;
import org.lance.Dataset;
@@ -317,6 +318,10 @@ public class LanceTableOperations extends
ManagedTableOperations {
return false;
} catch (IllegalArgumentException e) {
throw e;
+ } catch (OptimisticLockException e) {
+ // The table metadata conflicted with a concurrent writer, e.g. a lock
wait timeout, and was
+ // not changed, so the dataset is kept and the operation can be retried.
+ throw e;
} catch (Exception e) {
throw ExceptionMessages.wrap("Failed to purge Lance dataset for table "
+ ident, e);
}
@@ -354,6 +359,10 @@ public class LanceTableOperations extends
ManagedTableOperations {
return false;
} catch (IllegalArgumentException e) {
throw e;
+ } catch (OptimisticLockException e) {
+ // The table metadata conflicted with a concurrent writer, e.g. a lock
wait timeout, and was
+ // not changed, so the dataset is kept and the operation can be retried.
+ throw e;
} catch (Exception e) {
throw ExceptionMessages.wrap("Failed to drop Lance dataset for table " +
ident, e);
}
@@ -440,10 +449,19 @@ public class LanceTableOperations extends
ManagedTableOperations {
indexes);
} catch (NoSuchSchemaException e) {
throw e;
- } catch (TableAlreadyExistsException e) {
- // If the table metadata already exists, but the underlying lance table
was just created
- // successfully, we need to clean up the created lance table to avoid
orphaned datasets.
- Dataset.drop(location, storageProps);
+ } catch (TableAlreadyExistsException | OptimisticLockException e) {
+ // Metadata registration failed after creating the dataset. Attempt to
remove it so a retry
+ // can create it again; if cleanup fails, retain the registration error
and cleanup failure.
+ try {
+ Dataset.drop(location, storageProps);
+ } catch (Exception dropFailure) {
+ // JNI can throw checked exceptions even though Dataset.drop declares
none.
+ LOG.warn(
+ "Failed to drop the Lance dataset at {} after metadata
registration failed",
+ location,
+ dropFailure);
+ e.addSuppressed(dropFailure);
+ }
throw e;
} catch (IllegalArgumentException e) {
if (e.getMessage().contains("Dataset already exists")) {
@@ -626,8 +644,9 @@ public class LanceTableOperations extends
ManagedTableOperations {
*
* <p>Two concurrent loads can both read the old metadata. The first update
wins; the second gets
* {@link OptimisticLockException}. Repair is safe to run more than once, so
the loser waits
- * briefly, reads the winner's latest row, and tries again. Ordinary IO
failures are not safe to
- * retry here and are returned immediately.
+ * briefly, reads the winner's latest row, and tries again. Lock wait
timeouts are returned
+ * immediately: each attempt has already consumed the backend's lock wait
budget. Ordinary IO
+ * failures are not safe to retry here and are returned immediately.
*/
private TableEntity updateTableWithCasRetry(
NameIdentifier ident, Function<TableEntity, TableEntity> updater) throws
IOException {
@@ -636,6 +655,9 @@ public class LanceTableOperations extends
ManagedTableOperations {
try {
return store.update(ident, TableEntity.class, Entity.EntityType.TABLE,
updater);
} catch (OptimisticLockException e) {
+ if (ExceptionUtils.isLockWaitTimeout(e)) {
+ throw e;
+ }
lastConflict = e;
LOG.debug(
"Optimistic-lock conflict updating table {} metadata (attempt
{}/{}), {}",
diff --git
a/catalogs/catalog-lakehouse-generic/src/test/java/org/apache/gravitino/catalog/lakehouse/lance/TestLanceTableOperations.java
b/catalogs/catalog-lakehouse-generic/src/test/java/org/apache/gravitino/catalog/lakehouse/lance/TestLanceTableOperations.java
index 6d688b0d7b..bf48d6464f 100644
---
a/catalogs/catalog-lakehouse-generic/src/test/java/org/apache/gravitino/catalog/lakehouse/lance/TestLanceTableOperations.java
+++
b/catalogs/catalog-lakehouse-generic/src/test/java/org/apache/gravitino/catalog/lakehouse/lance/TestLanceTableOperations.java
@@ -25,9 +25,11 @@ import static
org.apache.gravitino.lance.common.utils.LanceConstants.LANCE_TABLE
import static
org.apache.gravitino.lance.common.utils.LanceConstants.LANCE_TABLE_REGISTER;
import static
org.apache.gravitino.lance.common.utils.LanceConstants.LANCE_TABLE_VERSION;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.anyList;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.spy;
@@ -36,6 +38,9 @@ import static org.mockito.Mockito.when;
import com.google.common.collect.Maps;
import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.sql.SQLException;
import java.time.Instant;
import java.util.List;
import java.util.Map;
@@ -47,9 +52,11 @@ import org.apache.arrow.vector.types.pojo.Schema;
import org.apache.gravitino.Entity;
import org.apache.gravitino.EntityStore;
import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.StringIdentifier;
import org.apache.gravitino.UserPrincipal;
import org.apache.gravitino.catalog.ManagedSchemaOperations;
import org.apache.gravitino.exceptions.OptimisticLockException;
+import org.apache.gravitino.exceptions.TableAlreadyExistsException;
import org.apache.gravitino.lance.common.ops.gravitino.LanceDataTypeConverter;
import org.apache.gravitino.meta.AuditInfo;
import org.apache.gravitino.meta.ColumnEntity;
@@ -63,6 +70,7 @@ import
org.apache.gravitino.rel.expressions.transforms.Transform;
import org.apache.gravitino.rel.indexes.Index;
import org.apache.gravitino.rel.types.Types;
import org.apache.gravitino.storage.IdGenerator;
+import org.apache.gravitino.storage.relational.utils.ExceptionUtils;
import org.apache.gravitino.utils.PrincipalUtils;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
@@ -198,6 +206,99 @@ public class TestLanceTableOperations {
verify(store, never()).delete(any(), any());
}
+ /**
+ * Verifies a conflict of the table metadata, such as a lock wait timeout,
reaches the caller as a
+ * conflict and the dataset created for the table is dropped again, so that
a retry can create it.
+ */
+ @Test
+ public void testCreateTableDropsDatasetOnMetadataConflict() throws
IOException {
+ NameIdentifier ident = NameIdentifier.of("catalog", "schema",
"table_conflict");
+ Column[] columns = new Column[] {Column.of("id", Types.IntegerType.get(),
"id column")};
+ String location = tempDir.resolve("conflict-create").toString();
+ Map<String, String> properties = Maps.newHashMap();
+ properties.put(Table.PROPERTY_LOCATION, location);
+ properties.put(Table.PROPERTY_TABLE_FORMAT, "lance");
+ when(idGenerator.nextId()).thenReturn(2L);
+ doThrow(new OptimisticLockException("mock
conflict")).when(store).put(any(), anyBoolean());
+
+ Assertions.assertThrows(
+ OptimisticLockException.class,
+ () ->
+ lanceTableOps.createTable(
+ ident,
+ columns,
+ null,
+
StringIdentifier.newPropertiesWithId(StringIdentifier.fromId(1L), properties),
+ new Transform[0],
+ null,
+ new SortOrder[0],
+ new Index[0]));
+ Assertions.assertFalse(
+ Files.exists(Paths.get(location, "_versions")), "the created dataset
must be dropped");
+ }
+
+ /** Verifies failed cleanup preserves the registration conflict and the
cleanup failure. */
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void
testCreateTablePreservesRegistrationFailureWhenDatasetCleanupFails(boolean
conflict)
+ throws IOException {
+ NameIdentifier ident = NameIdentifier.of("catalog", "schema",
"table_cleanup_conflict");
+ Column[] columns = new Column[] {Column.of("id", Types.IntegerType.get(),
"id column")};
+ String location = tempDir.resolve("conflict-cleanup").toString();
+ Map<String, String> properties =
+ Map.of(Table.PROPERTY_LOCATION, location, Table.PROPERTY_TABLE_FORMAT,
"lance");
+ Dataset dataset = mock(Dataset.class);
+ when(dataset.version()).thenReturn(1L);
+ Mockito.doReturn(dataset).when(lanceTableOps).createDataset(eq(location),
eq(columns), any());
+ when(idGenerator.nextId()).thenReturn(2L);
+ RuntimeException failure =
+ conflict
+ ? new OptimisticLockException("metadata lock timeout")
+ : new TableAlreadyExistsException("table already registered");
+ doThrow(failure).when(store).put(any(), anyBoolean());
+ // The native drop throws an IOException if the dataset disappeared before
cleanup.
+ RuntimeException actual =
+ Assertions.assertThrows(
+ failure.getClass(),
+ () ->
+ lanceTableOps.createTable(
+ ident,
+ columns,
+ null,
+
StringIdentifier.newPropertiesWithId(StringIdentifier.fromId(1L), properties),
+ new Transform[0],
+ null,
+ new SortOrder[0],
+ new Index[0]));
+ Assertions.assertSame(failure, actual);
+ Assertions.assertEquals(1, actual.getSuppressed().length);
+ Assertions.assertInstanceOf(IOException.class, actual.getSuppressed()[0]);
+ Assertions.assertTrue(actual.getSuppressed()[0].getMessage().contains("Not
found:"));
+ verify(dataset).close();
+ }
+
+ /** Verifies a conflict of the table metadata on drop reaches the caller as
a conflict. */
+ @Test
+ public void testDropTablePropagatesMetadataConflict() throws IOException {
+ NameIdentifier ident = NameIdentifier.of("catalog", "schema",
"table_drop_conflict");
+ when(store.get(eq(ident), eq(Entity.EntityType.TABLE),
eq(TableEntity.class)))
+ .thenReturn(
+ tableEntity(
+ ident,
+ List.of(),
+ Map.of(
+ Table.PROPERTY_LOCATION,
+ tempDir.resolve("conflict-drop").toString(),
+ Table.PROPERTY_TABLE_FORMAT,
+ "lance")));
+ doThrow(new OptimisticLockException("mock conflict"))
+ .when(store)
+ .delete(eq(ident), eq(Entity.EntityType.TABLE));
+
+ Assertions.assertThrows(OptimisticLockException.class, () ->
lanceTableOps.dropTable(ident));
+ Assertions.assertThrows(OptimisticLockException.class, () ->
lanceTableOps.purgeTable(ident));
+ }
+
/** Verifies the purge guard preserves the original
IllegalArgumentException. */
@Test
public void testPurgeTablePropagatesNonLanceGuard() throws IOException {
@@ -373,6 +474,51 @@ public class TestLanceTableOperations {
.update(eq(ident), eq(TableEntity.class), eq(Entity.EntityType.TABLE),
any());
}
+ /** Lock timeouts must not consume the CAS retry budget on either
metadata-repair path. */
+ @ParameterizedTest
+ @CsvSource({
+ "true, 1205, 40001",
+ "true, 50200, HYT00",
+ "true, 0, 55P03",
+ "false, 1205, 40001",
+ "false, 50200, HYT00",
+ "false, 0, 55P03"
+ })
+ public void testLoadDoesNotRetryLockWaitTimeout(
+ boolean nonEmptyDataset, int vendorCode, String sqlState) throws
Exception {
+ NameIdentifier ident;
+ if (nonEmptyDataset) {
+ ident = prepareDeclaredTableForRepair("repair-lock-timeout");
+ } else {
+ ident = NameIdentifier.of("schema", "table");
+ String location = tempDir.resolve("empty-lock-timeout").toString();
+ TableEntity entity = tableEntity(ident, List.of(),
Map.of(Table.PROPERTY_LOCATION, location));
+ when(store.get(eq(ident), eq(Entity.EntityType.TABLE),
eq(TableEntity.class)))
+ .thenReturn(entity);
+ Dataset dataset = mock(Dataset.class);
+ when(dataset.getSchema()).thenReturn(new Schema(List.of()));
+ when(dataset.version()).thenReturn(3L);
+ Mockito.doReturn(dataset).when(lanceTableOps).openDataset(location,
Map.of());
+ }
+ RuntimeException driverFailure =
+ new RuntimeException(new SQLException("lock wait timeout", sqlState,
vendorCode));
+ RuntimeException conflict =
+ ExceptionUtils.lockWaitTimeoutAsConflict(driverFailure,
Entity.EntityType.TABLE, ident);
+ when(store.update(eq(ident), eq(TableEntity.class),
eq(Entity.EntityType.TABLE), any()))
+ .thenThrow(conflict);
+
+ OptimisticLockException failure =
+ Assertions.assertThrows(
+ OptimisticLockException.class,
+ () ->
+ PrincipalUtils.doAs(
+ new UserPrincipal("tester"), () ->
lanceTableOps.loadTable(ident)));
+ Assertions.assertSame(conflict, failure);
+ Assertions.assertSame(driverFailure, failure.getCause());
+ verify(store).update(eq(ident), eq(TableEntity.class),
eq(Entity.EntityType.TABLE), any());
+ verify(lanceTableOps, never()).backoffBeforeRetry(any());
+ }
+
@Test
public void testRepairDoesNotRetryOrdinaryIoFailure() throws Exception {
NameIdentifier ident = prepareDeclaredTableForRepair("repair-io-failure");
diff --git
a/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java
b/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java
index 0ca99e3af8..0342080457 100644
--- a/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java
+++ b/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java
@@ -100,6 +100,7 @@ import
org.apache.gravitino.exceptions.NoSuchMetalakeException;
import org.apache.gravitino.exceptions.NoSuchSchemaException;
import org.apache.gravitino.exceptions.NonEmptyCatalogException;
import org.apache.gravitino.exceptions.NonEmptyEntityException;
+import org.apache.gravitino.exceptions.OptimisticLockException;
import org.apache.gravitino.file.FilesetCatalog;
import org.apache.gravitino.lock.LockType;
import org.apache.gravitino.lock.TreeLockUtils;
@@ -1403,6 +1404,9 @@ public class CatalogManager implements CatalogDispatcher,
Closeable {
if (dropped) {
secretManager.deleteSecretsFromProperties(schemaProps);
}
+ } catch (OptimisticLockException e) {
+ // A conflict with a concurrent writer, such as a lock wait
timeout.
+ throw e;
} catch (Exception e) {
LOG.warn("Failed to drop schema {}",
schema.nameIdentifier());
throw new RuntimeException("Failed to drop schema " +
schema.nameIdentifier(), e);
diff --git
a/core/src/main/java/org/apache/gravitino/catalog/SchemaOperationDispatcher.java
b/core/src/main/java/org/apache/gravitino/catalog/SchemaOperationDispatcher.java
index e2fdd1fdf3..71428deb8a 100644
---
a/core/src/main/java/org/apache/gravitino/catalog/SchemaOperationDispatcher.java
+++
b/core/src/main/java/org/apache/gravitino/catalog/SchemaOperationDispatcher.java
@@ -45,6 +45,7 @@ import org.apache.gravitino.exceptions.NoSuchCatalogException;
import org.apache.gravitino.exceptions.NoSuchEntityException;
import org.apache.gravitino.exceptions.NoSuchSchemaException;
import org.apache.gravitino.exceptions.NonEmptySchemaException;
+import org.apache.gravitino.exceptions.OptimisticLockException;
import org.apache.gravitino.exceptions.SchemaAlreadyExistsException;
import org.apache.gravitino.lock.LockType;
import org.apache.gravitino.lock.TreeLockUtils;
@@ -580,6 +581,8 @@ public class SchemaOperationDispatcher extends
OperationDispatcher implements Sc
droppedFromStore = store.delete(ident, SCHEMA, true);
} catch (NoSuchEntityException e) {
LOG.warn("The schema to be dropped does not exist in the store:
{}", ident, e);
+ } catch (OptimisticLockException e) {
+ throw e;
} catch (Exception e) {
throw new RuntimeException(e);
}
@@ -668,6 +671,9 @@ public class SchemaOperationDispatcher extends
OperationDispatcher implements Sc
store.put(schemaEntity, true);
} catch (EntityAlreadyExistsException e) {
throw e;
+ } catch (OptimisticLockException e) {
+ // A conflict with a concurrent writer, such as a lock wait timeout; the
load can be retried.
+ throw e;
} catch (Exception e) {
LOG.error(FormattedErrorMessages.STORE_OP_FAILURE, "put", identifier, e);
throw new RuntimeException("Failed to import schema entity to the
store", e);
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 ce9593d586..4b3381ae66 100644
---
a/core/src/main/java/org/apache/gravitino/catalog/TableOperationDispatcher.java
+++
b/core/src/main/java/org/apache/gravitino/catalog/TableOperationDispatcher.java
@@ -589,6 +589,9 @@ public class TableOperationDispatcher extends
OperationDispatcher implements Tab
store.put(tableEntity, stringId != null);
} catch (EntityAlreadyExistsException e) {
throw e;
+ } catch (OptimisticLockException e) {
+ // A conflict with a concurrent writer, such as a lock wait timeout; the
load can be retried.
+ throw e;
} catch (Exception e) {
LOG.error(FormattedErrorMessages.STORE_OP_FAILURE, "put", identifier, e);
throw new RuntimeException("Failed to import the table entity to the
store", e);
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 f6e683baee..b1bc4871e4 100644
---
a/core/src/main/java/org/apache/gravitino/catalog/TopicOperationDispatcher.java
+++
b/core/src/main/java/org/apache/gravitino/catalog/TopicOperationDispatcher.java
@@ -36,6 +36,7 @@ import org.apache.gravitino.connector.capability.Capability;
import org.apache.gravitino.exceptions.NoSuchEntityException;
import org.apache.gravitino.exceptions.NoSuchSchemaException;
import org.apache.gravitino.exceptions.NoSuchTopicException;
+import org.apache.gravitino.exceptions.OptimisticLockException;
import org.apache.gravitino.exceptions.TopicAlreadyExistsException;
import org.apache.gravitino.lock.LockType;
import org.apache.gravitino.lock.TreeLockUtils;
@@ -254,6 +255,8 @@ public class TopicOperationDispatcher extends
OperationDispatcher implements Top
droppedFromStore = store.delete(ident, TOPIC);
} catch (NoSuchEntityException e) {
LOG.warn("The topic to be dropped does not exist in the store:
{}", ident, e);
+ } catch (OptimisticLockException e) {
+ throw e;
} catch (Exception e) {
throw new RuntimeException(e);
}
@@ -310,6 +313,9 @@ public class TopicOperationDispatcher extends
OperationDispatcher implements Top
try {
store.put(topicEntity, true);
+ } catch (OptimisticLockException e) {
+ // A conflict with a concurrent writer, such as a lock wait timeout; the
load can be retried.
+ throw e;
} catch (Exception e) {
LOG.error(FormattedErrorMessages.STORE_OP_FAILURE, "put", identifier, e);
throw new RuntimeException("Failed to import topic entity to the store",
e);
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 9020ceeacf..622e9b6a6d 100644
---
a/core/src/main/java/org/apache/gravitino/catalog/ViewOperationDispatcher.java
+++
b/core/src/main/java/org/apache/gravitino/catalog/ViewOperationDispatcher.java
@@ -44,6 +44,7 @@ import
org.apache.gravitino.exceptions.GravitinoRuntimeException;
import org.apache.gravitino.exceptions.NoSuchEntityException;
import org.apache.gravitino.exceptions.NoSuchSchemaException;
import org.apache.gravitino.exceptions.NoSuchViewException;
+import org.apache.gravitino.exceptions.OptimisticLockException;
import org.apache.gravitino.exceptions.ViewAlreadyExistsException;
import org.apache.gravitino.lock.LockType;
import org.apache.gravitino.lock.TreeLockUtils;
@@ -342,6 +343,8 @@ public class ViewOperationDispatcher extends
OperationDispatcher implements View
store.delete(ident, VIEW);
} catch (NoSuchEntityException e) {
LOG.warn("The view to be dropped does not exist in the store:
{}", ident, e);
+ } catch (OptimisticLockException e) {
+ throw e;
} catch (Exception e) {
throw new RuntimeException(e);
}
@@ -581,6 +584,9 @@ public class ViewOperationDispatcher extends
OperationDispatcher implements View
throw new UnsupportedOperationException(
"View managed by multiple catalogs. This may cause unexpected issues
such as privilege conflicts. "
+ "To resolve: Remove all catalogs managing this view, then
recreate one catalog to ensure single-catalog management.");
+ } catch (OptimisticLockException e) {
+ // A conflict with a concurrent writer, such as a lock wait timeout; the
load can be retried.
+ throw e;
} catch (Exception e) {
LOG.error(FormattedErrorMessages.STORE_OP_FAILURE, "put", ident, e);
throw new RuntimeException("Failed to import the view entity to the
store", e);
diff --git a/core/src/main/java/org/apache/gravitino/job/JobManager.java
b/core/src/main/java/org/apache/gravitino/job/JobManager.java
index b789078389..32e9966b02 100644
--- a/core/src/main/java/org/apache/gravitino/job/JobManager.java
+++ b/core/src/main/java/org/apache/gravitino/job/JobManager.java
@@ -600,6 +600,31 @@ public class JobManager implements JobOperationDispatcher {
"Job template with name %s under metalake %s does not exist",
jobTemplateName,
metalake);
+ } catch (OptimisticLockException e) {
+ // Cancellation is asynchronous and cannot undo work that already
finished. Request it
+ // best-effort, but do not expose a retryable conflict after submitting
an execution.
+ LOG.warn(
+ "Job {} was submitted as execution {} but could not be registered
because of a "
+ + "concurrent operation, cancelling the execution",
+ jobEntity.name(),
+ jobExecutionId,
+ e);
+ try {
+ jobExecutor.cancelJob(jobExecutionId);
+ } catch (Exception cancelFailure) {
+ LOG.error(
+ "Failed to cancel the unregistered execution {} of job {}",
+ jobExecutionId,
+ jobEntity.name(),
+ cancelFailure);
+ e.addSuppressed(cancelFailure);
+ }
+ throw new RuntimeException(
+ String.format(
+ "Failed to register job %s after submitting execution %s;
cancellation was attempted "
+ + "but the execution may already have run. Check the
execution before retrying",
+ jobEntity.name(), jobExecutionId),
+ e);
} catch (IOException e) {
throw new RuntimeException("Failed to register the job entity " +
jobEntity, e);
}
@@ -793,10 +818,11 @@ public class JobManager implements JobOperationDispatcher
{
NameIdentifierUtil.ofJob(metalake, job.name()),
Entity.EntityType.JOB);
deleteJobStagingDir(job);
} catch (OptimisticLockException e) {
- // Keep the files when deletion loses its CAS. The next cleanup
run re-reads the
- // job and checks retention eligibility again; this batch can
process other jobs.
+ // Keep files when deletion conflicts or times out on a lock.
The next cleanup run
+ // re-reads the job and checks retention eligibility again;
other jobs in this batch
+ // can still be processed.
LOG.info(
- "Job {} under metalake {} changed concurrently; deferring
cleanup",
+ "Job {} under metalake {} could not be deleted because of a
concurrent operation; deferring cleanup",
job.name(),
metalake);
} catch (IOException e) {
@@ -1317,9 +1343,9 @@ public class JobManager implements JobOperationDispatcher
{
}));
} catch (OptimisticLockException e) {
// A later poll re-reads both executor state and metadata. Never stop
the scheduled
- // task or replay external submission/cancellation because a metadata
CAS lost.
+ // task or replay external submission/cancellation because a metadata
write conflicted.
LOG.info(
- "Job {} under metalake {} changed concurrently; deferring status
update",
+ "Job {} under metalake {} could not be updated because of a
concurrent operation; deferring status update",
job.name(),
metalake);
return Optional.empty();
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/RelationalEntityStore.java
b/core/src/main/java/org/apache/gravitino/storage/relational/RelationalEntityStore.java
index 3d9026739b..dd61110ffc 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/RelationalEntityStore.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/RelationalEntityStore.java
@@ -27,6 +27,7 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
+import java.util.Locale;
import java.util.Optional;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
@@ -56,9 +57,11 @@ import org.apache.gravitino.cache.EntityCache;
import org.apache.gravitino.cache.EntityCacheKey;
import org.apache.gravitino.cache.NoOpsCache;
import org.apache.gravitino.exceptions.NoSuchEntityException;
+import org.apache.gravitino.exceptions.OptimisticLockException;
import org.apache.gravitino.metrics.MetricsSystem;
import org.apache.gravitino.metrics.source.EntityChangeLogMetricsSource;
import org.apache.gravitino.storage.relational.service.EntityIdService;
+import org.apache.gravitino.storage.relational.utils.ExceptionUtils;
import org.apache.gravitino.utils.Executable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -67,6 +70,14 @@ import org.slf4j.LoggerFactory;
* Relation store to store entities. This means we can store entities in a
relational store. I.e.,
* MySQL, PostgreSQL, etc. If you want to use a different backend, you can
implement the {@link
* RelationalBackend} interface. The default JDBC backend is {@link
JDBCBackend}.
+ *
+ * <p>A write that fails because a statement waited on a row lock longer than
the backend's lock
+ * wait timeout is reported as an {@link OptimisticLockException}, so callers
see a conflict rather
+ * than an internal error. The store write is rolled back; retrying the whole
operation also
+ * requires accounting for any preceding external side effects. Writers queue
on the same rows
+ * inside the database, for example a table drop behind a statistic update of
that table, and the
+ * backend aborts the waiting statement once the timeout is reached. See {@link
+ * ExceptionUtils#isLockWaitTimeout}.
*/
public class RelationalEntityStore
implements EntityStore, SupportsRelationOperations,
SupportsEntityChangeLog {
@@ -215,7 +226,11 @@ public class RelationalEntityStore
@Override
public <E extends Entity & HasIdentifier> void put(E e, boolean overwritten)
throws IOException, EntityAlreadyExistsException {
- backend.insert(e, overwritten);
+ try {
+ backend.insert(e, overwritten);
+ } catch (RuntimeException re) {
+ throw ExceptionUtils.lockWaitTimeoutAsConflict(re, e.type(),
e.nameIdentifier());
+ }
if (overwritten) {
// An overwrite is resolved by the database, which may keep the identity
and version of the
// row it already had. Caching the copy handed in here would publish
values the stored row
@@ -230,7 +245,12 @@ public class RelationalEntityStore
public <E extends Entity & HasIdentifier> E update(
NameIdentifier ident, Class<E> type, Entity.EntityType entityType,
Function<E, E> updater)
throws IOException, NoSuchEntityException, EntityAlreadyExistsException {
- E updatedEntity = backend.update(ident, entityType, updater);
+ E updatedEntity;
+ try {
+ updatedEntity = backend.update(ident, entityType, updater);
+ } catch (RuntimeException re) {
+ throw ExceptionUtils.lockWaitTimeoutAsConflict(re, entityType, ident);
+ }
invalidateCache(ident, entityType);
return updatedEntity;
}
@@ -308,6 +328,8 @@ public class RelationalEntityStore
return deleted;
} catch (NoSuchEntityException e) {
return false;
+ } catch (RuntimeException re) {
+ throw ExceptionUtils.lockWaitTimeoutAsConflict(re, entityType, ident);
} finally {
invalidateCache(ident, entityType);
}
@@ -322,6 +344,8 @@ public class RelationalEntityStore
throws IOException {
try {
return backend.deleteAndGet(ident, entityType, clazz, postDeleteAction);
+ } catch (RuntimeException re) {
+ throw ExceptionUtils.lockWaitTimeoutAsConflict(re, entityType, ident);
} finally {
invalidateCache(ident, entityType);
}
@@ -424,7 +448,17 @@ public class RelationalEntityStore
Entity.EntityType dstType,
boolean override)
throws IOException {
- backend.insertRelation(relType, srcIdentifier, srcType, dstIdentifier,
dstType, override);
+ try {
+ backend.insertRelation(relType, srcIdentifier, srcType, dstIdentifier,
dstType, override);
+ } catch (RuntimeException re) {
+ throw ExceptionUtils.lockWaitTimeoutAsConflict(
+ re,
+ String.format(
+ "The %s relation of %s %s",
+ relType.name().toLowerCase(Locale.ROOT),
+ srcType.name().toLowerCase(Locale.ROOT),
+ srcIdentifier));
+ }
// Relation query results themselves are not cached, but both endpoints
may be cached entities
// (OWNER_REL, TAG_METADATA_OBJECT_REL and METADATA_OBJECT_ROLE_REL are
keyed by
// catalog/schema/table/... on the source side), so drop their entries
conservatively: a
@@ -447,8 +481,19 @@ public class RelationalEntityStore
if (srcIdentifiers == null || srcIdentifiers.isEmpty()) {
return;
}
- backend.batchInsertRelations(
- relType, srcIdentifiers, srcType, dstIdentifier, dstType, override);
+ try {
+ backend.batchInsertRelations(
+ relType, srcIdentifiers, srcType, dstIdentifier, dstType, override);
+ } catch (RuntimeException re) {
+ throw ExceptionUtils.lockWaitTimeoutAsConflict(
+ re,
+ String.format(
+ "The %s relations of %d %s entities starting with %s",
+ relType.name().toLowerCase(Locale.ROOT),
+ srcIdentifiers.size(),
+ srcType.name().toLowerCase(Locale.ROOT),
+ srcIdentifiers.get(0)));
+ }
// Invalidate both endpoints for the same reason as insertRelation,
including the hierarchy
// cascade noted there.
for (NameIdentifier ident : srcIdentifiers) {
@@ -476,9 +521,14 @@ public class RelationalEntityStore
return updateEntityRelations(update);
}
- List<E> result =
- backend.updateEntityRelations(
- relType, srcEntityIdent, srcEntityType, destEntitiesToAdd,
destEntitiesToRemove);
+ List<E> result;
+ try {
+ result =
+ backend.updateEntityRelations(
+ relType, srcEntityIdent, srcEntityType, destEntitiesToAdd,
destEntitiesToRemove);
+ } catch (RuntimeException re) {
+ throw ExceptionUtils.lockWaitTimeoutAsConflict(re, srcEntityType,
srcEntityIdent);
+ }
Entity.EntityType targetEntityType = relationUpdateTargetType(relType);
invalidateCache(srcEntityIdent, srcEntityType);
invalidateRelationTargetCache(targetEntityType, update.targetsToAdd());
@@ -508,7 +558,13 @@ public class RelationalEntityStore
RelationEdgeTarget[] targetsToAdd = update.targetsToAdd();
RelationEdgeTarget[] targetsToRemove = update.targetsToRemove();
- List<E> result = backend.updateEntityRelations(update);
+ List<E> result;
+ try {
+ result = backend.updateEntityRelations(update);
+ } catch (RuntimeException re) {
+ throw ExceptionUtils.lockWaitTimeoutAsConflict(
+ re, update.sourceEntityType(), update.sourceIdentifier());
+ }
// Invalidate after the backend write, not before: invalidating first
opens a window where a
// concurrent read could repopulate the cache with stale pre-commit data.
@@ -524,13 +580,42 @@ public class RelationalEntityStore
public int batchDelete(
List<Pair<NameIdentifier, Entity.EntityType>> entitiesToDelete, boolean
cascade)
throws IOException {
- return backend.batchDelete(entitiesToDelete, cascade);
+ try {
+ return backend.batchDelete(entitiesToDelete, cascade);
+ } catch (RuntimeException re) {
+ throw ExceptionUtils.lockWaitTimeoutAsConflict(
+ re,
+ describeBatch(
+ entitiesToDelete.size(),
+ entitiesToDelete.isEmpty() ? null :
entitiesToDelete.get(0).getRight(),
+ entitiesToDelete.isEmpty() ? null :
entitiesToDelete.get(0).getLeft()));
+ }
}
@Override
public <E extends Entity & HasIdentifier> void batchPut(List<E> entities,
boolean overwritten)
throws IOException, EntityAlreadyExistsException {
- backend.batchPut(entities, overwritten);
+ try {
+ backend.batchPut(entities, overwritten);
+ } catch (RuntimeException re) {
+ throw ExceptionUtils.lockWaitTimeoutAsConflict(
+ re,
+ describeBatch(
+ entities.size(),
+ entities.isEmpty() ? null : entities.get(0).type(),
+ entities.isEmpty() ? null : entities.get(0).nameIdentifier()));
+ }
+ }
+
+ /** Describes a batch write by its size and its first entity, for a conflict
message. */
+ private static String describeBatch(
+ int size, @Nullable Entity.EntityType firstType, @Nullable
NameIdentifier first) {
+ if (firstType == null) {
+ return String.format("The batch of %d entities", size);
+ }
+ return String.format(
+ "The batch of %d entities starting with %s %s",
+ size, firstType.name().toLowerCase(Locale.ROOT), first);
}
private void invalidateRelationTargetCache(
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/service/ModelVersionMetaService.java
b/core/src/main/java/org/apache/gravitino/storage/relational/service/ModelVersionMetaService.java
index 19db5554f1..71edd37485 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/service/ModelVersionMetaService.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/service/ModelVersionMetaService.java
@@ -128,6 +128,8 @@ public class ModelVersionMetaService {
mapper -> mapper.insertModelVersionAliasRels(aliasRelPOs));
});
} catch (RuntimeException re) {
+ ExceptionUtils.throwIfLockWaitTimeout(
+ re, "A version of the model " +
modelVersionEntity.modelIdentifier());
ExceptionUtils.checkSQLException(
re, Entity.EntityType.MODEL_VERSION,
modelVersionEntity.modelIdentifier().toString());
throw re;
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/service/StatisticMetaService.java
b/core/src/main/java/org/apache/gravitino/storage/relational/service/StatisticMetaService.java
index 7636af1356..63f87cf978 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/service/StatisticMetaService.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/service/StatisticMetaService.java
@@ -21,7 +21,6 @@ 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 java.sql.SQLException;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashMap;
@@ -46,6 +45,7 @@ import org.apache.gravitino.metrics.Monitored;
import org.apache.gravitino.storage.relational.mapper.StatisticMetaMapper;
import
org.apache.gravitino.storage.relational.mapper.StatisticSQLProviderFactory;
import org.apache.gravitino.storage.relational.po.StatisticPO;
+import org.apache.gravitino.storage.relational.utils.ExceptionUtils;
import org.apache.gravitino.storage.relational.utils.SessionUtils;
import org.apache.gravitino.utils.NameIdentifierUtil;
@@ -55,9 +55,7 @@ import org.apache.gravitino.utils.NameIdentifierUtil;
*/
public class StatisticMetaService {
- private static final int MYSQL_LOCK_WAIT_TIMEOUT = 1205;
private static final int MYSQL_DEADLOCK = 1213;
- private static final int H2_LOCK_TIMEOUT = 50200;
private static final StatisticMetaService INSTANCE = new
StatisticMetaService();
@@ -288,7 +286,7 @@ public class StatisticMetaService {
if (failure instanceof OptimisticLockException) {
return failure;
}
- if (isDuplicateKey(failure)) {
+ if (ExceptionUtils.isDuplicateKey(failure)) {
return statisticConflict(failure, names(pos), target);
}
if (isConcurrencyFailure(failure)) {
@@ -357,43 +355,19 @@ public class StatisticMetaService {
.build();
}
- /**
- * Returns whether a failure is a unique-key violation. PostgreSQL and H2
report SQLState 23505,
- * and MySQL reports error code 1062, the same codes the backend exception
converters match.
- */
- private static boolean isDuplicateKey(Throwable failure) {
- return hasSqlException(failure, Set.of("23505"), Set.of(1062));
- }
-
/**
* Returns whether a failure is a deadlock, serialization failure or lock
wait timeout caused by a
- * concurrent writer. The accepted codes intentionally cover more than the
current default
- * configuration emits. PostgreSQL reports 40P01 for deadlocks at our READ
COMMITTED isolation;
- * 40001 also covers serialization failures at stricter isolation levels,
and 55P03 covers a
- * server-configured lock_timeout. H2 reports 40001 for a deadlock and error
code 50200 (SQLState
- * HYT00) for a lock wait timeout. MySQL reports error code 1213 for a
deadlock and 1205 for a
- * lock wait timeout. Vendor error codes are matched because a timeout's
SQLState depends on the
- * driver.
+ * concurrent writer. Lock wait timeouts are classified by {@link
+ * ExceptionUtils#isLockWaitTimeout}, as for every other store write. Unlike
other writes, a batch
+ * statement here locks its rows in the order of the database's plan, which
differs between
+ * batches, so deadlocks are expected and accepted as conflicts too.
PostgreSQL reports 40P01 for
+ * deadlocks at our READ COMMITTED isolation; 40001 also covers
serialization failures at stricter
+ * isolation levels. H2 reports 40001 for a deadlock, and MySQL reports
error code 1213.
*/
private static boolean isConcurrencyFailure(Throwable failure) {
- return hasSqlException(
- failure,
- Set.of("40001", "40P01", "55P03"),
- Set.of(MYSQL_DEADLOCK, MYSQL_LOCK_WAIT_TIMEOUT, H2_LOCK_TIMEOUT));
- }
-
- private 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;
+ return ExceptionUtils.isLockWaitTimeout(failure)
+ || ExceptionUtils.hasSqlException(
+ failure, Set.of("40001", "40P01"), Set.of(MYSQL_DEADLOCK));
}
@Monitored(
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 100c8d7e1f..fdfa3c4f20 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
@@ -32,6 +32,7 @@ import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
+import java.util.Locale;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
@@ -540,6 +541,11 @@ public class TagMetaService {
return tagPOsToTagEntities(tagPOs, NamespaceUtil.ofTag(metalake));
} catch (RuntimeException e) {
+ ExceptionUtils.throwIfLockWaitTimeout(
+ e,
+ String.format(
+ "The tags of %s %s",
+ metadataObject.type().name().toLowerCase(Locale.ROOT),
objectIdent));
ExceptionUtils.checkSQLException(e, Entity.EntityType.TAG,
objectIdent.toString());
throw e;
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/utils/ExceptionUtils.java
b/core/src/main/java/org/apache/gravitino/storage/relational/utils/ExceptionUtils.java
index 7da3008500..6f6dcade43 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/utils/ExceptionUtils.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/utils/ExceptionUtils.java
@@ -20,17 +20,65 @@ package org.apache.gravitino.storage.relational.utils;
import java.io.IOException;
import java.sql.SQLException;
+import java.util.ArrayDeque;
+import java.util.Collections;
+import java.util.Deque;
+import java.util.IdentityHashMap;
import java.util.Locale;
+import java.util.Set;
+import javax.annotation.Nullable;
import org.apache.gravitino.Entity;
+import org.apache.gravitino.EntityAlreadyExistsException;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.exceptions.OptimisticLockException;
import
org.apache.gravitino.storage.relational.converters.SQLExceptionConverterFactory;
public class ExceptionUtils {
+
+ /** MySQL error code of a lock wait timeout (ER_LOCK_WAIT_TIMEOUT). */
+ private static final int MYSQL_LOCK_WAIT_TIMEOUT = 1205;
+
+ /** H2 error code of a lock wait timeout (LOCK_TIMEOUT_1, SQLState HYT00). */
+ private static final int H2_LOCK_TIMEOUT = 50200;
+
+ /** PostgreSQL SQLState of a lock that is not available within lock_timeout.
*/
+ private static final String POSTGRESQL_LOCK_NOT_AVAILABLE = "55P03";
+
+ private static final Set<String> LOCK_WAIT_TIMEOUT_SQL_STATES =
+ Set.of(POSTGRESQL_LOCK_NOT_AVAILABLE);
+
+ private static final Set<Integer> LOCK_WAIT_TIMEOUT_VENDOR_CODES =
+ Set.of(MYSQL_LOCK_WAIT_TIMEOUT, H2_LOCK_TIMEOUT);
+
+ private static final Set<String> DUPLICATE_KEY_SQL_STATES = Set.of("23505");
+ private static final Set<Integer> DUPLICATE_KEY_VENDOR_CODES = Set.of(1062);
+
private ExceptionUtils() {}
+ /**
+ * Converts a failure of a write whose cause is a {@link SQLException} into
the matching Gravitino
+ * exception. A lock wait timeout becomes an {@link
OptimisticLockException}, see {@link
+ * #lockWaitTimeoutAsConflict(RuntimeException, String)}. It is classified
here, before the
+ * backend converter, because the converter reports an unrecognised SQL
error as an {@link
+ * IOException}, which callers turn into an internal error. A duplicate-key
failure takes
+ * precedence over a timeout in the same exception chain: retrying cannot
resolve that violation.
+ *
+ * @param re The failure of the write.
+ * @param type The type of the entity being written.
+ * @param entityName The name of the entity being written.
+ * @throws IOException If the SQL error has no more specific mapping.
+ * @throws EntityAlreadyExistsException If the failure contains a
duplicate-key violation.
+ * @throws OptimisticLockException If the failure contains a lock wait
timeout without a
+ * duplicate-key violation.
+ */
public static void checkSQLException(
RuntimeException re, Entity.EntityType type, String entityName) throws
IOException {
+ if (isDuplicateKey(re)) {
+ throw new EntityAlreadyExistsException(
+ re, "The %s entity: %s already exists.", type.name(), entityName);
+ }
+ throwIfLockWaitTimeout(
+ re, String.format("The %s %s", type.name().toLowerCase(Locale.ROOT),
entityName));
if (re.getCause() instanceof SQLException) {
SQLExceptionConverterFactory.getConverter()
.toGravitinoException((SQLException) re.getCause(), type,
entityName);
@@ -69,4 +117,142 @@ public class ExceptionUtils {
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
without a duplicate-key
+ * violation.
+ * @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 and chained
SQL exceptions, or
+ * null.
+ * @return {@code true} if a lock wait timeout is found.
+ */
+ public static boolean isLockWaitTimeout(@Nullable Throwable failure) {
+ return hasSqlException(failure, LOCK_WAIT_TIMEOUT_SQL_STATES,
LOCK_WAIT_TIMEOUT_VENDOR_CODES);
+ }
+
+ /**
+ * Returns whether a failure contains a {@link SQLException} with one of the
given SQLStates or
+ * vendor error codes. Both causes and {@link
SQLException#getNextException()} are inspected;
+ * repeated exceptions are visited only once, even when the chains contain
cycles.
+ *
+ * @param failure The failure to inspect, including its causes and chained
SQL exceptions, or
+ * null.
+ * @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(
+ @Nullable Throwable failure, Set<String> sqlStates, Set<Integer>
vendorErrorCodes) {
+ if (failure == null) {
+ return false;
+ }
+ Set<Throwable> visited = Collections.newSetFromMap(new
IdentityHashMap<>());
+ Deque<Throwable> pending = new ArrayDeque<>();
+ pending.add(failure);
+ while (!pending.isEmpty()) {
+ Throwable cause = pending.removeLast();
+ if (!visited.add(cause)) {
+ continue;
+ }
+ if (cause instanceof SQLException) {
+ SQLException sql = (SQLException) cause;
+ if ((sql.getSQLState() != null &&
sqlStates.contains(sql.getSQLState()))
+ || vendorErrorCodes.contains(sql.getErrorCode())) {
+ return true;
+ }
+ if (sql.getNextException() != null) {
+ pending.add(sql.getNextException());
+ }
+ }
+ if (cause.getCause() != null) {
+ pending.add(cause.getCause());
+ }
+ }
+ return false;
+ }
+
+ /**
+ * Reports a write of one entity that failed on a lock wait timeout as a
conflict.
+ *
+ * @param failure The failure of the write.
+ * @param type The type of the entity being written.
+ * @param identifier The identifier of the entity being written.
+ * @return An {@link OptimisticLockException} for a lock wait timeout,
otherwise {@code failure}.
+ * @see #lockWaitTimeoutAsConflict(RuntimeException, String)
+ */
+ public static RuntimeException lockWaitTimeoutAsConflict(
+ RuntimeException failure, Entity.EntityType type, NameIdentifier
identifier) {
+ return lockWaitTimeoutAsConflict(
+ failure, String.format("The %s %s",
type.name().toLowerCase(Locale.ROOT), identifier));
+ }
+
+ /**
+ * Reports a write that failed on a lock wait timeout as a conflict. The
statement waited behind a
+ * concurrent writer of the same rows until the backend aborted it, and the
store rolls the
+ * transaction back, so the store write can be retried. Callers must account
for any external side
+ * effects that occurred before the store write before retrying the whole
operation. Any other
+ * failure, including one that is already an {@link
OptimisticLockException}, is returned
+ * unchanged. A failure containing a duplicate-key violation is also
returned unchanged. Callers
+ * that subsequently use {@link #checkSQLException} report that violation as
an already-exists
+ * error. Store write paths that do not use that classifier retain the
original backend failure,
+ * including for a chain containing both a duplicate key and a lock wait
timeout.
+ *
+ * @param failure The failure of the write.
+ * @param target What was being written, as the subject of the message, for
example "The table
+ * m1.c1.s1.t1".
+ * @return An {@link OptimisticLockException} for a lock wait timeout,
otherwise {@code failure}.
+ */
+ public static RuntimeException lockWaitTimeoutAsConflict(
+ RuntimeException failure, String target) {
+ if (failure instanceof OptimisticLockException
+ || !isLockWaitTimeout(failure)
+ || isDuplicateKey(failure)) {
+ return failure;
+ }
+ return new OptimisticLockException(
+ failure,
+ "%s could not be changed because a concurrent operation held its lock
longer than the"
+ + " backend's lock wait timeout; retry the operation",
+ target);
+ }
+
+ /**
+ * Returns whether a failure contains a duplicate primary-key or unique-key
violation.
+ *
+ * <p>Matches MySQL vendor code 1062 and PostgreSQL/H2 SQLState 23505
through causes and chained
+ * JDBC exceptions, using cycle protection. This only detects the violation;
callers decide
+ * whether it represents an already-existing entity or a concurrent
statistics write.
+ *
+ * @param failure the failure to inspect, including its causes and chained
JDBC exceptions, or
+ * null
+ * @return true if a duplicate-key violation is found
+ */
+ public static boolean isDuplicateKey(@Nullable Throwable failure) {
+ return hasSqlException(failure, DUPLICATE_KEY_SQL_STATES,
DUPLICATE_KEY_VENDOR_CODES);
+ }
}
diff --git a/core/src/main/java/org/apache/gravitino/tag/TagManager.java
b/core/src/main/java/org/apache/gravitino/tag/TagManager.java
index 77e6e4b4a9..b35731eed4 100644
--- a/core/src/main/java/org/apache/gravitino/tag/TagManager.java
+++ b/core/src/main/java/org/apache/gravitino/tag/TagManager.java
@@ -179,7 +179,7 @@ public class TagManager implements TagDispatcher {
// and the REST layer maps it to a conflict. Log it here so the
operator-facing record
// still names the tag and the metalake.
LOG.warn(
- "Failed to alter tag {} under metalake {} because it changed
concurrently",
+ "Failed to alter tag {} under metalake {} because of a concurrent
operation",
name,
metalake,
ole);
@@ -196,7 +196,7 @@ public class TagManager implements TagDispatcher {
return entityStore.delete(NameIdentifierUtil.ofTag(metalake, name),
Entity.EntityType.TAG);
} catch (OptimisticLockException ole) {
LOG.warn(
- "Failed to delete tag {} under metalake {} because it changed
concurrently",
+ "Failed to delete tag {} under metalake {} because of a concurrent
operation",
name,
metalake,
ole);
diff --git
a/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java
b/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java
index 58709a12cc..0515edfdab 100644
--- a/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java
+++ b/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java
@@ -78,6 +78,7 @@ import org.apache.gravitino.exceptions.NoSuchCatalogException;
import org.apache.gravitino.exceptions.NoSuchEntityException;
import org.apache.gravitino.exceptions.NoSuchMetalakeException;
import org.apache.gravitino.exceptions.NoSuchSchemaException;
+import org.apache.gravitino.exceptions.OptimisticLockException;
import org.apache.gravitino.lock.LockManager;
import org.apache.gravitino.lock.LockType;
import org.apache.gravitino.lock.TreeLockUtils;
@@ -1564,6 +1565,50 @@ public class TestCatalogManager {
Mockito.verify(catalogWrapper, Mockito.never()).doWithSchemaOps(any());
}
+ @Test
+ public void testForceDropManagedCatalogPropagatesSchemaConflict() throws
Exception {
+ NameIdentifier ident = NameIdentifier.of("metalake",
"test_force_drop_occ");
+ Map<String, String> props =
+ ImmutableMap.of(
+ "provider",
+ "test",
+ PROPERTY_KEY1,
+ "value1",
+ PROPERTY_KEY2,
+ "value2",
+ PROPERTY_KEY5_PREFIX + "1",
+ "value3");
+ catalogManager.createCatalog(ident, Catalog.Type.RELATIONAL, provider,
"comment", props);
+ BaseCatalog<?> catalog =
catalogManager.loadCatalogAndWrap(ident).catalog();
+ entityStore.put(
+ SchemaEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName("schema_occ")
+ .withNamespace(Namespace.of("metalake", "test_force_drop_occ"))
+ .withAuditInfo(
+ AuditInfo.builder()
+
.withCreator(PrincipalUtils.getCurrentPrincipal().getName())
+ .withCreateTime(Instant.now())
+ .build())
+ .build());
+ CatalogManager.CatalogWrapper catalogWrapper =
+ Mockito.mock(CatalogManager.CatalogWrapper.class);
+ Capability capability = Mockito.mock(Capability.class);
+ Mockito.when(catalogWrapper.tryAcquire()).thenReturn(true);
+ Mockito.doReturn(capability).when(catalogWrapper).capabilities();
+ // A managed-storage catalog drops each schema through its schema
operations.
+
Mockito.doReturn(CapabilityResult.SUPPORTED).when(capability).managedStorage(any());
+ Mockito.doReturn(catalog).when(catalogWrapper).catalog();
+ OptimisticLockException conflict = new OptimisticLockException("lock wait
timeout");
+ Mockito.doThrow(conflict).when(catalogWrapper).doWithSchemaOps(any());
+ catalogManager.getCatalogCache().put(ident, catalogWrapper);
+
+ Assertions.assertSame(
+ conflict,
+ Assertions.assertThrows(
+ OptimisticLockException.class, () ->
catalogManager.dropCatalog(ident, true)));
+ }
+
@Test
void testDropCatalogInvalidatesCacheAfterStoreDelete() throws Exception {
NameIdentifier ident = NameIdentifier.of("metalake", "cache_drop_test");
diff --git
a/core/src/test/java/org/apache/gravitino/catalog/TestSchemaOperationDispatcher.java
b/core/src/test/java/org/apache/gravitino/catalog/TestSchemaOperationDispatcher.java
index a04c549a4c..9063fdb1cd 100644
---
a/core/src/test/java/org/apache/gravitino/catalog/TestSchemaOperationDispatcher.java
+++
b/core/src/test/java/org/apache/gravitino/catalog/TestSchemaOperationDispatcher.java
@@ -51,6 +51,7 @@ import org.apache.gravitino.auth.AuthConstants;
import org.apache.gravitino.connector.HiddenPropertyMaskUtils;
import org.apache.gravitino.connector.TestCatalogOperations;
import org.apache.gravitino.exceptions.NoSuchEntityException;
+import org.apache.gravitino.exceptions.OptimisticLockException;
import org.apache.gravitino.exceptions.SchemaAlreadyExistsException;
import org.apache.gravitino.lock.LockManager;
import org.apache.gravitino.meta.AuditInfo;
@@ -381,6 +382,45 @@ public class TestSchemaOperationDispatcher extends
TestOperationDispatcher {
RuntimeException.class, () -> dispatcher.dropSchema(schemaIdent,
false));
}
+ @Test
+ public void testImportSchemaPropagatesOptimisticLockConflict() throws
IOException {
+ NameIdentifier schemaIdent = NameIdentifier.of(metalake, catalog,
"schema_import_occ");
+ dispatcher.createSchema(schemaIdent, "comment", ImmutableMap.of("k1",
"v1", "k2", "v2"));
+ entityStore.delete(schemaIdent, SCHEMA);
+
+ // The schema is only in the catalog, so loading it imports it into the
store.
+ doThrow(new NoSuchEntityException("mock error"))
+ .when(entityStore)
+ .get(any(), eq(Entity.EntityType.SCHEMA), any());
+ doThrow(new OptimisticLockException("mock conflict"))
+ .when(entityStore)
+ .put(any(SchemaEntity.class), anyBoolean());
+ try {
+ Assertions.assertThrows(
+ OptimisticLockException.class, () ->
dispatcher.loadSchema(schemaIdent));
+ } finally {
+ reset(entityStore);
+ }
+ }
+
+ @Test
+ public void testDropSchemaPropagatesOptimisticLockConflict() throws
IOException {
+ NameIdentifier schemaIdent = NameIdentifier.of(metalake, catalog,
"schema_drop_occ");
+ dispatcher.createSchema(schemaIdent, "comment", ImmutableMap.of("k1",
"v1", "k2", "v2"));
+
+ // A conflict of the store, such as a lock wait timeout, reaches the
caller as a conflict
+ // rather than an internal error.
+ doThrow(new OptimisticLockException("mock conflict"))
+ .when(entityStore)
+ .delete(any(), any(), anyBoolean());
+ try {
+ Assertions.assertThrows(
+ OptimisticLockException.class, () ->
dispatcher.dropSchema(schemaIdent, false));
+ } finally {
+ reset(entityStore);
+ }
+ }
+
@Test
public void testDropMissingSchemaPreservesStoredEntity() throws IOException {
reset(entityStore);
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 af0464e5a3..7da2d30644 100644
---
a/core/src/test/java/org/apache/gravitino/catalog/TestTableOperationDispatcher.java
+++
b/core/src/test/java/org/apache/gravitino/catalog/TestTableOperationDispatcher.java
@@ -877,6 +877,37 @@ public class TestTableOperationDispatcher extends
TestOperationDispatcher {
OptimisticLockException.class, () ->
tableOperationDispatcher.dropTable(tableIdent));
}
+ @Test
+ public void testImportTablePropagatesOptimisticLockConflict() throws
IOException {
+ NameIdentifier tableIdent =
+ NameIdentifier.of(metalake, catalog, "schema_import_occ",
"table_import_occ");
+ Map<String, String> props = ImmutableMap.of("k1", "v1", "k2", "v2");
+ Column[] columns =
+ new Column[] {
+ TestColumn.builder()
+ .withName("col1")
+ .withPosition(0)
+ .withType(Types.StringType.get())
+ .build()
+ };
+ schemaOperationDispatcher.createSchema(
+ NameIdentifier.of(tableIdent.namespace().levels()), "comment", props);
+ tableOperationDispatcher.createTable(tableIdent, columns, "comment",
props, new Transform[0]);
+ entityStore.delete(tableIdent, TABLE);
+
+ reset(entityStore);
+ doThrow(new NoSuchEntityException("mock
error")).when(entityStore).get(any(), eq(TABLE), any());
+ doThrow(new OptimisticLockException("mock conflict"))
+ .when(entityStore)
+ .put(any(TableEntity.class), anyBoolean());
+ try {
+ Assertions.assertThrows(
+ OptimisticLockException.class, () ->
tableOperationDispatcher.loadTable(tableIdent));
+ } finally {
+ reset(entityStore);
+ }
+ }
+
@Test
public void testPurgeTablePropagatesOptimisticLockConflict() throws
IOException {
NameIdentifier tableIdent =
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 23e06742b2..2078bd9b42 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.HiddenPropertyMaskUtils;
import org.apache.gravitino.connector.TestCatalogOperations;
import org.apache.gravitino.exceptions.NoSuchEntityException;
+import org.apache.gravitino.exceptions.OptimisticLockException;
import org.apache.gravitino.lock.LockManager;
import org.apache.gravitino.lock.LockType;
import org.apache.gravitino.messaging.Topic;
@@ -261,6 +262,48 @@ public class TestTopicOperationDispatcher extends
TestOperationDispatcher {
RuntimeException.class, () ->
topicOperationDispatcher.dropTopic(topicIdent));
}
+ @Test
+ public void testImportTopicPropagatesOptimisticLockConflict() throws
IOException {
+ Namespace topicNs = Namespace.of(metalake, catalog,
"schema_import_topic_occ");
+ Map<String, String> props = ImmutableMap.of("k1", "v1", "k2", "v2");
+
schemaOperationDispatcher.createSchema(NameIdentifier.of(topicNs.levels()),
"comment", props);
+ NameIdentifier topicIdent = NameIdentifier.of(topicNs, "topic_import_occ");
+ topicOperationDispatcher.createTopic(topicIdent, "comment", null, props);
+ entityStore.delete(topicIdent, Entity.EntityType.TOPIC);
+
+ doThrow(new NoSuchEntityException(""))
+ .when(entityStore)
+ .get(any(), eq(Entity.EntityType.TOPIC), any());
+ doThrow(new OptimisticLockException("mock conflict"))
+ .when(entityStore)
+ .put(any(TopicEntity.class), anyBoolean());
+ try {
+ Assertions.assertThrows(
+ OptimisticLockException.class, () ->
topicOperationDispatcher.loadTopic(topicIdent));
+ } finally {
+ reset(entityStore);
+ }
+ }
+
+ @Test
+ public void testDropTopicPropagatesOptimisticLockConflict() throws
IOException {
+ Namespace topicNs = Namespace.of(metalake, catalog,
"schema_drop_topic_occ");
+ Map<String, String> props = ImmutableMap.of("k1", "v1", "k2", "v2");
+
schemaOperationDispatcher.createSchema(NameIdentifier.of(topicNs.levels()),
"comment", props);
+ NameIdentifier topicIdent = NameIdentifier.of(topicNs, "topic_occ");
+ topicOperationDispatcher.createTopic(topicIdent, "comment", null, props);
+
+ doThrow(new OptimisticLockException("mock conflict"))
+ .when(entityStore)
+ .delete(any(), any(), anyBoolean());
+ try {
+ Assertions.assertThrows(
+ OptimisticLockException.class, () ->
topicOperationDispatcher.dropTopic(topicIdent));
+ } finally {
+ reset(entityStore);
+ }
+ }
+
@Test
public void testCreateTopicNeedImportingSchema() throws IOException {
Namespace topicNs = Namespace.of(metalake, catalog, "schema161");
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 f7bc7fe70e..b23ddad349 100644
---
a/core/src/test/java/org/apache/gravitino/catalog/TestViewOperationDispatcher.java
+++
b/core/src/test/java/org/apache/gravitino/catalog/TestViewOperationDispatcher.java
@@ -26,6 +26,7 @@ import static org.apache.gravitino.Entity.EntityType.VIEW;
import static org.apache.gravitino.StringIdentifier.ID_KEY;
import static org.apache.gravitino.TestBasePropertiesMetadata.COMMENT_KEY;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.doThrow;
@@ -57,6 +58,7 @@ import org.apache.gravitino.connector.TestCatalogOperations;
import org.apache.gravitino.exceptions.GravitinoRuntimeException;
import org.apache.gravitino.exceptions.NoSuchEntityException;
import org.apache.gravitino.exceptions.NoSuchViewException;
+import org.apache.gravitino.exceptions.OptimisticLockException;
import org.apache.gravitino.lock.LockManager;
import org.apache.gravitino.lock.LockType;
import org.apache.gravitino.meta.AuditInfo;
@@ -500,6 +502,54 @@ public class TestViewOperationDispatcher extends
TestOperationDispatcher {
Assertions.assertFalse(created.properties().containsKey(ID_KEY));
}
+ @Test
+ public void testImportViewPropagatesOptimisticLockConflict() throws
IOException {
+ Namespace viewNs = Namespace.of(metalake, catalog,
"schema_import_view_occ");
+ schemaOperationDispatcher.createSchema(
+ NameIdentifier.of(viewNs.levels()), "comment", ImmutableMap.of("k1",
"v1", "k2", "v2"));
+ NameIdentifier viewIdent = NameIdentifier.of(viewNs, "view_import_occ");
+ Representation[] representations = {
+ SQLRepresentation.builder().withDialect("spark").withSql("SELECT
1").build()
+ };
+ viewOperationDispatcher.createView(
+ viewIdent, null, new Column[0], representations, null, null,
ImmutableMap.of("k1", "v1"));
+ entityStore.delete(viewIdent, VIEW);
+
+ doThrow(new NoSuchEntityException("mock
error")).when(entityStore).get(any(), eq(VIEW), any());
+ doThrow(new OptimisticLockException("mock conflict"))
+ .when(entityStore)
+ .put(any(ViewEntity.class), anyBoolean());
+ try {
+ Assertions.assertThrows(
+ OptimisticLockException.class, () ->
viewOperationDispatcher.loadView(viewIdent));
+ } finally {
+ reset(entityStore);
+ }
+ }
+
+ @Test
+ public void testDropViewPropagatesOptimisticLockConflict() throws
IOException {
+ Namespace viewNs = Namespace.of(metalake, catalog, "schema_drop_view_occ");
+ schemaOperationDispatcher.createSchema(
+ NameIdentifier.of(viewNs.levels()), "comment", ImmutableMap.of("k1",
"v1", "k2", "v2"));
+ NameIdentifier viewIdent = NameIdentifier.of(viewNs, "view_occ");
+ Representation[] representations = {
+ SQLRepresentation.builder().withDialect("spark").withSql("SELECT
1").build()
+ };
+ viewOperationDispatcher.createView(
+ viewIdent, null, new Column[0], representations, null, null,
ImmutableMap.of("k1", "v1"));
+
+ doThrow(new OptimisticLockException("mock conflict"))
+ .when(entityStore)
+ .delete(any(), eq(VIEW), anyBoolean());
+ try {
+ Assertions.assertThrows(
+ OptimisticLockException.class, () ->
viewOperationDispatcher.dropView(viewIdent));
+ } finally {
+ reset(entityStore);
+ }
+ }
+
@Test
public void testDropView() throws IOException {
Namespace viewNs = Namespace.of(metalake, catalog, "schema_drop_view");
diff --git a/core/src/test/java/org/apache/gravitino/job/TestJobManager.java
b/core/src/test/java/org/apache/gravitino/job/TestJobManager.java
index 1d769c81cc..694bff92d9 100644
--- a/core/src/test/java/org/apache/gravitino/job/TestJobManager.java
+++ b/core/src/test/java/org/apache/gravitino/job/TestJobManager.java
@@ -219,6 +219,41 @@ public class TestJobManager {
}
}
+ @Test
+ public void testRunJobCancelsExecutionWhenRegistrationConflicts() throws
Exception {
+ JobTemplateEntity template = newShellJobTemplateEntity("shell_job", "A
shell job template");
+ doReturn(template).when(jobManager).getJobTemplate(metalake,
template.name());
+ Mockito.reset(entityStore, jobExecutor);
+ when(jobExecutor.submitJob(any(), any())).thenReturn("submitted_conflict");
+ OptimisticLockException conflict = new OptimisticLockException("lock wait
timeout");
+ doThrow(conflict).when(entityStore).put(any(JobEntity.class), eq(false));
+
+ // Cancellation cannot guarantee that the execution has not already
performed work.
+ RuntimeException failure =
+ Assertions.assertThrows(
+ RuntimeException.class,
+ () -> jobManager.runJob(metalake, template.name(),
Collections.emptyMap()));
+ Assertions.assertFalse(failure instanceof OptimisticLockException);
+ Assertions.assertSame(conflict, failure.getCause());
+ Assertions.assertTrue(failure.getMessage().contains("submitted_conflict"));
+ Assertions.assertTrue(failure.getMessage().contains("Check the execution
before retrying"));
+ verify(jobExecutor, times(1)).cancelJob("submitted_conflict");
+
+ // A failed cancellation is kept on the cause for diagnosis and is not a
retryable conflict.
+ Mockito.reset(jobExecutor);
+ when(jobExecutor.submitJob(any(),
any())).thenReturn("submitted_conflict_2");
+ RuntimeException cancelFailure = new RuntimeException("executor
unavailable");
+ doThrow(cancelFailure).when(jobExecutor).cancelJob("submitted_conflict_2");
+ RuntimeException second =
+ Assertions.assertThrows(
+ RuntimeException.class,
+ () -> jobManager.runJob(metalake, template.name(),
Collections.emptyMap()));
+ Assertions.assertFalse(second instanceof OptimisticLockException);
+ Assertions.assertSame(conflict, second.getCause());
+
Assertions.assertTrue(second.getMessage().contains("submitted_conflict_2"));
+
Assertions.assertTrue(List.of(second.getCause().getSuppressed()).contains(cancelFailure));
+ }
+
@Test
public void testAlterJobTemplateDistinguishesMissingFromConflict() throws
IOException {
NoSuchEntityException missing = new NoSuchEntityException("Template was
deleted");
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/TestRelationalEntityStoreBackendLockWaitTimeout.java
b/core/src/test/java/org/apache/gravitino/storage/relational/TestRelationalEntityStoreBackendLockWaitTimeout.java
new file mode 100644
index 0000000000..6ee5afb5d0
--- /dev/null
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/TestRelationalEntityStoreBackendLockWaitTimeout.java
@@ -0,0 +1,206 @@
+/*
+ * 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.nio.file.Files;
+import java.nio.file.Path;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.List;
+import org.apache.commons.dbcp2.BasicDataSource;
+import org.apache.commons.io.FileUtils;
+import org.apache.commons.lang3.reflect.FieldUtils;
+import org.apache.gravitino.Entity;
+import org.apache.gravitino.Namespace;
+import org.apache.gravitino.cache.EntityCache;
+import org.apache.gravitino.exceptions.OptimisticLockException;
+import org.apache.gravitino.meta.SchemaEntity;
+import org.apache.gravitino.meta.TableEntity;
+import org.apache.gravitino.storage.RandomIdGenerator;
+import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper;
+import org.apache.gravitino.storage.relational.session.SqlSessions;
+import org.apache.ibatis.mapping.Environment;
+import org.apache.ibatis.session.Configuration;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.function.Executable;
+import org.mockito.Mockito;
+
+/**
+ * Verifies real backend lock timeouts, transaction cleanup, and retries
through the entity store.
+ */
+@Tag("gravitino-docker-test")
+public class TestRelationalEntityStoreBackendLockWaitTimeout extends
TestJDBCBackend {
+
+ @AfterEach
+ void closeBackend() throws Exception {
+ String databasePath = null;
+ try {
+ if ("h2".equals(backendType)) {
+ BasicDataSource source =
+ (BasicDataSource)
+ SqlSessionFactoryHelper.getInstance()
+ .getSqlSessionFactory()
+ .getConfiguration()
+ .getEnvironment()
+ .getDataSource();
+ databasePath =
source.getUrl().substring("jdbc:h2:file:".length()).split(";")[0];
+ // The shared fixture uses DB_CLOSE_DELAY=-1, so close the engine
before deleting its files.
+ try (Connection connection = source.getConnection();
+ Statement statement = connection.createStatement()) {
+ statement.execute("SHUTDOWN");
+ }
+ }
+ } finally {
+ try {
+ backend.close();
+ } finally {
+ if (databasePath != null) {
+ FileUtils.deleteDirectory(Path.of(databasePath).toFile());
+ Files.deleteIfExists(Path.of(databasePath + ".mv.db"));
+ Files.deleteIfExists(Path.of(databasePath + ".trace.db"));
+ }
+ }
+ }
+ }
+
+ @TestTemplate
+ void testDropTimeoutPreservesTableAndRetrySucceeds() throws Exception {
+ createParents();
+ TableEntity table =
+ createAndInsertTableEntity(Namespace.of("timeout_metalake", "catalog",
"schema"), "table");
+ RelationalEntityStore store = createStore();
+
+ writeWhileRowHeld(
+ "table_meta",
+ "table_id",
+ table.id(),
+ () -> store.delete(table.nameIdentifier(), Entity.EntityType.TABLE));
+
+ Assertions.assertTrue(backend.exists(table.nameIdentifier(),
Entity.EntityType.TABLE));
+ Assertions.assertTrue(store.delete(table.nameIdentifier(),
Entity.EntityType.TABLE));
+ Assertions.assertFalse(backend.exists(table.nameIdentifier(),
Entity.EntityType.TABLE));
+ }
+
+ @TestTemplate
+ void testCreateTimeoutLeavesNoTableAndRetrySucceeds() throws Exception {
+ SchemaEntity schema = createParents();
+ TableEntity table =
+ createTableEntity(
+ RandomIdGenerator.INSTANCE.nextId(),
+ Namespace.of("timeout_metalake", "catalog", "schema"),
+ "table",
+ AUDIT_INFO);
+ RelationalEntityStore store = createStore();
+
+ writeWhileRowHeld("schema_meta", "schema_id", schema.id(), () ->
store.put(table, false));
+
+ Assertions.assertFalse(backend.exists(table.nameIdentifier(),
Entity.EntityType.TABLE));
+ store.put(table, false);
+ Assertions.assertEquals(
+ table.id(), backend.get(table.nameIdentifier(),
Entity.EntityType.TABLE).id());
+ }
+
+ private SchemaEntity createParents() throws Exception {
+ createAndInsertMakeLake("timeout_metalake");
+ createAndInsertCatalog("timeout_metalake", "catalog");
+ return createAndInsertSchema("timeout_metalake", "catalog", "schema");
+ }
+
+ private RelationalEntityStore createStore() throws IllegalAccessException {
+ RelationalEntityStore store = new RelationalEntityStore();
+ // Use the extension's real backend; only the unrelated cache is mocked.
+ FieldUtils.writeField(store, "backend", backend, true);
+ FieldUtils.writeField(store, "cache", Mockito.mock(EntityCache.class),
true);
+ return store;
+ }
+
+ @SuppressWarnings("deprecation")
+ private void writeWhileRowHeld(String table, String idColumn, long id,
Executable write)
+ throws Exception {
+ Configuration configuration =
+
SqlSessionFactoryHelper.getInstance().getSqlSessionFactory().getConfiguration();
+ Environment original = configuration.getEnvironment();
+ BasicDataSource source = (BasicDataSource) original.getDataSource();
+ String timeoutSql;
+ switch (backendType) {
+ case "h2":
+ timeoutSql = "SET LOCK_TIMEOUT 250";
+ break;
+ case "mysql":
+ timeoutSql = "SET SESSION innodb_lock_wait_timeout = 1";
+ break;
+ case "postgresql":
+ timeoutSql = "SET lock_timeout = '250ms'";
+ break;
+ default:
+ throw new IllegalStateException("Unsupported backend: " + backendType);
+ }
+
+ // Configure a separate writer pool before it opens connections. The
production write still
+ // owns its ordinary transaction, so a test-managed rollback cannot hide
missing cleanup.
+ try (BasicDataSource writers = new BasicDataSource();
+ Connection holder = source.getConnection()) {
+ writers.setUrl(source.getUrl());
+ writers.setDriverClassName(source.getDriverClassName());
+ writers.setUsername(source.getUsername());
+ writers.setPassword(source.getPassword());
+ writers.setDefaultAutoCommit(false);
+ writers.setConnectionInitSqls(List.of(timeoutSql));
+ configuration.setEnvironment(
+ new Environment(original.getId(), original.getTransactionFactory(),
writers));
+ holder.setAutoCommit(false);
+ try {
+ try (Statement statement = holder.createStatement();
+ ResultSet rows =
+ statement.executeQuery(
+ String.format(
+ "SELECT %s FROM %s WHERE %s = %d FOR UPDATE",
+ idColumn, table, idColumn, id))) {
+ Assertions.assertTrue(rows.next());
+ }
+ OptimisticLockException conflict =
+ Assertions.assertThrows(OptimisticLockException.class, write);
+ Assertions.assertTrue(conflict.getMessage().contains("retry the
operation"));
+ Throwable cause = conflict;
+ while (cause != null && !(cause instanceof SQLException)) {
+ cause = cause.getCause();
+ }
+ Assertions.assertNotNull(cause, "the real driver timeout must remain
attached");
+ SQLException timeout = (SQLException) cause;
+ if ("postgresql".equals(backendType)) {
+ Assertions.assertEquals("55P03", timeout.getSQLState());
+ } else {
+ Assertions.assertEquals(
+ "mysql".equals(backendType) ? 1205 : 50200,
timeout.getErrorCode());
+ }
+ Assertions.assertFalse(
+ SqlSessions.hasActiveSession(), "the failed writer must close its
session");
+ } finally {
+ holder.rollback();
+ }
+ } finally {
+ configuration.setEnvironment(original);
+ }
+ }
+}
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/TestRelationalEntityStoreLockWaitTimeout.java
b/core/src/test/java/org/apache/gravitino/storage/relational/TestRelationalEntityStoreLockWaitTimeout.java
new file mode 100644
index 0000000000..2c745ea60e
--- /dev/null
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/TestRelationalEntityStoreLockWaitTimeout.java
@@ -0,0 +1,518 @@
+/*
+ * 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.BatchUpdateException;
+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) {
+ try {
+ store.close();
+ } finally {
+ FieldUtils.writeField(GravitinoEnv.getInstance(), "config",
previousConfig, true);
+ FieldUtils.writeField(GravitinoEnv.getInstance(), "idGenerator",
previousIdGenerator, true);
+ FileUtils.deleteDirectory(new File(dbPath));
+ dbPath = null;
+ }
+ }
+ 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());
+ }
+ }
+ }
+
+ /** Mixed duplicate-key/timeout chains remain unclassified on direct store
write paths. */
+ @Test
+ void testMixedDuplicateKeyAndTimeoutPreservesBackendFailure() throws
Exception {
+ for (boolean duplicateFirst : List.of(true, false)) {
+ BatchUpdateException batch = new BatchUpdateException("batch failed",
new int[0]);
+ SQLException duplicate = new SQLException("duplicate key", "23505");
+ SQLException timeout = new SQLException("lock timeout", "55P03");
+ batch.setNextException(duplicateFirst ? duplicate : timeout);
+ batch.setNextException(duplicateFirst ? timeout : duplicate);
+ PersistenceException failure = new PersistenceException(batch);
+ initFailingStore(failure);
+ for (Pair<String, Executable> write : writes()) {
+ Assertions.assertSame(
+ failure,
+ Assertions.assertThrows(PersistenceException.class,
write.getRight()),
+ 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;MODE=MYSQL", new
File(dbPath, "db").getPath());
+
+ Config config = Mockito.mock(Config.class);
+
Mockito.when(config.get(Configs.ENTITY_STORE)).thenReturn(Configs.RELATIONAL_ENTITY_STORE);
+ Mockito.when(config.get(Configs.ENTITY_RELATIONAL_STORE))
+ .thenReturn(Configs.DEFAULT_ENTITY_RELATIONAL_STORE);
+
Mockito.when(config.get(Configs.ENTITY_RELATIONAL_JDBC_BACKEND_URL)).thenReturn(jdbcUrl);
+
Mockito.when(config.get(Configs.ENTITY_RELATIONAL_JDBC_BACKEND_USER)).thenReturn(USER);
+
Mockito.when(config.get(Configs.ENTITY_RELATIONAL_JDBC_BACKEND_PASSWORD)).thenReturn(PASSWORD);
+ Mockito.when(config.get(Configs.ENTITY_RELATIONAL_JDBC_BACKEND_DRIVER))
+ .thenReturn("org.h2.Driver");
+
Mockito.when(config.get(Configs.ENTITY_RELATIONAL_JDBC_BACKEND_MAX_CONNECTIONS))
+ .thenReturn(Configs.DEFAULT_RELATIONAL_JDBC_BACKEND_MAX_CONNECTIONS);
+
Mockito.when(config.get(Configs.ENTITY_RELATIONAL_JDBC_BACKEND_MAX_IDLE_CONNECTIONS))
+ .thenReturn(10);
+
Mockito.when(config.get(Configs.ENTITY_RELATIONAL_JDBC_BACKEND_WAIT_MILLISECONDS))
+
.thenReturn(Configs.DEFAULT_RELATIONAL_JDBC_BACKEND_MAX_WAIT_MILLISECONDS);
+ Mockito.when(config.get(Configs.STORE_DELETE_AFTER_TIME)).thenReturn(20 *
60 * 1000L);
+ Mockito.when(config.get(Configs.VERSION_RETENTION_COUNT)).thenReturn(1L);
+
Mockito.when(config.get(Configs.ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS)).thenReturn(3L);
+ Mockito.when(config.get(Configs.ENTITY_CHANGE_LOG_POLL_BATCH_SIZE))
+ .thenReturn(Configs.DEFAULT_ENTITY_CHANGE_LOG_POLL_BATCH_SIZE);
+
Mockito.when(config.get(Configs.ENTITY_CHANGE_LOG_RETENTION_SECS)).thenReturn(24
* 60 * 60L);
+
Mockito.when(config.get(Configs.ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS)).thenReturn(60
* 60L);
+ Mockito.when(config.get(Configs.CACHE_ENABLED)).thenReturn(false);
+ Mockito.when(config.get(Configs.SCHEMA_SEPARATOR)).thenReturn(":");
+
+ previousConfig = FieldUtils.readField(GravitinoEnv.getInstance(),
"config", true);
+ previousIdGenerator = FieldUtils.readField(GravitinoEnv.getInstance(),
"idGenerator", true);
+ FieldUtils.writeField(GravitinoEnv.getInstance(), "config", config, true);
+ FieldUtils.writeField(
+ GravitinoEnv.getInstance(), "idGenerator", RandomIdGenerator.INSTANCE,
true);
+
+ store = new RelationalEntityStore();
+ store.initialize(config);
+ return jdbcUrl;
+ }
+
+ private static SQLException findSqlException(Throwable failure) {
+ for (Throwable cause = failure; cause != null; cause = cause.getCause()) {
+ if (cause instanceof SQLException) {
+ return (SQLException) cause;
+ }
+ }
+ return null;
+ }
+
+ private static BaseMetalake metalake() {
+ return BaseMetalake.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName(METALAKE)
+ .withAuditInfo(AUDIT_INFO)
+ .withComment("")
+ .withProperties(null)
+ .withVersion(SchemaVersion.V_0_1)
+ .build();
+ }
+
+ private static CatalogEntity catalog() {
+ return CatalogEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName(CATALOG)
+ .withNamespace(Namespace.of(METALAKE))
+ .withType(Catalog.Type.RELATIONAL)
+ .withProvider("test")
+ .withComment("")
+ .withProperties(null)
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ }
+
+ private static SchemaEntity schema() {
+ return SchemaEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName(SCHEMA)
+ .withNamespace(Namespace.of(METALAKE, CATALOG))
+ .withComment("")
+ .withProperties(null)
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ }
+
+ private static TableEntity table() {
+ return TableEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName(TABLE_IDENT.name())
+ .withNamespace(TABLE_IDENT.namespace())
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ }
+}
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestStatisticMetaService.java
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestStatisticMetaService.java
index 014e4c7319..d113eefb58 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestStatisticMetaService.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestStatisticMetaService.java
@@ -436,24 +436,36 @@ public class TestStatisticMetaService extends
TestJDBCBackend {
// backend
// collation are rejected before the write, so a duplicate key can only
come from a concurrent
// creator, even if that creator's row is gone again by the time the
caller looks.
- SQLException duplicate = new SQLException("duplicate key", "23505");
- StatisticMetaService duplicating = failingStatements(new
PersistenceException(duplicate));
- OptimisticLockException writeDuplicate =
- Assertions.assertThrows(
- OptimisticLockException.class,
- () ->
- duplicating.writeStatisticsWithVersion(
- List.of(createNamedStatistic("b", 2L)),
- table.nameIdentifier(),
- Entity.EntityType.TABLE));
- Assertions.assertSame(duplicate, writeDuplicate.getCause().getCause());
- OptimisticLockException dropDuplicate =
- Assertions.assertThrows(
- OptimisticLockException.class,
- () ->
- duplicating.batchDeleteStatisticPOs(
- table.nameIdentifier(), Entity.EntityType.TABLE,
List.of("a")));
- Assertions.assertSame(duplicate, dropDuplicate.getCause().getCause());
+ for (SQLException duplicate :
+ List.of(
+ new SQLException("MySQL duplicate", null, 1062),
+ new SQLException("H2 duplicate", "23505", 23505),
+ new SQLException("PostgreSQL duplicate", "23505"))) {
+ SQLException mixed = new SQLException("batch failed");
+ mixed.setNextException(new SQLException("lock timeout", "55P03"));
+ mixed.setNextException(duplicate);
+ for (SQLException failure : List.of(duplicate, mixed)) {
+ StatisticMetaService duplicating = failingStatements(new
PersistenceException(failure));
+ OptimisticLockException writeDuplicate =
+ Assertions.assertThrows(
+ OptimisticLockException.class,
+ () ->
+ duplicating.writeStatisticsWithVersion(
+ List.of(createNamedStatistic("b", 2L)),
+ table.nameIdentifier(),
+ Entity.EntityType.TABLE));
+ Assertions.assertSame(failure, writeDuplicate.getCause().getCause());
+ OptimisticLockException dropDuplicate =
+ Assertions.assertThrows(
+ OptimisticLockException.class,
+ () ->
+ duplicating.batchDeleteStatisticPOs(
+ table.nameIdentifier(), Entity.EntityType.TABLE,
List.of("a")));
+ Assertions.assertSame(failure, dropDuplicate.getCause().getCause());
+ Assertions.assertTrue(writeDuplicate.getMessage().contains("were
changed concurrently"));
+ Assertions.assertTrue(dropDuplicate.getMessage().contains("were
changed concurrently"));
+ }
+ }
PersistenceException connectionFailure =
new PersistenceException(new SQLException("connection lost", "08006"));
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/utils/TestExceptionUtils.java
b/core/src/test/java/org/apache/gravitino/storage/relational/utils/TestExceptionUtils.java
index 0d489be688..72275aea97 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/utils/TestExceptionUtils.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/utils/TestExceptionUtils.java
@@ -19,10 +19,23 @@
package org.apache.gravitino.storage.relational.utils;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import java.sql.BatchUpdateException;
+import java.sql.SQLException;
+import java.time.Duration;
+import java.util.List;
+import java.util.Set;
import org.apache.gravitino.Entity;
+import org.apache.gravitino.EntityAlreadyExistsException;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.exceptions.OptimisticLockException;
+import org.apache.ibatis.exceptions.PersistenceException;
import org.junit.jupiter.api.Test;
public class TestExceptionUtils {
@@ -47,4 +60,195 @@ public class TestExceptionUtils {
"A schema under catalog m1.c1 was modified concurrently; retry the
operation",
e.getMessage());
}
+
+ @Test
+ public void testIsLockWaitTimeout() {
+ for (SQLException timeout :
+ List.of(
+ // The MySQL driver reports 1205 as SQLState 40001 or HY000
depending on its version.
+ 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"))) {
+ assertTrue(ExceptionUtils.isLockWaitTimeout(timeout),
timeout.getMessage());
+ // MyBatis wraps the driver exception, possibly more than once.
+ assertTrue(
+ ExceptionUtils.isLockWaitTimeout(new RuntimeException(new
PersistenceException(timeout))),
+ timeout.getMessage());
+ }
+
+ for (Throwable other :
+ List.of(
+ new SQLException("MySQL deadlock", "40001", 1213),
+ new SQLException("H2 deadlock", "40001", 40001),
+ new SQLException("PostgreSQL deadlock", "40P01"),
+ new SQLException("serialization failure", "40001"),
+ new SQLException("H2 statement timeout", "57014", 57014),
+ new SQLException("connection lost", "08006"),
+ new PersistenceException("no SQL cause"))) {
+ assertFalse(ExceptionUtils.isLockWaitTimeout(other), other.getMessage());
+ }
+ }
+
+ @Test
+ public void testLockWaitTimeoutAsConflict() {
+ PersistenceException timeout =
+ new PersistenceException(new SQLException("H2 lock wait timeout",
"HYT00", 50200));
+ RuntimeException conflict =
+ ExceptionUtils.lockWaitTimeoutAsConflict(
+ timeout, Entity.EntityType.TABLE, NameIdentifier.of("m1", "c1",
"s1", "t1"));
+ assertInstanceOf(OptimisticLockException.class, conflict);
+ assertSame(timeout, conflict.getCause());
+ assertEquals(
+ "The table m1.c1.s1.t1 could not be changed because a concurrent
operation held its lock"
+ + " longer than the backend's lock wait timeout; retry the
operation",
+ conflict.getMessage());
+
+ assertEquals(
+ "The batch of 2 entities could not be changed because a concurrent
operation held its"
+ + " lock longer than the backend's lock wait timeout; retry the
operation",
+ ExceptionUtils.lockWaitTimeoutAsConflict(timeout, "The batch of 2
entities").getMessage());
+
+ // A conflict is never wrapped again, and any other failure is returned as
is.
+ OptimisticLockException existing = new OptimisticLockException(timeout,
"already a conflict");
+ assertSame(existing, ExceptionUtils.lockWaitTimeoutAsConflict(existing,
"The table t1"));
+ PersistenceException deadlock =
+ new PersistenceException(new SQLException("MySQL deadlock", "40001",
1213));
+ assertSame(deadlock, ExceptionUtils.lockWaitTimeoutAsConflict(deadlock,
"The table t1"));
+ }
+
+ @Test
+ public void testCheckSQLExceptionReportsLockWaitTimeoutAsConflict() {
+ // Services call checkSQLException before the store sees the failure, so
the timeout must be
+ // classified here; the backend converter would turn it into an
IOException.
+ PersistenceException timeout =
+ new PersistenceException(new SQLException("MySQL lock wait timeout",
"40001", 1205));
+ OptimisticLockException conflict =
+ assertThrows(
+ OptimisticLockException.class,
+ () ->
+ ExceptionUtils.checkSQLException(timeout,
Entity.EntityType.TABLE, "m1.c1.s1.t1"));
+ assertSame(timeout, conflict.getCause());
+ assertTrue(
+ conflict.getMessage().startsWith("The table m1.c1.s1.t1 could not be
changed"),
+ conflict.getMessage());
+ }
+
+ @Test
+ void testLockWaitTimeoutInBatchExceptionChain() {
+ BatchUpdateException batch = new BatchUpdateException("batch failed", new
int[0]);
+ batch.setNextException(new SQLException("statement failed"));
+ batch.setNextException(new SQLException("MySQL lock wait timeout", null,
1205));
+ PersistenceException failure = new PersistenceException(batch);
+
+ assertTrue(ExceptionUtils.isLockWaitTimeout(failure));
+ OptimisticLockException conflict =
+ assertThrows(
+ OptimisticLockException.class,
+ () ->
+ ExceptionUtils.checkSQLException(failure,
Entity.EntityType.TABLE, "m1.c1.s1.t1"));
+ assertSame(failure, conflict.getCause());
+ assertTrue(conflict.getMessage().contains("m1.c1.s1.t1"));
+ }
+
+ @Test
+ void testSharedDuplicateKeyClassifier() {
+ for (SQLException duplicate :
+ List.of(
+ new SQLException("MySQL duplicate", null, 1062),
+ new SQLException("H2 duplicate", "23505", 23505),
+ new SQLException("PostgreSQL duplicate", "23505"))) {
+ assertTrue(ExceptionUtils.isDuplicateKey(duplicate));
+ BatchUpdateException batch = new BatchUpdateException("batch failed",
new int[0]);
+ batch.setNextException(new SQLException("lock timeout", "55P03"));
+ batch.setNextException(duplicate);
+ assertTrue(ExceptionUtils.isDuplicateKey(new
PersistenceException(batch)));
+ }
+ assertFalse(ExceptionUtils.isDuplicateKey(new SQLException("foreign key",
"23503")));
+ assertFalse(ExceptionUtils.isDuplicateKey(new SQLException("lock timeout",
"HY000", 1205)));
+ assertFalse(ExceptionUtils.isDuplicateKey(new IllegalStateException("not a
SQL failure")));
+ assertFalse(ExceptionUtils.isDuplicateKey(null));
+ }
+
+ @Test
+ void testDuplicateKeyTakesPrecedenceOverChainedLockWaitTimeout() {
+ for (int backend = 0; backend < 3; backend++) {
+ for (boolean duplicateFirst : List.of(true, false)) {
+ SQLException duplicate =
+ new SQLException(
+ "duplicate key", backend == 0 ? "23000" : "23505", backend ==
0 ? 1062 : 23505);
+ SQLException timeout =
+ new SQLException(
+ "lock wait timeout",
+ backend == 0 ? "40001" : backend == 1 ? "HYT00" : "55P03",
+ backend == 0 ? 1205 : backend == 1 ? 50200 : 0);
+ BatchUpdateException batch = new BatchUpdateException("batch failed",
new int[0]);
+ batch.setNextException(duplicateFirst ? duplicate : timeout);
+ batch.setNextException(duplicateFirst ? timeout : duplicate);
+ PersistenceException failure = new PersistenceException(batch);
+
+ assertTrue(ExceptionUtils.isLockWaitTimeout(failure));
+ // Custom association messages must not classify the mixed chain
before checkSQLException.
+ assertSame(failure, ExceptionUtils.lockWaitTimeoutAsConflict(failure,
"The association"));
+ ExceptionUtils.throwIfLockWaitTimeout(failure, "The association");
+ EntityAlreadyExistsException exists =
+ assertThrows(
+ EntityAlreadyExistsException.class,
+ () ->
+ ExceptionUtils.checkSQLException(
+ failure, Entity.EntityType.TABLE, "m1.c1.s1.t1"));
+ assertSame(failure, exists.getCause());
+ assertEquals("The TABLE entity: m1.c1.s1.t1 already exists.",
exists.getMessage());
+ }
+ }
+ }
+
+ @Test
+ void testCauseOfChainedSqlExceptionIsInspected() {
+ SQLException batch = new SQLException("batch failed");
+ SQLException chained =
+ new SQLException(
+ "statement failed", new RuntimeException(new SQLException("lock
timeout", "55P03")));
+ batch.setNextException(chained);
+
+ assertTrue(ExceptionUtils.isLockWaitTimeout(new
PersistenceException(batch)));
+ }
+
+ @Test
+ void testOtherSqlClassificationsInspectNextExceptions() {
+ SQLException batch = new SQLException("batch failed");
+ batch.setNextException(new SQLException("duplicate key", "23505"));
+ assertTrue(ExceptionUtils.hasSqlException(batch, Set.of("23505"),
Set.of(1062)));
+ assertFalse(ExceptionUtils.isLockWaitTimeout(batch));
+
+ SQLException deadlock = new SQLException("batch failed");
+ deadlock.setNextException(new SQLException("MySQL deadlock", "40001",
1213));
+ assertTrue(ExceptionUtils.hasSqlException(deadlock, Set.of("40P01"),
Set.of(1213)));
+ assertFalse(ExceptionUtils.isLockWaitTimeout(deadlock));
+ }
+
+ @Test
+ void testExceptionCyclesTerminateWithoutMatch() {
+ SQLException first = new SQLException("first");
+ SQLException second = new SQLException("second");
+ first.setNextException(second);
+ second.setNextException(first);
+ first.initCause(new RuntimeException(second));
+
+ assertTimeoutPreemptively(
+ Duration.ofSeconds(5), () ->
assertFalse(ExceptionUtils.isLockWaitTimeout(first)));
+ }
+
+ @Test
+ void testSharedCauseDoesNotHideChainedTimeout() {
+ SQLException first = new SQLException("first");
+ SQLException second = new SQLException("second");
+ first.initCause(second);
+ first.setNextException(second);
+ second.setNextException(new SQLException("H2 lock timeout", "HYT00",
50200));
+
+ assertTrue(ExceptionUtils.isLockWaitTimeout(first));
+ assertFalse(ExceptionUtils.isLockWaitTimeout(null));
+ assertFalse(ExceptionUtils.isLockWaitTimeout(new SQLException("no
SQLState")));
+ }
}
diff --git
a/server/src/test/java/org/apache/gravitino/server/web/rest/TestExceptionHandlers.java
b/server/src/test/java/org/apache/gravitino/server/web/rest/TestExceptionHandlers.java
index ff03f3462b..afbc556a9f 100644
---
a/server/src/test/java/org/apache/gravitino/server/web/rest/TestExceptionHandlers.java
+++
b/server/src/test/java/org/apache/gravitino/server/web/rest/TestExceptionHandlers.java
@@ -46,6 +46,23 @@ public class TestExceptionHandlers {
}
}
+ @Test
+ void testSubmittedJobRegistrationFailureDoesNotReturnRetryableConflict() {
+ RuntimeException failure =
+ new RuntimeException(
+ "Execution submitted_conflict may already have run; check it
before retrying",
+ new OptimisticLockException("lock wait timeout"));
+ try (Response response =
+ ExceptionHandlers.handleJobException(OperationType.RUN, "job",
"metalake", failure)) {
+ Assertions.assertEquals(
+ Response.Status.INTERNAL_SERVER_ERROR.getStatusCode(),
response.getStatus());
+ ErrorResponse error = (ErrorResponse) response.getEntity();
+ Assertions.assertEquals(RuntimeException.class.getSimpleName(),
error.getType());
+ Assertions.assertTrue(error.getMessage().contains("submitted_conflict"));
+ Assertions.assertTrue(error.getMessage().contains("check it before
retrying"));
+ }
+ }
+
@Test
public void testMissingJobParentsReturnNotFound() {
Response runResponse =
