This is an automated email from the ASF dual-hosted git repository.
jerryshao 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 8f41d92f87 [#12651] improvement(core): add OCC for topic writes
(#12680)
8f41d92f87 is described below
commit 8f41d92f87485730fbe4d75971bf5088551d37e9
Author: Qi Yu <[email protected]>
AuthorDate: Mon Aug 31 20:31:51 2026 +0800
[#12651] improvement(core): add OCC for topic writes (#12680)
### What changes were proposed in this pull request?
- Use `topic_meta.current_version` as the OCC token for topic alters and
direct deletes.
- Run the root topic CAS and dependent metadata cleanup in one
transaction.
- Classify stale writes as either `OptimisticLockException` or
`NoSuchEntityException`.
- Lock the parent schema during topic creation.
- Preserve monotonic versions and persisted topic IDs during overwrite
across H2, MySQL, and PostgreSQL.
- Invalidate the entity cache after overwrite because the database may
preserve existing identity.
- Add SQL-provider, service, converter, and cache tests for the OCC
behavior.
### Why are the changes needed?
Concurrent topic writes could overwrite newer metadata, and a stale
delete could remove relationships belonging to a newer topic. Overwrite
could also reset the OCC version or cache an entity whose ID differed
from the row retained by the database.
Fix: #12651
### Does this PR introduce _any_ user-facing change?
Yes. Concurrent stale topic writes now fail with an optimistic-lock
conflict, while writes against a topic that was deleted or renamed
report that the topic no longer exists.
No public API or property key is changed.
### How was this patch tested?
- `./gradlew --no-daemon :core:spotlessApply`
- `SKIP_DOCKER_TESTS=true ./gradlew --no-daemon :core:check -PskipITs
-PskipDockerTests=true`
- Targeted Topic service, SQL-provider, PO converter, and relational
entity-store tests
---
.../storage/relational/mapper/TopicMetaMapper.java | 19 +-
.../mapper/TopicMetaSQLProviderFactory.java | 34 +-
.../provider/base/TopicMetaBaseSQLProvider.java | 62 +++-
.../postgresql/TopicMetaPostgreSQLProvider.java | 58 ++-
.../relational/service/TopicMetaService.java | 143 +++++---
.../storage/relational/utils/POConverters.java | 7 +-
.../relational/TestRelationalEntityStore.java | 26 ++
.../mapper/TestTopicMetaSQLProviderFactory.java | 39 +++
.../base/TestTopicMetaBaseSQLProvider.java | 69 ++++
.../TestTopicMetaPostgreSQLProvider.java | 67 ++++
.../relational/service/TestTopicMetaService.java | 388 ++++++++++++++++++++-
.../storage/relational/utils/TestPOConverters.java | 56 ++-
12 files changed, 856 insertions(+), 112 deletions(-)
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TopicMetaMapper.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TopicMetaMapper.java
index fd447015d8..23b781129d 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TopicMetaMapper.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TopicMetaMapper.java
@@ -69,6 +69,15 @@ public interface TopicMetaMapper {
@SelectProvider(type = TopicMetaSQLProviderFactory.class, method =
"selectTopicMetaById")
TopicPO selectTopicMetaById(@Param("topicId") Long topicId);
+ /**
+ * Selects and exclusively locks an active topic metadata row.
+ *
+ * @param topicId the topic ID
+ * @return the active topic metadata, or {@code null} when it no longer
exists
+ */
+ @SelectProvider(type = TopicMetaSQLProviderFactory.class, method =
"selectTopicMetaByIdForUpdate")
+ TopicPO selectTopicMetaByIdForUpdate(@Param("topicId") Long topicId);
+
@UpdateProvider(type = TopicMetaSQLProviderFactory.class, method =
"updateTopicMeta")
Integer updateTopicMeta(
@Param("newTopicMeta") TopicPO newTopicPO, @Param("oldTopicMeta")
TopicPO oldTopicPO);
@@ -79,10 +88,18 @@ public interface TopicMetaMapper {
Long selectTopicIdBySchemaIdAndName(
@Param("schemaId") Long schemaId, @Param("topicName") String name);
+ /**
+ * Soft-deletes a topic only if its version has not changed since the caller
read it.
+ *
+ * @param topicId the topic ID
+ * @param currentVersion the version observed by the caller
+ * @return the number of deleted rows; zero means the topic changed or
disappeared
+ */
@UpdateProvider(
type = TopicMetaSQLProviderFactory.class,
method = "softDeleteTopicMetasByTopicId")
- Integer softDeleteTopicMetasByTopicId(@Param("topicId") Long topicId);
+ Integer softDeleteTopicMetasByTopicId(
+ @Param("topicId") Long topicId, @Param("currentVersion") Long
currentVersion);
@UpdateProvider(
type = TopicMetaSQLProviderFactory.class,
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TopicMetaSQLProviderFactory.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TopicMetaSQLProviderFactory.java
index 32d6c1b52e..3e808b53f1 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TopicMetaSQLProviderFactory.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/TopicMetaSQLProviderFactory.java
@@ -50,7 +50,17 @@ public class TopicMetaSQLProviderFactory {
static class TopicMetaMySQLProvider extends TopicMetaBaseSQLProvider {}
- static class TopicMetaH2Provider extends TopicMetaBaseSQLProvider {}
+ static class TopicMetaH2Provider extends TopicMetaBaseSQLProvider {
+
+ /** {@inheritDoc} */
+ @Override
+ protected String overwriteVersionAssignments() {
+ // H2 evaluates both right-hand sides against the row before the update,
so each assignment
+ // must calculate the next version independently.
+ return " current_version = GREATEST(current_version, last_version) + 1,"
+ + " last_version = GREATEST(current_version, last_version) + 1,";
+ }
+ }
public static String insertTopicMeta(@Param("topicMeta") TopicPO topicPO) {
return getProvider().insertTopicMeta(topicPO);
@@ -93,6 +103,16 @@ public class TopicMetaSQLProviderFactory {
return getProvider().selectTopicMetaById(topicId);
}
+ /**
+ * Returns SQL that locks an active topic metadata row by ID.
+ *
+ * @param topicId the topic ID
+ * @return the locking select SQL
+ */
+ public static String selectTopicMetaByIdForUpdate(@Param("topicId") Long
topicId) {
+ return getProvider().selectTopicMetaByIdForUpdate(topicId);
+ }
+
public static String updateTopicMeta(
@Param("newTopicMeta") TopicPO newTopicPO, @Param("oldTopicMeta")
TopicPO oldTopicPO) {
return getProvider().updateTopicMeta(newTopicPO, oldTopicPO);
@@ -103,8 +123,16 @@ public class TopicMetaSQLProviderFactory {
return getProvider().selectTopicIdBySchemaIdAndName(schemaId, name);
}
- public static String softDeleteTopicMetasByTopicId(@Param("topicId") Long
topicId) {
- return getProvider().softDeleteTopicMetasByTopicId(topicId);
+ /**
+ * Returns SQL that soft-deletes a topic by ID and expected version.
+ *
+ * @param topicId the topic ID
+ * @param currentVersion the version observed by the caller
+ * @return the version-checked delete SQL
+ */
+ public static String softDeleteTopicMetasByTopicId(
+ @Param("topicId") Long topicId, @Param("currentVersion") Long
currentVersion) {
+ return getProvider().softDeleteTopicMetasByTopicId(topicId,
currentVersion);
}
public static String softDeleteTopicMetasByCatalogId(@Param("catalogId")
Long catalogId) {
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/TopicMetaBaseSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/TopicMetaBaseSQLProvider.java
index 8d508b351b..271695b87f 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/TopicMetaBaseSQLProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/TopicMetaBaseSQLProvider.java
@@ -78,8 +78,7 @@ public class TopicMetaBaseSQLProvider {
+ " comment = #{topicMeta.comment},"
+ " properties = #{topicMeta.properties},"
+ " audit_info = #{topicMeta.auditInfo},"
- + " current_version = #{topicMeta.currentVersion},"
- + " last_version = #{topicMeta.lastVersion},"
+ + overwriteVersionAssignments()
+ " deleted_at = #{topicMeta.deletedAt}";
}
@@ -219,6 +218,29 @@ public class TopicMetaBaseSQLProvider {
+ " WHERE topic_id = #{topicId} AND deleted_at = 0";
}
+ /**
+ * Returns an active topic metadata row and locks it for the current
transaction.
+ *
+ * <p>The stable ID lets a failed CAS distinguish a newer topic from one
that was deleted,
+ * renamed, or moved while the caller was writing.
+ *
+ * @param topicId the topic ID
+ * @return the locking select SQL
+ */
+ public String selectTopicMetaByIdForUpdate(@Param("topicId") Long topicId) {
+ return selectTopicMetaById(topicId) + " FOR UPDATE";
+ }
+
+ /**
+ * Returns SQL that updates a topic only while its OCC version is unchanged.
+ *
+ * <p>The version is the concurrency token. Comparing payload columns would
miss a writer that
+ * changes a value and then changes it back before this update runs.
+ *
+ * @param newTopicPO the new topic values
+ * @param oldTopicPO the topic values and version observed by the caller
+ * @return the version-checked update SQL
+ */
public String updateTopicMeta(
@Param("newTopicMeta") TopicPO newTopicPO, @Param("oldTopicMeta")
TopicPO oldTopicPO) {
return "UPDATE "
@@ -234,16 +256,7 @@ public class TopicMetaBaseSQLProvider {
+ " last_version = #{newTopicMeta.lastVersion},"
+ " deleted_at = #{newTopicMeta.deletedAt}"
+ " WHERE topic_id = #{oldTopicMeta.topicId}"
- + " AND topic_name = #{oldTopicMeta.topicName}"
- + " AND metalake_id = #{oldTopicMeta.metalakeId}"
- + " AND catalog_id = #{oldTopicMeta.catalogId}"
- + " AND schema_id = #{oldTopicMeta.schemaId}"
- + " AND (comment = #{oldTopicMeta.comment}"
- + " OR (comment IS NULL and #{oldTopicMeta.comment} IS NULL))"
- + " AND properties = #{oldTopicMeta.properties}"
- + " AND audit_info = #{oldTopicMeta.auditInfo}"
+ " AND current_version = #{oldTopicMeta.currentVersion}"
- + " AND last_version = #{oldTopicMeta.lastVersion}"
+ " AND deleted_at = 0";
}
@@ -255,12 +268,21 @@ public class TopicMetaBaseSQLProvider {
+ " AND deleted_at = 0";
}
- public String softDeleteTopicMetasByTopicId(@Param("topicId") Long topicId) {
+ /**
+ * Returns SQL that deletes only the topic version observed by the caller.
+ *
+ * @param topicId the topic ID
+ * @param currentVersion the version observed by the caller
+ * @return the version-checked delete SQL
+ */
+ public String softDeleteTopicMetasByTopicId(
+ @Param("topicId") Long topicId, @Param("currentVersion") Long
currentVersion) {
return "UPDATE "
+ TABLE_NAME
+ " SET deleted_at = (UNIX_TIMESTAMP() * 1000.0)"
+ " + EXTRACT(MICROSECOND FROM CURRENT_TIMESTAMP(3)) / 1000"
- + " WHERE topic_id = #{topicId} AND deleted_at = 0";
+ + " WHERE topic_id = #{topicId}"
+ + " AND current_version = #{currentVersion} AND deleted_at = 0";
}
public String softDeleteTopicMetasByCatalogId(@Param("catalogId") Long
catalogId) {
@@ -334,4 +356,18 @@ public class TopicMetaBaseSQLProvider {
+ " AND tm.deleted_at = 0 AND sm.deleted_at = 0 AND cm.deleted_at = 0
AND mm.deleted_at = 0"
+ "</script>";
}
+
+ /**
+ * Returns MySQL assignments that advance an overwritten topic beyond both
stored version markers.
+ *
+ * <p>MySQL evaluates assignments from left to right. Updating {@code
current_version} first lets
+ * {@code last_version} copy the same newly computed value without
evaluating the maximum again
+ * against a partially updated row.
+ *
+ * @return the overwrite version assignments
+ */
+ protected String overwriteVersionAssignments() {
+ return " current_version = GREATEST(current_version, last_version) + 1,"
+ + " last_version = current_version,";
+ }
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/TopicMetaPostgreSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/TopicMetaPostgreSQLProvider.java
index 711c951934..3014403c14 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/TopicMetaPostgreSQLProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/TopicMetaPostgreSQLProvider.java
@@ -26,43 +26,13 @@ import org.apache.gravitino.storage.relational.po.TopicPO;
import org.apache.ibatis.annotations.Param;
public class TopicMetaPostgreSQLProvider extends TopicMetaBaseSQLProvider {
-
- @Override
- public String updateTopicMeta(
- @Param("newTopicMeta") TopicPO newTopicPO, @Param("oldTopicMeta")
TopicPO oldTopicPO) {
- return "UPDATE "
- + TABLE_NAME
- + " SET topic_name = #{newTopicMeta.topicName},"
- + " metalake_id = #{newTopicMeta.metalakeId},"
- + " catalog_id = #{newTopicMeta.catalogId},"
- + " schema_id = #{newTopicMeta.schemaId},"
- + " comment = #{newTopicMeta.comment},"
- + " properties = #{newTopicMeta.properties},"
- + " audit_info = #{newTopicMeta.auditInfo},"
- + " current_version = #{newTopicMeta.currentVersion},"
- + " last_version = #{newTopicMeta.lastVersion},"
- + " deleted_at = #{newTopicMeta.deletedAt}"
- + " WHERE topic_id = #{oldTopicMeta.topicId}"
- + " AND topic_name = #{oldTopicMeta.topicName}"
- + " AND metalake_id = #{oldTopicMeta.metalakeId}"
- + " AND catalog_id = #{oldTopicMeta.catalogId}"
- + " AND schema_id = #{oldTopicMeta.schemaId}"
- + " AND (comment = #{oldTopicMeta.comment}"
- + " OR (CAST(comment AS VARCHAR) IS NULL"
- + " AND CAST(#{oldTopicMeta.comment} AS VARCHAR) IS NULL))"
- + " AND properties = #{oldTopicMeta.properties}"
- + " AND audit_info = #{oldTopicMeta.auditInfo}"
- + " AND current_version = #{oldTopicMeta.currentVersion}"
- + " AND last_version = #{oldTopicMeta.lastVersion}"
- + " AND deleted_at = 0";
- }
-
@Override
- public String softDeleteTopicMetasByTopicId(Long topicId) {
+ public String softDeleteTopicMetasByTopicId(Long topicId, Long
currentVersion) {
return "UPDATE "
+ TABLE_NAME
+ " SET deleted_at = CAST(EXTRACT(EPOCH FROM CURRENT_TIMESTAMP) * 1000
AS BIGINT)"
- + " WHERE topic_id = #{topicId} AND deleted_at = 0";
+ + " WHERE topic_id = #{topicId}"
+ + " AND current_version = #{currentVersion} AND deleted_at = 0";
}
@Override
@@ -115,7 +85,10 @@ public class TopicMetaPostgreSQLProvider extends
TopicMetaBaseSQLProvider {
+ " #{topicMeta.lastVersion},"
+ " #{topicMeta.deletedAt}"
+ " )"
- + " ON CONFLICT (topic_id) DO UPDATE SET"
+ // Overwrite is selected by name, and an import can carry a different
ID for a topic that
+ // already has an active registration. Target the natural key so
PostgreSQL preserves the
+ // stored topic ID, matching MySQL and H2.
+ + " ON CONFLICT (schema_id, topic_name, deleted_at) DO UPDATE SET"
+ " topic_name = #{topicMeta.topicName},"
+ " metalake_id = #{topicMeta.metalakeId},"
+ " catalog_id = #{topicMeta.catalogId},"
@@ -123,8 +96,21 @@ public class TopicMetaPostgreSQLProvider extends
TopicMetaBaseSQLProvider {
+ " comment = #{topicMeta.comment},"
+ " properties = #{topicMeta.properties},"
+ " audit_info = #{topicMeta.auditInfo},"
- + " current_version = #{topicMeta.currentVersion},"
- + " last_version = #{topicMeta.lastVersion},"
+ // PostgreSQL evaluates both assignments against the stored row.
Qualifying the columns
+ // distinguishes them from the row that caused the conflict, and
taking the larger marker
+ // prevents an inconsistent legacy row from moving either version
backwards.
+ + " current_version = "
+ + "GREATEST("
+ + TABLE_NAME
+ + ".current_version, "
+ + TABLE_NAME
+ + ".last_version) + 1,"
+ + " last_version = "
+ + "GREATEST("
+ + TABLE_NAME
+ + ".current_version, "
+ + TABLE_NAME
+ + ".last_version) + 1,"
+ " deleted_at = #{topicMeta.deletedAt}";
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/service/TopicMetaService.java
b/core/src/main/java/org/apache/gravitino/storage/relational/service/TopicMetaService.java
index ca33b4fe2e..367c33de63 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/service/TopicMetaService.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/service/TopicMetaService.java
@@ -24,7 +24,6 @@ import com.google.common.base.Preconditions;
import java.io.IOException;
import java.util.List;
import java.util.Objects;
-import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
import java.util.stream.Collectors;
import org.apache.gravitino.Entity;
@@ -122,28 +121,28 @@ public class TopicMetaService {
newEntity.id(),
oldTopicEntity.id());
- AtomicInteger updateResult = new AtomicInteger(0);
try {
+ TopicPO newTopicPO = POConverters.updateTopicPOWithVersion(oldTopicPO,
newEntity);
SessionUtils.doMultipleWithCommit(
- () ->
- updateResult.set(
- SessionUtils.getWithoutCommit(
- TopicMetaMapper.class,
- mapper ->
- mapper.updateTopicMeta(
-
POConverters.updateTopicPOWithVersion(oldTopicPO, newEntity),
- oldTopicPO))));
+ () -> {
+ // current_version is the decision point for the whole write. Even
if another writer
+ // changes the payload and later restores it, that writer still
advances the version,
+ // so this stale update changes zero rows.
+ int updated =
+ SessionUtils.getWithoutCommit(
+ TopicMetaMapper.class,
+ mapper -> mapper.updateTopicMeta(newTopicPO, oldTopicPO));
+ if (updated == 0) {
+ throw topicWriteFailure(ident, oldTopicPO);
+ }
+ });
} catch (RuntimeException re) {
ExceptionUtils.checkSQLException(
re, Entity.EntityType.TOPIC, newEntity.nameIdentifier().toString());
throw re;
}
- if (updateResult.get() > 0) {
- return newEntity;
- } else {
- throw new IOException("Failed to update the entity: " + ident);
- }
+ return newEntity;
}
private TopicPO getTopicPOBySchemaIdAndName(Long schemaId, String topicName)
{
@@ -280,44 +279,8 @@ public class TopicMetaService {
@Monitored(metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
baseMetricName = "deleteTopic")
public boolean deleteTopic(NameIdentifier identifier) {
TopicPO topicPO = getTopicPOByIdentifier(identifier);
- Long topicId = topicPO.getTopicId();
-
- AtomicInteger deleteResult = new AtomicInteger(0);
- SessionUtils.doMultipleWithCommit(
- () ->
- deleteResult.set(
- SessionUtils.getWithoutCommit(
- TopicMetaMapper.class,
- mapper -> mapper.softDeleteTopicMetasByTopicId(topicId))),
- () -> {
- if (deleteResult.get() > 0) {
- SessionUtils.doWithoutCommit(
- OwnerMetaMapper.class,
- mapper ->
- mapper.softDeleteOwnerRelByMetadataObjectIdAndType(
- topicId, MetadataObject.Type.TOPIC.name()));
- SessionUtils.doWithoutCommit(
- SecurableObjectMapper.class,
- mapper ->
- mapper.softDeleteObjectRelsByMetadataObject(
- topicId, MetadataObject.Type.TOPIC.name()));
- SessionUtils.doWithoutCommit(
- TagMetadataObjectRelMapper.class,
- mapper ->
- mapper.softDeleteTagMetadataObjectRelsByMetadataObject(
- topicId, MetadataObject.Type.TOPIC.name()));
- SessionUtils.doWithoutCommit(
- StatisticMetaMapper.class,
- mapper -> mapper.softDeleteStatisticsByEntityId(topicId));
- SessionUtils.doWithoutCommit(
- PolicyMetadataObjectRelMapper.class,
- mapper ->
- mapper.softDeletePolicyMetadataObjectRelsByMetadataObject(
- topicId, MetadataObject.Type.TOPIC.name()));
- }
- });
-
- return deleteResult.get() > 0;
+ deleteTopicWithVersion(identifier, topicPO);
+ return true;
}
@Monitored(
@@ -370,4 +333,78 @@ public class TopicMetaService {
return POConverters.fromTopicPOs(topicPOs, firstIdent.namespace());
});
}
+
+ /**
+ * Deletes the observed topic and its dependent rows in one transaction.
+ *
+ * <p>Package access lets concurrency tests submit a stale snapshot while
exercising the same
+ * root-first ordering as the public delete path.
+ *
+ * @param identifier the topic identity observed by the caller
+ * @param observedTopicPO the topic row and OCC version observed by the
caller
+ */
+ void deleteTopicWithVersion(NameIdentifier identifier, TopicPO
observedTopicPO) {
+ SessionUtils.doMultipleWithCommit(
+ () -> {
+ // Check the root version before touching relationships. If this
snapshot is stale,
+ // throwing here stops the transaction before any dependent row can
be deleted.
+ int deleted =
+ SessionUtils.getWithoutCommit(
+ TopicMetaMapper.class,
+ mapper ->
+ mapper.softDeleteTopicMetasByTopicId(
+ observedTopicPO.getTopicId(),
observedTopicPO.getCurrentVersion()));
+ if (deleted == 0) {
+ throw topicWriteFailure(identifier, observedTopicPO);
+ }
+ },
+ () -> deleteTopicDependents(observedTopicPO.getTopicId()));
+ }
+
+ private void deleteTopicDependents(Long topicId) {
+ // The topic row has passed its version check. Every cleanup below uses
the same transaction,
+ // so a later failure also restores the root row and all earlier
relationship changes.
+ SessionUtils.doWithoutCommit(
+ OwnerMetaMapper.class,
+ mapper ->
+ mapper.softDeleteOwnerRelByMetadataObjectIdAndType(
+ topicId, MetadataObject.Type.TOPIC.name()));
+ SessionUtils.doWithoutCommit(
+ SecurableObjectMapper.class,
+ mapper ->
+ mapper.softDeleteObjectRelsByMetadataObject(topicId,
MetadataObject.Type.TOPIC.name()));
+ SessionUtils.doWithoutCommit(
+ TagMetadataObjectRelMapper.class,
+ mapper ->
+ mapper.softDeleteTagMetadataObjectRelsByMetadataObject(
+ topicId, MetadataObject.Type.TOPIC.name()));
+ SessionUtils.doWithoutCommit(
+ StatisticMetaMapper.class, mapper ->
mapper.softDeleteStatisticsByEntityId(topicId));
+ SessionUtils.doWithoutCommit(
+ PolicyMetadataObjectRelMapper.class,
+ mapper ->
+ mapper.softDeletePolicyMetadataObjectRelsByMetadataObject(
+ topicId, MetadataObject.Type.TOPIC.name()));
+ }
+
+ private RuntimeException topicWriteFailure(NameIdentifier identifier,
TopicPO observedTopicPO) {
+ // A zero-row CAS means either the same topic has a newer version, or the
topic disappeared
+ // from the name the caller used. The stable-ID lock waits for an
in-flight writer to finish so
+ // the result is classified from committed identity data.
+ TopicPO currentTopicPO =
+ SessionUtils.getWithoutCommit(
+ TopicMetaMapper.class,
+ mapper ->
mapper.selectTopicMetaByIdForUpdate(observedTopicPO.getTopicId()));
+ if (currentTopicPO == null
+ || !Objects.equals(currentTopicPO.getTopicName(),
observedTopicPO.getTopicName())
+ || !Objects.equals(currentTopicPO.getSchemaId(),
observedTopicPO.getSchemaId())
+ || !Objects.equals(currentTopicPO.getCatalogId(),
observedTopicPO.getCatalogId())
+ || !Objects.equals(currentTopicPO.getMetalakeId(),
observedTopicPO.getMetalakeId())) {
+ return new NoSuchEntityException(
+ NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
+ Entity.EntityType.TOPIC.name().toLowerCase(),
+ identifier.name());
+ }
+ return ExceptionUtils.concurrentModification(Entity.EntityType.TOPIC,
identifier);
+ }
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/utils/POConverters.java
b/core/src/main/java/org/apache/gravitino/storage/relational/utils/POConverters.java
index 49a59d32e6..889aa0d6c4 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/utils/POConverters.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/utils/POConverters.java
@@ -908,9 +908,10 @@ public class POConverters {
}
public static TopicPO updateTopicPOWithVersion(TopicPO oldTopicPO,
TopicEntity newEntity) {
- Long lastVersion = oldTopicPO.getLastVersion();
- // Will set the version to the last version + 1 when having some fields
need be multiple version
- Long nextVersion = lastVersion;
+ // Every successful alter advances beyond both stored version markers.
They normally match, but
+ // taking the larger value also prevents an inconsistent legacy row from
moving either marker
+ // backwards and making an old request current again.
+ Long nextVersion = Math.max(oldTopicPO.getCurrentVersion(),
oldTopicPO.getLastVersion()) + 1;
try {
return TopicPO.builder()
.withTopicId(oldTopicPO.getTopicId())
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/TestRelationalEntityStore.java
b/core/src/test/java/org/apache/gravitino/storage/relational/TestRelationalEntityStore.java
index 3d7e38e9ac..09a003cebc 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/TestRelationalEntityStore.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/TestRelationalEntityStore.java
@@ -38,6 +38,7 @@ import org.apache.gravitino.cache.Coherence;
import org.apache.gravitino.cache.EntityCache;
import org.apache.gravitino.cache.NoOpsCache;
import org.apache.gravitino.exceptions.NoSuchEntityException;
+import org.apache.gravitino.meta.TopicEntity;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -85,6 +86,31 @@ public class TestRelationalEntityStore {
inOrder.verify(cache).invalidate(ident, Entity.EntityType.CATALOG);
}
+ @Test
+ void testOverwriteInvalidatesCacheAfterBackendInsert()
+ throws IOException, EntityAlreadyExistsException, IllegalAccessException
{
+ NameIdentifier ident = NameIdentifier.of("metalake", "catalog", "schema",
"topic");
+ TopicEntity topic = Mockito.mock(TopicEntity.class);
+ Mockito.when(topic.nameIdentifier()).thenReturn(ident);
+ Mockito.when(topic.type()).thenReturn(Entity.EntityType.TOPIC);
+ NoOpsCache cache = (NoOpsCache) FieldUtils.readField(store, "cache", true);
+
+ Mockito.doAnswer(
+ invocation -> {
+ Mockito.verify(cache, Mockito.never()).invalidate(ident,
Entity.EntityType.TOPIC);
+ return null;
+ })
+ .when(backend)
+ .insert(topic, true);
+
+ store.put(topic, true);
+
+ InOrder inOrder = Mockito.inOrder(backend, cache);
+ inOrder.verify(backend).insert(topic, true);
+ inOrder.verify(cache).invalidate(ident, Entity.EntityType.TOPIC);
+ Mockito.verify(cache, Mockito.never()).put(topic);
+ }
+
@Test
void testDeleteInvalidatesCacheAfterBackendDelete()
throws IOException, NoSuchEntityException, IllegalAccessException {
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/mapper/TestTopicMetaSQLProviderFactory.java
b/core/src/test/java/org/apache/gravitino/storage/relational/mapper/TestTopicMetaSQLProviderFactory.java
new file mode 100644
index 0000000000..65c5e8cacf
--- /dev/null
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/mapper/TestTopicMetaSQLProviderFactory.java
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.storage.relational.mapper;
+
+import
org.apache.gravitino.storage.relational.mapper.provider.base.TopicMetaBaseSQLProvider;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class TestTopicMetaSQLProviderFactory {
+
+ private static final TopicMetaBaseSQLProvider H2_PROVIDER =
+ new TopicMetaSQLProviderFactory.TopicMetaH2Provider();
+
+ @Test
+ void testH2OverwriteCalculatesBothVersionsFromStoredRow() {
+ String sql = H2_PROVIDER.insertTopicMetaOnDuplicateKeyUpdate(null);
+ String updateClause = sql.substring(sql.indexOf(" ON DUPLICATE KEY
UPDATE"));
+ String nextVersionExpression = "GREATEST(current_version, last_version) +
1";
+
+ Assertions.assertTrue(updateClause.contains("current_version = " +
nextVersionExpression));
+ Assertions.assertTrue(updateClause.contains("last_version = " +
nextVersionExpression));
+ }
+}
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/base/TestTopicMetaBaseSQLProvider.java
b/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/base/TestTopicMetaBaseSQLProvider.java
new file mode 100644
index 0000000000..7a4e98ea2e
--- /dev/null
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/base/TestTopicMetaBaseSQLProvider.java
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.storage.relational.mapper.provider.base;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class TestTopicMetaBaseSQLProvider {
+
+ private static final TopicMetaBaseSQLProvider PROVIDER = new
TopicMetaBaseSQLProvider();
+
+ @Test
+ void testOverwriteAdvancesBeyondBothStoredVersions() {
+ String sql = PROVIDER.insertTopicMetaOnDuplicateKeyUpdate(null);
+ String updateClause = sql.substring(sql.indexOf(" ON DUPLICATE KEY
UPDATE"));
+
+ Assertions.assertTrue(
+ updateClause.contains("current_version = GREATEST(current_version,
last_version) + 1"));
+ Assertions.assertTrue(updateClause.contains("last_version =
current_version"));
+ Assertions.assertTrue(
+ updateClause.indexOf("current_version =") <
updateClause.indexOf("last_version ="));
+
Assertions.assertFalse(updateClause.contains("#{topicMeta.currentVersion}"));
+ Assertions.assertFalse(updateClause.contains("#{topicMeta.lastVersion}"));
+ }
+
+ @Test
+ void testUpdateUsesOnlyIdVersionAndActiveStateForCas() {
+ String sql = PROVIDER.updateTopicMeta(null, null);
+ String whereClause = sql.substring(sql.indexOf(" WHERE"));
+
+ Assertions.assertEquals(
+ " WHERE topic_id = #{oldTopicMeta.topicId}"
+ + " AND current_version = #{oldTopicMeta.currentVersion}"
+ + " AND deleted_at = 0",
+ whereClause);
+ }
+
+ @Test
+ void testDirectDeleteUsesVersionCas() {
+ String sql = PROVIDER.softDeleteTopicMetasByTopicId(null, null);
+
+ Assertions.assertTrue(sql.contains("AND current_version =
#{currentVersion}"));
+ Assertions.assertTrue(sql.endsWith("AND deleted_at = 0"));
+ }
+
+ @Test
+ void testConflictReadLocksActiveTopicByStableId() {
+ String sql = PROVIDER.selectTopicMetaByIdForUpdate(null);
+
+ Assertions.assertTrue(sql.contains("WHERE topic_id = #{topicId} AND
deleted_at = 0"));
+ Assertions.assertTrue(sql.endsWith("FOR UPDATE"));
+ }
+}
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/TestTopicMetaPostgreSQLProvider.java
b/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/TestTopicMetaPostgreSQLProvider.java
new file mode 100644
index 0000000000..10c5b44c31
--- /dev/null
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/TestTopicMetaPostgreSQLProvider.java
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.storage.relational.mapper.provider.postgresql;
+
+import org.apache.gravitino.storage.relational.mapper.TopicMetaMapper;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class TestTopicMetaPostgreSQLProvider {
+
+ private static final TopicMetaPostgreSQLProvider PROVIDER = new
TopicMetaPostgreSQLProvider();
+
+ @Test
+ void testOverwriteAdvancesBeyondBothStoredVersions() {
+ String sql = PROVIDER.insertTopicMetaOnDuplicateKeyUpdate(null);
+ String conflictClause = sql.substring(sql.indexOf(" ON CONFLICT"));
+ String nextVersionExpression =
+ "GREATEST("
+ + TopicMetaMapper.TABLE_NAME
+ + ".current_version, "
+ + TopicMetaMapper.TABLE_NAME
+ + ".last_version) + 1";
+
+ Assertions.assertTrue(
+ conflictClause.startsWith(" ON CONFLICT (schema_id, topic_name,
deleted_at)"));
+ Assertions.assertTrue(conflictClause.contains("current_version = " +
nextVersionExpression));
+ Assertions.assertTrue(conflictClause.contains("last_version = " +
nextVersionExpression));
+
Assertions.assertFalse(conflictClause.contains("#{topicMeta.currentVersion}"));
+
Assertions.assertFalse(conflictClause.contains("#{topicMeta.lastVersion}"));
+ }
+
+ @Test
+ void testUpdateUsesVersionCas() {
+ String sql = PROVIDER.updateTopicMeta(null, null);
+ String whereClause = sql.substring(sql.indexOf(" WHERE"));
+
+ Assertions.assertEquals(
+ " WHERE topic_id = #{oldTopicMeta.topicId}"
+ + " AND current_version = #{oldTopicMeta.currentVersion}"
+ + " AND deleted_at = 0",
+ whereClause);
+ }
+
+ @Test
+ void testDirectDeleteUsesVersionCas() {
+ String sql = PROVIDER.softDeleteTopicMetasByTopicId(null, null);
+
+ Assertions.assertTrue(sql.contains("AND current_version =
#{currentVersion}"));
+ Assertions.assertTrue(sql.endsWith("AND deleted_at = 0"));
+ }
+}
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestTopicMetaService.java
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestTopicMetaService.java
index a8a3c6fc1f..45c6c97313 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestTopicMetaService.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestTopicMetaService.java
@@ -28,30 +28,48 @@ import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.time.Instant;
import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.function.Function;
import org.apache.gravitino.Entity;
import org.apache.gravitino.EntityAlreadyExistsException;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.Namespace;
import org.apache.gravitino.exceptions.NoSuchEntityException;
+import org.apache.gravitino.exceptions.OptimisticLockException;
+import org.apache.gravitino.meta.SchemaEntity;
+import org.apache.gravitino.meta.TagEntity;
import org.apache.gravitino.meta.TopicEntity;
import org.apache.gravitino.storage.RandomIdGenerator;
import org.apache.gravitino.storage.relational.TestJDBCBackend;
+import org.apache.gravitino.storage.relational.mapper.SchemaMetaMapper;
+import org.apache.gravitino.storage.relational.mapper.TopicMetaMapper;
+import org.apache.gravitino.storage.relational.po.SchemaPO;
+import org.apache.gravitino.storage.relational.po.TopicPO;
+import org.apache.gravitino.storage.relational.utils.SessionUtils;
import org.apache.gravitino.utils.NameIdentifierUtil;
import org.apache.gravitino.utils.NamespaceUtil;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.function.Executable;
public class TestTopicMetaService extends TestJDBCBackend {
private final String metalakeName = "metalake_for_topic_test";
private final String catalogName = "catalog_for_topic_test";
private final String schemaName = "schema_for_topic_test";
+ private SchemaEntity schema;
@BeforeEach
public void prepare() throws IOException {
createAndInsertMakeLake(metalakeName);
createAndInsertCatalog(metalakeName, catalogName);
- createAndInsertSchema(metalakeName, catalogName, schemaName);
+ schema = createAndInsertSchema(metalakeName, catalogName, schemaName);
}
@TestTemplate
@@ -72,6 +90,24 @@ public class TestTopicMetaService extends TestJDBCBackend {
assertThrows(EntityAlreadyExistsException.class, () ->
backend.insert(topicCopy, false));
}
+ @TestTemplate
+ public void testInsertWaitsForConcurrentSchemaDelete() throws Exception {
+ SchemaPO observedSchemaPO =
+ SessionUtils.getWithoutCommit(
+ SchemaMetaMapper.class, mapper ->
mapper.selectSchemaMetaById(schema.id()));
+ TopicEntity topic = createTopic("topic_racing_schema_delete", "comment");
+
+ Throwable insertFailure =
+ runWhileSchemaDeleteUncommitted(
+ observedSchemaPO, () ->
TopicMetaService.getInstance().insertTopic(topic, false));
+
+ Assertions.assertInstanceOf(NoSuchEntityException.class, insertFailure);
+ Assertions.assertTrue(
+ SessionUtils.getWithoutCommit(
+ TopicMetaMapper.class, mapper ->
mapper.listTopicPOsByTopicIds(List.of(topic.id())))
+ .isEmpty());
+ }
+
@TestTemplate
public void testMetaLifeCycleFromCreationToDeletion() throws IOException {
TopicEntity topic =
@@ -152,6 +188,232 @@ public class TestTopicMetaService extends TestJDBCBackend
{
createTopicEntity(topicCopy.id(), topicCopy.namespace(),
"topic", AUDIT_INFO)));
}
+ @TestTemplate
+ public void testAlterDetectsChangeThenChangeBack() throws IOException {
+ TopicEntity topic = createTopic("topic_change_back", "original");
+ TopicMetaService.getInstance().insertTopic(topic, false);
+ TopicPO initialPO = getTopicPO(topic.id());
+
+ assertThrows(
+ OptimisticLockException.class,
+ () ->
+ TopicMetaService.getInstance()
+ .updateTopic(
+ topic.nameIdentifier(),
+ entity -> {
+ // Restore the original payload through a second
committed alter. A full-row
+ // comparison would miss both writes, but the OCC
version must still expose
+ // the stale outer update.
+ updateTopicUnchecked(
+ topic.nameIdentifier(),
+ current -> copyTopic(current, current.name(),
"temporary"));
+ updateTopicUnchecked(
+ topic.nameIdentifier(),
+ current -> copyTopic(current, current.name(),
"original"));
+ TopicEntity stale = (TopicEntity) entity;
+ return copyTopic(stale, stale.name(), "stale update");
+ }));
+
+ TopicEntity stored =
+
TopicMetaService.getInstance().getTopicByIdentifier(topic.nameIdentifier());
+ TopicPO currentPO = getTopicPO(topic.id());
+ Assertions.assertEquals("original", stored.comment());
+ Assertions.assertEquals(
+ initialPO.getCurrentVersion() + 2,
currentPO.getCurrentVersion().longValue());
+ Assertions.assertEquals(currentPO.getCurrentVersion(),
currentPO.getLastVersion());
+ }
+
+ @TestTemplate
+ public void testOverwriteAdvancesVersionAndRejectsStaleAlter() throws
IOException {
+ TopicEntity topic = createTopic("topic_overwrite_occ", "original");
+ TopicMetaService.getInstance().insertTopic(topic, false);
+ TopicPO initialPO = getTopicPO(topic.id());
+ TopicEntity replacement = copyTopic(topic, topic.name(), "overwrite
winner");
+
+ assertThrows(
+ OptimisticLockException.class,
+ () ->
+ TopicMetaService.getInstance()
+ .updateTopic(
+ topic.nameIdentifier(),
+ entity -> {
+ insertTopicUnchecked(replacement, true);
+ TopicEntity stale = (TopicEntity) entity;
+ return copyTopic(stale, stale.name(), "stale alter");
+ }));
+
+ TopicEntity stored =
+
TopicMetaService.getInstance().getTopicByIdentifier(topic.nameIdentifier());
+ TopicPO currentPO = getTopicPO(topic.id());
+ Assertions.assertEquals("overwrite winner", stored.comment());
+ Assertions.assertEquals(
+ initialPO.getCurrentVersion() + 1,
currentPO.getCurrentVersion().longValue());
+ Assertions.assertEquals(currentPO.getCurrentVersion(),
currentPO.getLastVersion());
+ }
+
+ @TestTemplate
+ public void testNaturalKeyOverwritePreservesStoredTopicId() throws
IOException {
+ // Every backend resolves overwrite by the identifier and keeps the ID
already stored for it.
+ // Relationships keyed by that ID must remain attached to the same topic.
+ TopicEntity topic = createTopic("topic_natural_key_overwrite", "original");
+ TopicMetaService.getInstance().insertTopic(topic, false);
+ TopicPO initialPO = getTopicPO(topic.id());
+ TopicEntity replacement =
+ TopicEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName(topic.name())
+ .withNamespace(topic.namespace())
+ .withComment("replacement")
+ .withProperties(topic.properties())
+ .withAuditInfo(topic.auditInfo())
+ .build();
+
+ TopicMetaService.getInstance().insertTopic(replacement, true);
+
+ TopicEntity stored =
+
TopicMetaService.getInstance().getTopicByIdentifier(topic.nameIdentifier());
+ TopicPO currentPO = getTopicPO(topic.id());
+ Assertions.assertEquals(topic.id(), stored.id());
+ Assertions.assertEquals("replacement", stored.comment());
+ Assertions.assertEquals(
+ initialPO.getCurrentVersion() + 1,
currentPO.getCurrentVersion().longValue());
+ }
+
+ @TestTemplate
+ public void testOverwriteAdvancesBeyondBothVersionMarkers() throws
IOException {
+ TopicEntity topic = createTopic("topic_overwrite_mismatched_versions",
"original");
+ TopicMetaService.getInstance().insertTopic(topic, false);
+ TopicPO initialPO = getTopicPO(topic.id());
+ TopicPO inconsistentLegacyPO = copyTopicPOWithVersions(initialPO, 3L, 5L);
+ SessionUtils.doWithCommit(
+ TopicMetaMapper.class,
+ mapper ->
+ Assertions.assertEquals(1,
mapper.updateTopicMeta(inconsistentLegacyPO, initialPO)));
+
+ TopicEntity replacement = copyTopic(topic, topic.name(), "replacement");
+ TopicMetaService.getInstance().insertTopic(replacement, true);
+
+ TopicPO currentPO = getTopicPO(topic.id());
+ Assertions.assertEquals(6L, currentPO.getCurrentVersion());
+ Assertions.assertEquals(6L, currentPO.getLastVersion());
+ Assertions.assertEquals(
+ "replacement",
+
TopicMetaService.getInstance().getTopicByIdentifier(topic.nameIdentifier()).comment());
+ }
+
+ @TestTemplate
+ public void testAlterReportsNoSuchWhenDeletedConcurrently() throws
IOException {
+ TopicEntity topic = createTopic("topic_alter_deleted", "original");
+ TopicMetaService.getInstance().insertTopic(topic, false);
+
+ assertThrows(
+ NoSuchEntityException.class,
+ () ->
+ TopicMetaService.getInstance()
+ .updateTopic(
+ topic.nameIdentifier(),
+ entity -> {
+
TopicMetaService.getInstance().deleteTopic(topic.nameIdentifier());
+ TopicEntity stale = (TopicEntity) entity;
+ return copyTopic(stale, stale.name(), "stale alter");
+ }));
+
+ assertThrows(
+ NoSuchEntityException.class,
+ () ->
TopicMetaService.getInstance().getTopicByIdentifier(topic.nameIdentifier()));
+ }
+
+ @TestTemplate
+ public void testAlterReportsNoSuchWhenRenamedConcurrently() throws
IOException {
+ TopicEntity topic = createTopic("topic_rename_conflict", "original");
+ TopicMetaService.getInstance().insertTopic(topic, false);
+ String renamedName = topic.name() + "_winner";
+ NameIdentifier renamedIdentifier = NameIdentifier.of(topic.namespace(),
renamedName);
+
+ assertThrows(
+ NoSuchEntityException.class,
+ () ->
+ TopicMetaService.getInstance()
+ .updateTopic(
+ topic.nameIdentifier(),
+ entity -> {
+ updateTopicUnchecked(
+ topic.nameIdentifier(),
+ current -> copyTopic(current, renamedName, "rename
winner"));
+ TopicEntity stale = (TopicEntity) entity;
+ return copyTopic(stale, stale.name(), "stale alter");
+ }));
+
+ assertThrows(
+ NoSuchEntityException.class,
+ () ->
TopicMetaService.getInstance().getTopicByIdentifier(topic.nameIdentifier()));
+ Assertions.assertEquals(
+ "rename winner",
+
TopicMetaService.getInstance().getTopicByIdentifier(renamedIdentifier).comment());
+ }
+
+ @TestTemplate
+ public void testStaleDeleteKeepsNewerTopicAndTagRelation() throws
IOException {
+ TopicEntity topic = createTopic("topic_stale_delete", "original");
+ TopicMetaService.getInstance().insertTopic(topic, false);
+ TagEntity tag =
+ TagEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName("topic_occ_tag")
+ .withNamespace(NamespaceUtil.ofTag(metalakeName))
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ TagMetaService.getInstance().insertTag(tag, false);
+ TagMetaService.getInstance()
+ .associateTagsWithMetadataObject(
+ topic.nameIdentifier(),
+ topic.type(),
+ new NameIdentifier[] {tag.nameIdentifier()},
+ new NameIdentifier[0]);
+ TopicPO stalePO = getTopicPO(topic.id());
+
+ TopicMetaService.getInstance()
+ .updateTopic(
+ topic.nameIdentifier(),
+ entity -> {
+ TopicEntity current = (TopicEntity) entity;
+ return copyTopic(current, current.name(), "winning alter");
+ });
+
+ assertThrows(
+ OptimisticLockException.class,
+ () ->
+
TopicMetaService.getInstance().deleteTopicWithVersion(topic.nameIdentifier(),
stalePO));
+
+ TopicEntity current =
+
TopicMetaService.getInstance().getTopicByIdentifier(topic.nameIdentifier());
+ Assertions.assertEquals("winning alter", current.comment());
+ Assertions.assertEquals(
+ List.of(tag),
+ TagMetaService.getInstance()
+ .listTagsForMetadataObject(topic.nameIdentifier(), topic.type()));
+
+
assertTrue(TopicMetaService.getInstance().deleteTopic(topic.nameIdentifier()));
+ assertTrue(
+ TagMetaService.getInstance()
+ .listAssociatedMetadataObjectsForTag(tag.nameIdentifier())
+ .isEmpty());
+ }
+
+ @TestTemplate
+ public void testDeleteReportsNoSuchWhenDeletedConcurrently() throws
IOException {
+ TopicEntity topic = createTopic("topic_double_delete", "original");
+ TopicMetaService.getInstance().insertTopic(topic, false);
+ TopicPO stalePO = getTopicPO(topic.id());
+
+ TopicMetaService.getInstance().deleteTopic(topic.nameIdentifier());
+
+ assertThrows(
+ NoSuchEntityException.class,
+ () ->
+
TopicMetaService.getInstance().deleteTopicWithVersion(topic.nameIdentifier(),
stalePO));
+ }
+
@TestTemplate
public void
testGetTopicByFullQualifiedNameMalformedNamespaceThrowsNoSuchEntityException()
throws Exception {
@@ -170,4 +432,128 @@ public class TestTopicMetaService extends TestJDBCBackend
{
assertInstanceOf(NoSuchEntityException.class,
invocationTargetException.getCause());
}
+
+ private TopicEntity createTopic(String name, String comment) {
+ return TopicEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName(name)
+ .withNamespace(NamespaceUtil.ofTopic(metalakeName, catalogName,
schemaName))
+ .withComment(comment)
+ .withProperties(Map.of("key", "value"))
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ }
+
+ /**
+ * Holds an uncommitted schema delete open and runs {@code victim} while the
schema row is locked.
+ *
+ * <p>The victim must wait for the delete to commit. It can then report the
missing parent without
+ * leaving an active topic below that deleted schema.
+ */
+ private Throwable runWhileSchemaDeleteUncommitted(SchemaPO observedSchemaPO,
Executable victim)
+ throws Exception {
+ CountDownLatch schemaDeleteLocked = new CountDownLatch(1);
+ CountDownLatch allowDeleteCommit = new CountDownLatch(1);
+ CountDownLatch victimStarted = new CountDownLatch(1);
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+ Future<Throwable> deleteResult =
+ executor.submit(
+ () -> {
+ try {
+ SessionUtils.doMultipleWithCommit(
+ () -> {
+ int deleted =
+ SessionUtils.getWithoutCommit(
+ SchemaMetaMapper.class,
+ mapper ->
+
mapper.softDeleteSchemaMetaBySchemaIdAndVersion(
+ observedSchemaPO.getSchemaId(),
+ observedSchemaPO.getCurrentVersion()));
+ Assertions.assertEquals(1, deleted);
+ schemaDeleteLocked.countDown();
+ try {
+ assertTrue(allowDeleteCommit.await(30,
TimeUnit.SECONDS));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ });
+ return null;
+ } catch (Throwable throwable) {
+ return throwable;
+ }
+ });
+ try {
+ assertTrue(schemaDeleteLocked.await(30, TimeUnit.SECONDS));
+ Future<Throwable> victimResult =
+ executor.submit(
+ () -> {
+ victimStarted.countDown();
+ try {
+ victim.execute();
+ return null;
+ } catch (Throwable throwable) {
+ return throwable;
+ }
+ });
+ assertTrue(victimStarted.await(30, TimeUnit.SECONDS));
+ assertThrows(TimeoutException.class, () -> victimResult.get(500,
TimeUnit.MILLISECONDS));
+
+ allowDeleteCommit.countDown();
+ Assertions.assertNull(deleteResult.get(30, TimeUnit.SECONDS));
+ return victimResult.get(30, TimeUnit.SECONDS);
+ } finally {
+ allowDeleteCommit.countDown();
+ executor.shutdownNow();
+ }
+ }
+
+ private TopicEntity copyTopic(TopicEntity source, String name, String
comment) {
+ return TopicEntity.builder()
+ .withId(source.id())
+ .withName(name)
+ .withNamespace(source.namespace())
+ .withComment(comment)
+ .withProperties(source.properties())
+ .withAuditInfo(source.auditInfo())
+ .build();
+ }
+
+ private TopicPO getTopicPO(Long topicId) {
+ return SessionUtils.getWithoutCommit(
+ TopicMetaMapper.class, mapper -> mapper.selectTopicMetaById(topicId));
+ }
+
+ private TopicPO copyTopicPOWithVersions(TopicPO source, Long currentVersion,
Long lastVersion) {
+ return TopicPO.builder()
+ .withTopicId(source.getTopicId())
+ .withTopicName(source.getTopicName())
+ .withMetalakeId(source.getMetalakeId())
+ .withCatalogId(source.getCatalogId())
+ .withSchemaId(source.getSchemaId())
+ .withComment(source.getComment())
+ .withProperties(source.getProperties())
+ .withAuditInfo(source.getAuditInfo())
+ .withCurrentVersion(currentVersion)
+ .withLastVersion(lastVersion)
+ .withDeletedAt(source.getDeletedAt())
+ .build();
+ }
+
+ private void updateTopicUnchecked(
+ NameIdentifier identifier, Function<TopicEntity, TopicEntity> updater) {
+ try {
+ TopicMetaService.getInstance().updateTopic(identifier, updater);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private void insertTopicUnchecked(TopicEntity topic, boolean overwrite) {
+ try {
+ TopicMetaService.getInstance().insertTopic(topic, overwrite);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
}
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/utils/TestPOConverters.java
b/core/src/test/java/org/apache/gravitino/storage/relational/utils/TestPOConverters.java
index e7b0beae5f..e3121b74c5 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/utils/TestPOConverters.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/utils/TestPOConverters.java
@@ -370,6 +370,44 @@ public class TestPOConverters {
assertEquals(expectedTopic.properties(), convertedTopic.properties());
}
+ @Test
+ public void testUpdateTopicPOAdvancesOccVersion() throws
JsonProcessingException {
+ TopicPO oldTopicPO =
+ createTopicPO(1L, "test", 1L, 1L, 1L, "old comment",
ImmutableMap.of("key", "value"));
+ TopicEntity updatedTopic =
+ createTopic(
+ 1L,
+ "test",
+ NamespaceUtil.ofTopic("test_metalake", "test_catalog",
"test_schema"),
+ "new comment",
+ ImmutableMap.of("key", "new value"));
+
+ TopicPO updatedTopicPO = POConverters.updateTopicPOWithVersion(oldTopicPO,
updatedTopic);
+
+ assertEquals(2L, updatedTopicPO.getCurrentVersion());
+ assertEquals(2L, updatedTopicPO.getLastVersion());
+ assertEquals("new comment", updatedTopicPO.getComment());
+ assertEquals(
+ updatedTopic.properties(),
+ JsonUtils.anyFieldMapper().readValue(updatedTopicPO.getProperties(),
Map.class));
+
+ TopicPO currentVersionAhead =
+ createTopicPO(
+ 1L, "test", 1L, 1L, 1L, "old comment", ImmutableMap.of("key",
"value"), 5L, 3L);
+ TopicPO lastVersionAhead =
+ createTopicPO(
+ 1L, "test", 1L, 1L, 1L, "old comment", ImmutableMap.of("key",
"value"), 3L, 5L);
+
+ // A legacy row with mismatched markers must advance beyond both values,
whichever is larger.
+ assertEquals(
+ 6L,
+ POConverters.updateTopicPOWithVersion(currentVersionAhead,
updatedTopic)
+ .getCurrentVersion());
+ assertEquals(
+ 6L,
+ POConverters.updateTopicPOWithVersion(lastVersionAhead,
updatedTopic).getCurrentVersion());
+ }
+
@Test
public void testFromMetalakePOs() throws JsonProcessingException {
MetalakePO metalakePO1 = createMetalakePO(1L, "test", "this is test");
@@ -1577,6 +1615,20 @@ public class TestPOConverters {
String comment,
Map<String, String> properties)
throws JsonProcessingException {
+ return createTopicPO(id, name, metalakeId, catalogId, schemaId, comment,
properties, 1L, 1L);
+ }
+
+ private static TopicPO createTopicPO(
+ Long id,
+ String name,
+ Long metalakeId,
+ Long catalogId,
+ Long schemaId,
+ String comment,
+ Map<String, String> properties,
+ Long currentVersion,
+ Long lastVersion)
+ throws JsonProcessingException {
AuditInfo auditInfo =
AuditInfo.builder().withCreator("creator").withCreateTime(FIX_INSTANT).build();
return TopicPO.builder()
@@ -1588,8 +1640,8 @@ public class TestPOConverters {
.withComment(comment)
.withProperties(JsonUtils.anyFieldMapper().writeValueAsString(properties))
.withAuditInfo(JsonUtils.anyFieldMapper().writeValueAsString(auditInfo))
- .withCurrentVersion(1L)
- .withLastVersion(1L)
+ .withCurrentVersion(currentVersion)
+ .withLastVersion(lastVersion)
.withDeletedAt(0L)
.build();
}
