This is an automated email from the ASF dual-hosted git repository.
jerryshao pushed a commit to branch branch-domain-dev
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/branch-domain-dev by this push:
new 4510187346 [#13884] feat(core): Add locked domain operations to the
entity store (#14496)
4510187346 is described below
commit 4510187346981459138b714acf6eba50862653e9
Author: Jerry Shao <[email protected]>
AuthorDate: Tue Nov 3 14:23:52 2026 +0800
[#13884] feat(core): Add locked domain operations to the entity store
(#14496)
### What changes were proposed in this pull request?
Add `SupportsDomainOperations`, exposed through
`EntityStore.domainOperations()`, for transactional domain creation,
parent changes and status changes.
Operations lock the relevant rows and run the caller's validation
against the locked state. This includes atomic sub-domain creation,
attach/move/detach, bounded retries for status changes, and protection
against generic writes bypassing these locks.
This is the storage portion of #13884. The manager, backing tags,
authorization and events are separate changes.
### Why are the changes needed?
Hierarchy and status checks must observe related domains under the same
locks as the write. Independent entity and relation writes cannot
provide that transaction boundary.
Part of #13884.
### Does this PR introduce _any_ user-facing change?
No. This adds internal storage operations for the domain manager.
### How was this patch tested?
Java and test compilation, Spotless, and `:core:javadoc` passed.
Related core unit tests and H2 tests passed, covering rollback,
parent-change retries, stale session-cache reads and row-lock
concurrency.
MySQL and PostgreSQL verification is left to CI.
---------
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../java/org/apache/gravitino/EntityStore.java | 10 +
.../apache/gravitino/SupportsDomainOperations.java | 210 +++++++
.../gravitino/storage/relational/JDBCBackend.java | 41 ++
.../storage/relational/RelationalBackend.java | 4 +-
.../storage/relational/RelationalEntityStore.java | 68 +-
.../relational/service/DomainMetaService.java | 121 +++-
.../relational/service/DomainRelService.java | 405 ++++++++++--
.../relational/service/LiveEndpointService.java | 9 +-
.../storage/relational/utils/SessionUtils.java | 15 +
.../TestRelationalEntityStoreLockWaitTimeout.java | 36 +-
.../relational/service/TestDomainMetaService.java | 43 +-
.../relational/service/TestDomainRelService.java | 698 ++++++++++++++++++++-
12 files changed, 1566 insertions(+), 94 deletions(-)
diff --git a/core/src/main/java/org/apache/gravitino/EntityStore.java
b/core/src/main/java/org/apache/gravitino/EntityStore.java
index 4ed86ff207..77402fc860 100644
--- a/core/src/main/java/org/apache/gravitino/EntityStore.java
+++ b/core/src/main/java/org/apache/gravitino/EntityStore.java
@@ -368,4 +368,14 @@ public interface EntityStore extends Closeable {
default SupportsRelationOperations relationOperations() {
throw new UnsupportedOperationException("relation operations are not
supported");
}
+
+ /**
+ * Get the domain operations that are supported by the entity store.
+ *
+ * @return the domain operations that are supported by the entity store
+ * @throws UnsupportedOperationException if the domain operations are not
supported
+ */
+ default SupportsDomainOperations domainOperations() {
+ throw new UnsupportedOperationException("domain operations are not
supported");
+ }
}
diff --git
a/core/src/main/java/org/apache/gravitino/SupportsDomainOperations.java
b/core/src/main/java/org/apache/gravitino/SupportsDomainOperations.java
new file mode 100644
index 0000000000..cfeaac272a
--- /dev/null
+++ b/core/src/main/java/org/apache/gravitino/SupportsDomainOperations.java
@@ -0,0 +1,210 @@
+/*
+ * 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;
+
+import com.google.common.collect.ImmutableList;
+import java.io.IOException;
+import java.util.List;
+import java.util.Optional;
+import java.util.function.Consumer;
+import javax.annotation.Nullable;
+import org.apache.gravitino.domain.Domain;
+import org.apache.gravitino.exceptions.NoSuchEntityException;
+import org.apache.gravitino.exceptions.OptimisticLockException;
+import org.apache.gravitino.meta.AuditInfo;
+import org.apache.gravitino.meta.DomainEntity;
+
+/**
+ * The domain changes whose rules depend on other domains, or whose
authorization depends on the
+ * current parent: creating a domain, changing or removing its parent and
changing its status. Each
+ * change runs in one transaction. It locks the rows of the domains involved,
reads them, hands them
+ * to the caller's check as a {@link LockedDomainState}, and writes the change
only if the check
+ * returns. A check that throws rolls the change back, and the exception
reaches the caller
+ * unchanged, unless it is caused by a database error, which is reported like
any other failure of
+ * the store.
+ *
+ * <p>The store only locks, reads and writes; the rules themselves, such as
how deep the hierarchy
+ * may be or which status a parent needs, are the caller's check. Since the
check sees the domains
+ * as they are under the locks, a rule it enforces cannot be broken by a
concurrent change that the
+ * locks exclude.
+ *
+ * <p>In a {@link LockedDomainState}, the domain being changed and its parent
are read with locking
+ * reads, which do not read the backing tag name, so their {@link
DomainEntity#backingTag()} is
+ * empty; their backing tag ID and value are set. For an insert, the domain is
the caller's new
+ * entity.
+ */
+public interface SupportsDomainOperations {
+
+ /**
+ * Inserts a domain. If the domain has a parent, it is inserted as a
sub-domain of that parent,
+ * together with its parent edge.
+ *
+ * <p>The metalake is locked shared and the parent, if any, is locked
shared, so the parent cannot
+ * be attached elsewhere, deleted or unpublished before the change commits.
The domain is written
+ * before the check runs, so a duplicate name fails first. The check gets
the new domain, the
+ * parent with its own parent set, and no sub-domains. This operation must
own its transaction so
+ * a failed check rolls back the inserted domain; it cannot run inside
another transaction.
+ *
+ * @param domain The domain to insert.
+ * @param check The check of the locked state.
+ * @throws IOException If the domain cannot be stored.
+ * @throws EntityAlreadyExistsException If a domain with the same name
exists.
+ * @throws NoSuchEntityException If the metalake or the parent does not
exist.
+ * @throws IllegalStateException If it is called inside a transaction.
+ */
+ void insertDomain(DomainEntity domain, Consumer<LockedDomainState> check)
+ throws IOException, EntityAlreadyExistsException, NoSuchEntityException;
+
+ /**
+ * Sets the parent of a domain: attaches a top-level domain, or moves a
sub-domain to another
+ * parent. Setting the current parent changes nothing.
+ *
+ * <p>The domain is locked exclusively and the new parent shared, in domain
ID order, so before
+ * the change commits neither one's parent can change, the new parent's
status cannot change, and
+ * no sub-domain can be added to the domain. The check gets the domain with
its current parent
+ * set, the new parent with its own parent set, and the sub-domains of the
domain.
+ *
+ * @param ident The identifier of the domain.
+ * @param parentIdent The identifier of the new parent, in the same metalake.
+ * @param auditInfo The audit info of the change. A new parent edge is
created by its creator; a
+ * moved edge keeps its creator and takes its last modifier.
+ * @param check The check of the locked state.
+ * @return The domain after the change.
+ * @throws IOException If the change cannot be stored.
+ * @throws NoSuchEntityException If the domain or the new parent does not
exist.
+ * @throws IllegalArgumentException If the new parent is the domain itself
or is in another
+ * metalake.
+ */
+ DomainEntity setDomainParentAndGet(
+ NameIdentifier ident,
+ NameIdentifier parentIdent,
+ AuditInfo auditInfo,
+ Consumer<LockedDomainState> check)
+ throws IOException, NoSuchEntityException;
+
+ /**
+ * Removes the parent edge of a domain, which makes it top-level. Removing
the parent of a
+ * top-level domain changes nothing.
+ *
+ * <p>The domain is locked exclusively, so its parent cannot change before
the change commits. The
+ * check gets the domain with its current parent set, which is the parent
the edge is removed
+ * from; the parent itself is not locked, so the state has no parent entity
and no sub-domains.
+ *
+ * @param ident The identifier of the domain.
+ * @param check The check of the locked state.
+ * @return The domain after the change.
+ * @throws IOException If the change cannot be stored.
+ * @throws NoSuchEntityException If the domain does not exist.
+ */
+ DomainEntity removeDomainParentAndGet(NameIdentifier ident,
Consumer<LockedDomainState> check)
+ throws IOException, NoSuchEntityException;
+
+ /**
+ * Sets the status of a domain. Setting the current status writes nothing.
+ *
+ * <p>The domain is locked exclusively and its current parent shared, in
domain ID order, so
+ * before the change commits the parent's status and the domain's parent
cannot change, and no
+ * sub-domain can be added to the domain or change its status; a sub-domain
can still be detached
+ * or deleted. The parent is read before the locks are taken; if it changed
by the time they are
+ * held, the change is rolled back and retried with the new parent, a
bounded number of times, so
+ * it cannot run inside another transaction. The check gets the domain with
its current parent
+ * set, that parent, and the sub-domains of the domain.
+ *
+ * @param ident The identifier of the domain.
+ * @param status The new status.
+ * @param auditInfo The audit info of the change; the domain keeps its
creator and takes its last
+ * modifier.
+ * @param check The check of the locked state.
+ * @return The domain after the change.
+ * @throws IOException If the change cannot be stored.
+ * @throws NoSuchEntityException If the domain does not exist.
+ * @throws OptimisticLockException If the parent kept changing until the
retries ran out.
+ * @throws IllegalStateException If it is called inside a transaction.
+ */
+ DomainEntity updateDomainStatusAndGet(
+ NameIdentifier ident,
+ Domain.Status status,
+ AuditInfo auditInfo,
+ Consumer<LockedDomainState> check)
+ throws IOException, NoSuchEntityException;
+
+ /**
+ * Lists the sub-domains of a domain.
+ *
+ * @param parentIdent The identifier of the parent domain.
+ * @return The sub-domains, each with its parent set.
+ * @throws IOException If the sub-domains cannot be read.
+ * @throws NoSuchEntityException If the parent domain does not exist.
+ */
+ List<DomainEntity> listSubDomains(NameIdentifier parentIdent)
+ throws IOException, NoSuchEntityException;
+
+ /** The domains involved in a change, as read under the change's row locks.
*/
+ final class LockedDomainState {
+
+ private final DomainEntity domain;
+ @Nullable private final DomainEntity parent;
+ private final List<DomainEntity> subDomains;
+
+ /**
+ * Creates the locked state of a change.
+ *
+ * @param domain The domain being changed, with its current parent set.
+ * @param parent The parent involved in the change, with its own parent
set, or null if none.
+ * @param subDomains The sub-domains of the domain being changed.
+ */
+ public LockedDomainState(
+ DomainEntity domain, @Nullable DomainEntity parent, List<DomainEntity>
subDomains) {
+ this.domain = domain;
+ this.parent = parent;
+ this.subDomains = ImmutableList.copyOf(subDomains);
+ }
+
+ /**
+ * Returns the domain being changed, with its current parent set. For an
insert, it is the new
+ * domain.
+ *
+ * @return The domain being changed.
+ */
+ public DomainEntity domain() {
+ return domain;
+ }
+
+ /**
+ * Returns the parent involved in the change, with its own parent set: the
parent of a new
+ * sub-domain, the new parent when the parent is set, or the current
parent when the status
+ * changes. It is empty when the parent is removed, since that parent is
not locked.
+ *
+ * @return The parent, or empty if the change involves none.
+ */
+ public Optional<DomainEntity> parent() {
+ return Optional.ofNullable(parent);
+ }
+
+ /**
+ * Returns the sub-domains of the domain being changed. For an insert,
there are none, and when
+ * the parent is removed they are not read.
+ *
+ * @return The sub-domains.
+ */
+ public List<DomainEntity> subDomains() {
+ return subDomains;
+ }
+ }
+}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/JDBCBackend.java
b/core/src/main/java/org/apache/gravitino/storage/relational/JDBCBackend.java
index 96add71e43..40fabba0d1 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/JDBCBackend.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/JDBCBackend.java
@@ -50,7 +50,9 @@ import org.apache.gravitino.RelationalEntity;
import org.apache.gravitino.SupportsRelationOperations;
import org.apache.gravitino.UnsupportedEntityTypeException;
import org.apache.gravitino.cache.BaseEntityCache;
+import org.apache.gravitino.domain.Domain;
import org.apache.gravitino.exceptions.NoSuchEntityException;
+import org.apache.gravitino.meta.AuditInfo;
import org.apache.gravitino.meta.BaseMetalake;
import org.apache.gravitino.meta.CatalogEntity;
import org.apache.gravitino.meta.DomainEntity;
@@ -77,6 +79,7 @@ import
org.apache.gravitino.storage.relational.database.H2Database;
import org.apache.gravitino.storage.relational.po.cache.OperateType;
import org.apache.gravitino.storage.relational.service.CatalogMetaService;
import org.apache.gravitino.storage.relational.service.DomainMetaService;
+import org.apache.gravitino.storage.relational.service.DomainRelService;
import org.apache.gravitino.storage.relational.service.FilesetMetaService;
import org.apache.gravitino.storage.relational.service.FunctionMetaService;
import org.apache.gravitino.storage.relational.service.GroupMetaService;
@@ -749,6 +752,44 @@ public class JDBCBackend implements RelationalBackend,
SupportsOrphanedRelationC
}
}
+ @Override
+ public void insertDomain(DomainEntity domain, Consumer<LockedDomainState>
check)
+ throws IOException {
+ DomainMetaService.getInstance().insertDomain(domain, check);
+ }
+
+ @Override
+ public DomainEntity setDomainParentAndGet(
+ NameIdentifier ident,
+ NameIdentifier parentIdent,
+ AuditInfo auditInfo,
+ Consumer<LockedDomainState> check)
+ throws IOException {
+ return DomainRelService.getInstance()
+ .setDomainParentAndGet(ident, parentIdent, auditInfo, check);
+ }
+
+ @Override
+ public DomainEntity removeDomainParentAndGet(
+ NameIdentifier ident, Consumer<LockedDomainState> check) throws
IOException {
+ return DomainRelService.getInstance().removeDomainParentAndGet(ident,
check);
+ }
+
+ @Override
+ public DomainEntity updateDomainStatusAndGet(
+ NameIdentifier ident,
+ Domain.Status status,
+ AuditInfo auditInfo,
+ Consumer<LockedDomainState> check)
+ throws IOException {
+ return DomainRelService.getInstance().updateDomainStatusAndGet(ident,
status, auditInfo, check);
+ }
+
+ @Override
+ public List<DomainEntity> listSubDomains(NameIdentifier parentIdent) {
+ return DomainRelService.getInstance().listSubDomains(parentIdent);
+ }
+
@Override
public void insertRelation(
SupportsRelationOperations.Type relType,
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/RelationalBackend.java
b/core/src/main/java/org/apache/gravitino/storage/relational/RelationalBackend.java
index fa64b409dc..d5593298c0 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/RelationalBackend.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/RelationalBackend.java
@@ -32,12 +32,14 @@ import org.apache.gravitino.EntityStore;
import org.apache.gravitino.HasIdentifier;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.Namespace;
+import org.apache.gravitino.SupportsDomainOperations;
import org.apache.gravitino.SupportsRelationOperations;
import org.apache.gravitino.exceptions.NoSuchEntityException;
import org.apache.gravitino.meta.SchemaEntity;
/** Interface defining the operations for a Relation Backend. */
-public interface RelationalBackend extends Closeable,
SupportsRelationOperations {
+public interface RelationalBackend
+ extends Closeable, SupportsRelationOperations, SupportsDomainOperations {
/**
* Initializes the Relational Backend environment with the provided
configuration.
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 2a596f8f54..6a683d028d 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
@@ -48,6 +48,7 @@ import org.apache.gravitino.RelationEdgeTarget;
import org.apache.gravitino.RelationQuery;
import org.apache.gravitino.RelationUpdate;
import org.apache.gravitino.RelationalEntity;
+import org.apache.gravitino.SupportsDomainOperations;
import org.apache.gravitino.SupportsRelationOperations;
import org.apache.gravitino.cache.BaseEntityCache;
import org.apache.gravitino.cache.CacheFactory;
@@ -56,8 +57,11 @@ import org.apache.gravitino.cache.Coherence;
import org.apache.gravitino.cache.EntityCache;
import org.apache.gravitino.cache.EntityCacheKey;
import org.apache.gravitino.cache.NoOpsCache;
+import org.apache.gravitino.domain.Domain;
import org.apache.gravitino.exceptions.NoSuchEntityException;
import org.apache.gravitino.exceptions.OptimisticLockException;
+import org.apache.gravitino.meta.AuditInfo;
+import org.apache.gravitino.meta.DomainEntity;
import org.apache.gravitino.meta.SchemaEntity;
import org.apache.gravitino.metrics.MetricsSystem;
import org.apache.gravitino.metrics.source.EntityChangeLogMetricsSource;
@@ -81,7 +85,10 @@ import org.slf4j.LoggerFactory;
* ExceptionUtils#isLockWaitTimeout}.
*/
public class RelationalEntityStore
- implements EntityStore, SupportsRelationOperations,
SupportsEntityChangeLog {
+ implements EntityStore,
+ SupportsRelationOperations,
+ SupportsDomainOperations,
+ SupportsEntityChangeLog {
private static final Logger LOGGER =
LoggerFactory.getLogger(RelationalEntityStore.class);
public static final ImmutableMap<String, String> RELATIONAL_BACKENDS =
ImmutableMap.of(
@@ -409,6 +416,65 @@ public class RelationalEntityStore
return this;
}
+ @Override
+ public SupportsDomainOperations domainOperations() {
+ return this;
+ }
+
+ @Override
+ public void insertDomain(DomainEntity domain, Consumer<LockedDomainState>
check)
+ throws IOException {
+ try {
+ backend.insertDomain(domain, check);
+ } catch (RuntimeException re) {
+ throw ExceptionUtils.lockWaitTimeoutAsConflict(
+ re, Entity.EntityType.DOMAIN, domain.nameIdentifier());
+ }
+ }
+
+ @Override
+ public DomainEntity setDomainParentAndGet(
+ NameIdentifier ident,
+ NameIdentifier parentIdent,
+ AuditInfo auditInfo,
+ Consumer<LockedDomainState> check)
+ throws IOException {
+ try {
+ return backend.setDomainParentAndGet(ident, parentIdent, auditInfo,
check);
+ } catch (RuntimeException re) {
+ throw ExceptionUtils.lockWaitTimeoutAsConflict(re,
Entity.EntityType.DOMAIN, ident);
+ }
+ }
+
+ @Override
+ public DomainEntity removeDomainParentAndGet(
+ NameIdentifier ident, Consumer<LockedDomainState> check) throws
IOException {
+ try {
+ return backend.removeDomainParentAndGet(ident, check);
+ } catch (RuntimeException re) {
+ throw ExceptionUtils.lockWaitTimeoutAsConflict(re,
Entity.EntityType.DOMAIN, ident);
+ }
+ }
+
+ @Override
+ public DomainEntity updateDomainStatusAndGet(
+ NameIdentifier ident,
+ Domain.Status status,
+ AuditInfo auditInfo,
+ Consumer<LockedDomainState> check)
+ throws IOException {
+ try {
+ return backend.updateDomainStatusAndGet(ident, status, auditInfo, check);
+ } catch (RuntimeException re) {
+ throw ExceptionUtils.lockWaitTimeoutAsConflict(re,
Entity.EntityType.DOMAIN, ident);
+ }
+ }
+
+ @Override
+ public List<DomainEntity> listSubDomains(NameIdentifier parentIdent) throws
IOException {
+ return backend.listSubDomains(parentIdent);
+ }
+
@Override
public <E extends Entity & HasIdentifier> List<E> listEntitiesByRelation(
Type relType, NameIdentifier nameIdentifier, Entity.EntityType
identType, boolean allFields)
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/service/DomainMetaService.java
b/core/src/main/java/org/apache/gravitino/storage/relational/service/DomainMetaService.java
index 07dde79f3d..a9acff58ee 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/service/DomainMetaService.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/service/DomainMetaService.java
@@ -26,6 +26,7 @@ import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
+import java.util.function.Consumer;
import java.util.function.Function;
import java.util.stream.Collectors;
import org.apache.gravitino.Entity;
@@ -34,8 +35,10 @@ import org.apache.gravitino.HasIdentifier;
import org.apache.gravitino.MetadataObject;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.Namespace;
+import org.apache.gravitino.SupportsDomainOperations.LockedDomainState;
import org.apache.gravitino.exceptions.NoSuchEntityException;
import org.apache.gravitino.exceptions.NonEmptyEntityException;
+import org.apache.gravitino.meta.AuditInfo;
import org.apache.gravitino.meta.DomainEntity;
import org.apache.gravitino.metrics.Monitored;
import org.apache.gravitino.storage.relational.mapper.DomainMetaMapper;
@@ -124,32 +127,32 @@ public class DomainMetaService {
}
/**
- * Inserts a domain.
+ * Inserts a top-level domain. A sub-domain, whose parent must be locked and
checked, is inserted
+ * with {@link #insertDomain(DomainEntity, Consumer)}. Replacing a domain
keeps its status and its
+ * parent: the status is changed only under the locks its rules need, by
{@link
+ * DomainRelService#updateDomainStatusAndGet}.
*
* @param domainEntity The domain to insert.
* @param overwritten Whether to replace an active domain with the same name
or ID.
* @throws IOException If the domain cannot be stored.
* @throws EntityAlreadyExistsException If a domain with the same name
exists and {@code
* overwritten} is false.
+ * @throws IllegalArgumentException If the domain has a parent, or replaces
a domain with another
+ * status.
*/
@Monitored(
metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
baseMetricName = "insertDomain")
public void insertDomain(DomainEntity domainEntity, boolean overwritten)
throws IOException {
+ Preconditions.checkArgument(
+ !domainEntity.parent().isPresent(),
+ "The domain %s has a parent and must be inserted with its parent
locked",
+ domainEntity.nameIdentifier());
Namespace ns = domainEntity.namespace();
String metalakeName = ns.level(0);
try {
- MetalakePO metalakePO =
- SessionUtils.getWithoutCommit(
- MetalakeMetaMapper.class, mapper ->
mapper.selectMetalakeMetaByName(metalakeName));
- if (metalakePO == null) {
- throw new NoSuchEntityException(
- NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
- Entity.EntityType.METALAKE.name().toLowerCase(),
- metalakeName);
- }
-
+ MetalakePO metalakePO = getMetalakePO(metalakeName);
DomainPO domainPO = DomainPO.initializeDomainPO(domainEntity,
metalakePO.getMetalakeId());
SessionUtils.doMultipleWithCommit(
@@ -173,6 +176,75 @@ public class DomainMetaService {
}
}
+ /**
+ * Inserts a new domain after a check of the locked state. If the domain has
a parent, it is
+ * inserted as a sub-domain of that parent, together with its parent edge,
in one transaction.
+ *
+ * <p>The metalake row is locked shared, then the parent row, if any, is
locked shared, so the
+ * parent cannot be attached elsewhere, deleted or have its status changed
until the domain is
+ * committed. The check gets the new domain, the parent with its own parent
set, and no
+ * sub-domains.
+ *
+ * <p>This operation must own its transaction so a failed check rolls back
the inserted domain. It
+ * cannot run inside another transaction.
+ *
+ * @param domainEntity The domain to insert; its parent, if set, is the
parent to insert it under.
+ * @param check The check of the locked state; nothing is inserted if it
throws.
+ * @throws IOException If the domain cannot be stored.
+ * @throws EntityAlreadyExistsException If a domain with the same name
exists.
+ * @throws NoSuchEntityException If the metalake or the parent does not
exist.
+ * @throws IllegalStateException If it is called inside a transaction.
+ */
+ @Monitored(
+ metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
+ baseMetricName = "insertDomainWithCheck")
+ public void insertDomain(DomainEntity domainEntity,
Consumer<LockedDomainState> check)
+ throws IOException {
+ Preconditions.checkState(
+ !SessionUtils.isInTransaction(),
+ "The domain %s cannot be inserted with a check inside another
transaction",
+ domainEntity.nameIdentifier());
+ Namespace ns = domainEntity.namespace();
+ String metalakeName = ns.level(0);
+
+ try {
+ MetalakePO metalakePO = getMetalakePO(metalakeName);
+ DomainPO domainPO = DomainPO.initializeDomainPO(domainEntity,
metalakePO.getMetalakeId());
+
+ SessionUtils.doMultipleWithCommit(
+ () ->
+ lockMetalakeForDomainWrite(metalakePO.getMetalakeName(),
metalakePO.getMetalakeId()),
+ () -> {
+ DomainEntity parent =
+ domainEntity
+ .parent()
+ .map(
+ parentName ->
+ DomainRelService.getInstance()
+
.lockAndGetParentOfNewSubDomain(NameIdentifier.of(ns, parentName)))
+ .orElse(null);
+ // The domain is written before the check, so that a duplicate
name fails as such.
+ insertNewDomainWithoutCommit(domainPO);
+ check.accept(new LockedDomainState(domainEntity, parent,
Collections.emptyList()));
+ if (parent != null) {
+ DomainRelService.getInstance()
+ .insertDomainRelOfNewSubDomain(
+ metalakePO.getMetalakeId(),
+ parent.id(),
+ domainPO.domainId(),
+ AuditInfo.builder()
+ .withCreator(domainEntity.auditInfo().creator())
+
.withCreateTime(domainEntity.auditInfo().createTime())
+ .build());
+ }
+ });
+ } catch (RuntimeException e) {
+ ExceptionUtils.checkSQLException(
+ e, Entity.EntityType.DOMAIN,
domainEntity.nameIdentifier().toString());
+ throw e;
+ }
+ }
+
/**
* Updates a domain. The domain row is locked before the updater runs, so
the updater is given the
* current domain and concurrent alters apply one after another instead of
failing on each other.
@@ -187,8 +259,9 @@ public class DomainMetaService {
* changing the new domain; a later call with the name acts on the new
domain.
*
* <p>The returned domain is read back in the same transaction, so its
backing tag name comes from
- * the tag it is now bound to rather than from the updater's entity. The
parent cannot be changed
- * by an update, an updater that changes it is rejected; {@link
DomainRelService} changes it.
+ * the tag it is now bound to rather than from the updater's entity. The
parent and the status
+ * cannot be changed by an update, an updater that changes either is
rejected; {@link
+ * DomainRelService} changes them under the locks their rules need.
*
* @param identifier The domain identifier.
* @param updater The function that returns the updated domain.
@@ -230,6 +303,11 @@ public class DomainMetaService {
Objects.equals(oldDomainEntity.parent(),
updatedDomainEntity.parent()),
"The parent of domain %s cannot be changed by an update, use
DomainRelService",
identifier);
+ Preconditions.checkArgument(
+ oldDomainEntity.status() == updatedDomainEntity.status(),
+ "The status of domain %s cannot be changed by an update, use"
+ + " DomainRelService#updateDomainStatusAndGet",
+ identifier);
DomainPO newDomainPO = DomainPO.updateDomainPO(lockedDomainPO,
updatedDomainEntity);
updateDomainWithVersion(identifier, mapper, lockedDomainPO,
newDomainPO);
@@ -414,6 +492,10 @@ public class DomainMetaService {
return;
}
+ Preconditions.checkArgument(
+ Objects.equals(existingDomainPO.status(),
domainEntity.status().name()),
+ "The status of domain %s cannot be changed by replacing it",
+ domainEntity.nameIdentifier());
DomainPO replacementDomainPO = DomainPO.updateDomainPO(existingDomainPO,
domainEntity);
NameIdentifier observedIdentifier =
NameIdentifier.of(domainEntity.namespace(),
existingDomainPO.domainName());
@@ -499,6 +581,19 @@ public class DomainMetaService {
&& Objects.equals(current.metalakeId(),
observedDomainPO.metalakeId()));
}
+ private MetalakePO getMetalakePO(String metalakeName) {
+ MetalakePO metalakePO =
+ SessionUtils.getWithoutCommit(
+ MetalakeMetaMapper.class, mapper ->
mapper.selectMetalakeMetaByName(metalakeName));
+ if (metalakePO == null) {
+ throw new NoSuchEntityException(
+ NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE,
+ Entity.EntityType.METALAKE.name().toLowerCase(),
+ metalakeName);
+ }
+ return metalakePO;
+ }
+
private DomainPO getDomainPOByMetalakeAndName(String metalakeName, String
domainName) {
DomainPO domainPO =
SessionUtils.getWithoutCommit(
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/service/DomainRelService.java
b/core/src/main/java/org/apache/gravitino/storage/relational/service/DomainRelService.java
index a2718d553a..bc0fdc0cca 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/service/DomainRelService.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/service/DomainRelService.java
@@ -20,6 +20,7 @@ package org.apache.gravitino.storage.relational.service;
import static
org.apache.gravitino.metrics.source.MetricsSource.GRAVITINO_RELATIONAL_STORE_METRIC_NAME;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import java.io.IOException;
import java.util.Collections;
@@ -27,13 +28,19 @@ import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
+import java.util.function.Consumer;
import java.util.stream.Collectors;
import javax.annotation.Nullable;
+import org.apache.commons.lang3.tuple.Pair;
import org.apache.gravitino.Entity;
import org.apache.gravitino.EntityAlreadyExistsException;
import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.Namespace;
+import org.apache.gravitino.SupportsDomainOperations.LockedDomainState;
+import org.apache.gravitino.domain.Domain;
import org.apache.gravitino.exceptions.NoSuchEntityException;
import org.apache.gravitino.exceptions.NonEmptyEntityException;
+import org.apache.gravitino.exceptions.OptimisticLockException;
import org.apache.gravitino.meta.AuditInfo;
import org.apache.gravitino.meta.DomainEntity;
import org.apache.gravitino.metrics.Monitored;
@@ -61,13 +68,20 @@ import
org.apache.gravitino.storage.relational.utils.SessionUtils;
* the sub-domain, so the edge read under that lock is current. Each locked
row is re-read and must
* still have the observed name, otherwise the change fails as if it did not
exist.
*
- * <p>This service does not enforce the shape of the hierarchy. The domain
manager keeps it one
- * level deep, which also keeps it acyclic.
+ * <p>Status changes are made here too, since the status of a domain must
agree with its parent and
+ * sub-domains: they lock the domain exclusively and its current parent
shared, in the same order.
+ *
+ * <p>This service does not enforce the shape of the hierarchy or the status
rules. The changes that
+ * set a parent or a status take a check from the caller, which runs on the
domains as they are
+ * under the locks; the domain manager's checks keep the hierarchy one level
deep, which also keeps
+ * it acyclic.
*/
public class DomainRelService {
private static final DomainRelService INSTANCE = new DomainRelService();
+ private static final int MAX_STATUS_UPDATE_ATTEMPTS = 3;
+
private DomainRelService() {}
/**
@@ -140,9 +154,242 @@ public class DomainRelService {
.collect(Collectors.toList());
}
+ /**
+ * Sets the parent of a domain: inserts its parent edge if it is top-level,
or re-points the edge
+ * if it is a sub-domain. Setting the current parent changes nothing.
+ *
+ * <p>Whether the edge is inserted or re-pointed is decided under the lock
on the domain, so a
+ * concurrent attach or detach of the same domain cannot make the change
fail. The check gets the
+ * domain with its current parent, the new parent with its own parent, and
the sub-domains of the
+ * domain, all read under the locks.
+ *
+ * @param childIdent The identifier of the domain.
+ * @param parentIdent The identifier of the new parent domain, in the same
metalake.
+ * @param auditInfo The audit info of the change. A new edge is created by
its creator; a
+ * re-pointed edge keeps its creator and takes its last modifier.
+ * @param check The check of the locked state; the change is rolled back if
it throws.
+ * @return The domain after the change.
+ * @throws IOException If the edge cannot be stored.
+ * @throws NoSuchEntityException If either domain does not exist.
+ * @throws IllegalArgumentException If the parent is the domain itself or is
in another metalake.
+ */
+ @Monitored(
+ metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
+ baseMetricName = "setDomainParentAndGet")
+ public DomainEntity setDomainParentAndGet(
+ NameIdentifier childIdent,
+ NameIdentifier parentIdent,
+ AuditInfo auditInfo,
+ Consumer<LockedDomainState> check)
+ throws IOException {
+ checkSameMetalakeAndDistinct(childIdent, parentIdent);
+ DomainEntity[] changed = new DomainEntity[1];
+ try {
+ DomainPO childPO = getDomainPO(childIdent);
+ DomainPO parentPO = getDomainPO(parentIdent);
+ checkDistinctRows(childIdent, childPO, parentPO);
+ SessionUtils.doMultipleWithCommit(
+ () -> {
+ Pair<DomainPO, DomainPO> locked =
lockMetalakeAndDomains(childIdent, childPO, parentPO);
+ Namespace namespace = childIdent.namespace();
+ DomainRelPO oldDomainRelPO = selectDomainRel(childPO.domainId());
+ check.accept(
+ new LockedDomainState(
+ toEntityWithParent(locked.getLeft(), namespace),
+ toEntityWithParent(locked.getRight(), namespace),
+ listSubDomainsOf(locked.getLeft(), namespace)));
+
+ if (oldDomainRelPO == null) {
+ DomainRelPO domainRelPO =
+ DomainRelPO.initializeDomainRelPO(
+ childPO.metalakeId(), parentPO.domainId(),
childPO.domainId(), auditInfo);
+ SessionUtils.doWithoutCommit(
+ DomainRelMapper.class, mapper ->
mapper.insertDomainRel(domainRelPO));
+ } else if (!Objects.equals(oldDomainRelPO.parentId(),
parentPO.domainId())) {
+ DomainRelPO newDomainRelPO =
+ DomainRelPO.updateDomainRelPO(oldDomainRelPO,
parentPO.domainId(), auditInfo);
+ OccWriteSupport.updateWithVersion(
+ () ->
+ SessionUtils.getWithoutCommit(
+ DomainRelMapper.class,
+ mapper -> mapper.updateDomainRel(newDomainRelPO,
oldDomainRelPO)),
+ () ->
+
ExceptionUtils.concurrentModification(Entity.EntityType.DOMAIN, childIdent));
+ }
+ changed[0] = readDomain(childPO.domainId(), namespace);
+ });
+ } catch (RuntimeException e) {
+ ExceptionUtils.checkSQLException(e, Entity.EntityType.DOMAIN,
childIdent.toString());
+ throw e;
+ }
+ return changed[0];
+ }
+
+ /**
+ * Removes the parent edge of a domain, which makes it top-level, and
returns the domain as it is
+ * after the change. Removing the parent of a top-level domain changes
nothing.
+ *
+ * <p>The check gets the domain with its current parent, read under the
exclusive lock on the
+ * domain, which every writer of its edge holds; so that parent is the one
the edge is removed
+ * from.
+ *
+ * @param childIdent The identifier of the domain.
+ * @param check The check of the locked state; the change is rolled back if
it throws.
+ * @return The domain after the change.
+ * @throws IOException If the edge cannot be removed.
+ * @throws NoSuchEntityException If the domain does not exist.
+ */
+ @Monitored(
+ metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
+ baseMetricName = "removeDomainParentAndGet")
+ public DomainEntity removeDomainParentAndGet(
+ NameIdentifier childIdent, Consumer<LockedDomainState> check) throws
IOException {
+ DomainEntity[] changed = new DomainEntity[1];
+ try {
+ DomainPO childPO = getDomainPO(childIdent);
+ SessionUtils.doMultipleWithCommit(
+ () -> {
+ DomainPO lockedPO = lockMetalakeAndDomains(childIdent, childPO,
null).getLeft();
+ check.accept(
+ new LockedDomainState(
+ toEntityWithParent(lockedPO, childIdent.namespace()),
+ null,
+ Collections.emptyList()));
+ softDeleteDomainRel(childPO.domainId());
+ changed[0] = readDomain(childPO.domainId(),
childIdent.namespace());
+ });
+ } catch (RuntimeException e) {
+ ExceptionUtils.checkSQLException(e, Entity.EntityType.DOMAIN,
childIdent.toString());
+ throw e;
+ }
+ return changed[0];
+ }
+
+ /**
+ * Sets the status of a domain. Setting the current status writes nothing.
+ *
+ * <p>The domain is locked exclusively and its current parent shared, so
until the change commits
+ * the parent's status and the domain's parent cannot change, and no
sub-domain can be added to
+ * the domain or change its status: every writer of them locks one of these
rows in a conflicting
+ * mode. A sub-domain can still be detached or deleted, which only removes
it. The parent decides
+ * which rows to lock, but it is read before they are locked. Once the locks
are held, the parent
+ * is read again, and if it changed meanwhile, the change is rolled back and
retried with the new
+ * parent, a bounded number of times.
+ *
+ * <p>A retry must start a new transaction, so this method cannot run inside
one.
+ *
+ * @param identifier The identifier of the domain.
+ * @param status The new status.
+ * @param auditInfo The audit info of the change; the domain keeps its
creator and takes its last
+ * modifier.
+ * @param check The check of the locked state; the change is rolled back if
it throws.
+ * @return The domain after the change.
+ * @throws IOException If the domain cannot be stored.
+ * @throws NoSuchEntityException If the domain does not exist.
+ * @throws OptimisticLockException If the parent kept changing until the
retries ran out.
+ * @throws IllegalStateException If it is called inside a transaction.
+ */
+ @Monitored(
+ metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
+ baseMetricName = "updateDomainStatusAndGet")
+ public DomainEntity updateDomainStatusAndGet(
+ NameIdentifier identifier,
+ Domain.Status status,
+ AuditInfo auditInfo,
+ Consumer<LockedDomainState> check)
+ throws IOException {
+ Preconditions.checkState(
+ !SessionUtils.isInTransaction(),
+ "The status of domain %s cannot be changed inside another transaction",
+ identifier);
+ for (int attempt = 0; attempt < MAX_STATUS_UPDATE_ATTEMPTS; attempt++) {
+ DomainPO observedPO = getDomainPO(identifier);
+ DomainRelPO observedRelPO = selectDomainRel(observedPO.domainId());
+ Optional<DomainEntity> updated =
+ tryUpdateDomainStatusOnce(
+ identifier, status, auditInfo, check, observedPO, observedRelPO);
+ if (updated.isPresent()) {
+ return updated.get();
+ }
+ }
+ throw ExceptionUtils.concurrentModification(Entity.EntityType.DOMAIN,
identifier);
+ }
+
+ /**
+ * Sets the status of a domain once, locking the parent the caller observed.
+ *
+ * @return The domain after the change, or empty if the parent of the domain
is no longer the
+ * observed one; nothing is written then.
+ */
+ @VisibleForTesting
+ Optional<DomainEntity> tryUpdateDomainStatusOnce(
+ NameIdentifier identifier,
+ Domain.Status status,
+ AuditInfo auditInfo,
+ Consumer<LockedDomainState> check,
+ DomainPO observedPO,
+ @Nullable DomainRelPO observedRelPO)
+ throws IOException {
+ Long observedParentId = observedRelPO == null ? null :
observedRelPO.parentId();
+ Namespace namespace = identifier.namespace();
+ DomainEntity[] updated = new DomainEntity[1];
+ try {
+ SessionUtils.doMultipleWithCommit(
+ () -> {
+ Pair<DomainPO, DomainPO> locked =
+ lockMetalakeDomainAndParent(identifier, observedPO,
observedParentId);
+ DomainRelPO lockedRelPO = selectDomainRel(observedPO.domainId());
+ Long lockedParentId = lockedRelPO == null ? null :
lockedRelPO.parentId();
+ if (!Objects.equals(lockedParentId, observedParentId) || locked ==
null) {
+ throw new ParentChangedException();
+ }
+
+ DomainPO lockedPO = locked.getLeft();
+ DomainEntity domain = toEntityWithParent(lockedPO, namespace);
+ check.accept(
+ new LockedDomainState(
+ domain,
+ locked.getRight() == null
+ ? null
+ : toEntityWithParent(locked.getRight(), namespace),
+ listSubDomainsOf(lockedPO, namespace)));
+
+ if (domain.status() != status) {
+ DomainEntity newDomain =
+ domain.toBuilder()
+ .withStatus(status)
+ .withAuditInfo(
+ AuditInfo.builder()
+ .withCreator(domain.auditInfo().creator())
+ .withCreateTime(domain.auditInfo().createTime())
+ .withLastModifier(auditInfo.lastModifier())
+
.withLastModifiedTime(auditInfo.lastModifiedTime())
+ .build())
+ .build();
+ DomainPO newPO = DomainPO.updateDomainPO(lockedPO, newDomain);
+ OccWriteSupport.updateWithVersion(
+ () ->
+ SessionUtils.getWithoutCommit(
+ DomainMetaMapper.class,
+ mapper -> mapper.updateDomainMeta(newPO, lockedPO)),
+ () ->
+
ExceptionUtils.concurrentModification(Entity.EntityType.DOMAIN, identifier));
+ }
+ updated[0] = readDomain(observedPO.domainId(), namespace);
+ });
+ } catch (ParentChangedException e) {
+ return Optional.empty();
+ } catch (RuntimeException e) {
+ ExceptionUtils.checkSQLException(e, Entity.EntityType.DOMAIN,
identifier.toString());
+ throw e;
+ }
+ return Optional.of(updated[0]);
+ }
+
/**
* Inserts the parent edge of a top-level domain, which makes it a
sub-domain of {@code
- * parentIdent}.
+ * parentIdent}. It takes no check of the locked state; production code sets
a parent with {@link
+ * #setDomainParentAndGet}, so that the hierarchy rules are checked.
*
* @param childIdent The identifier of the domain.
* @param parentIdent The identifier of the parent domain, in the same
metalake.
@@ -155,13 +402,14 @@ public class DomainRelService {
@Monitored(
metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
baseMetricName = "insertDomainRel")
- public void insertDomainRel(
- NameIdentifier childIdent, NameIdentifier parentIdent, AuditInfo
auditInfo)
+ @VisibleForTesting
+ void insertDomainRel(NameIdentifier childIdent, NameIdentifier parentIdent,
AuditInfo auditInfo)
throws IOException {
checkSameMetalakeAndDistinct(childIdent, parentIdent);
try {
DomainPO childPO = getDomainPO(childIdent);
DomainPO parentPO = getDomainPO(parentIdent);
+ checkDistinctRows(childIdent, childPO, parentPO);
SessionUtils.doMultipleWithCommit(
() -> lockMetalakeAndDomains(childIdent, childPO, parentPO),
() -> {
@@ -183,7 +431,8 @@ public class DomainRelService {
/**
* Re-points the parent edge of a sub-domain to another parent. Re-pointing
it to its current
- * parent changes nothing.
+ * parent changes nothing. It takes no check of the locked state; production
code sets a parent
+ * with {@link #setDomainParentAndGet}, so that the hierarchy rules are
checked.
*
* @param childIdent The identifier of the sub-domain.
* @param newParentIdent The identifier of the new parent domain, in the
same metalake.
@@ -196,13 +445,15 @@ public class DomainRelService {
@Monitored(
metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
baseMetricName = "updateDomainRel")
- public void updateDomainRel(
+ @VisibleForTesting
+ void updateDomainRel(
NameIdentifier childIdent, NameIdentifier newParentIdent, AuditInfo
auditInfo)
throws IOException {
checkSameMetalakeAndDistinct(childIdent, newParentIdent);
try {
DomainPO childPO = getDomainPO(childIdent);
DomainPO newParentPO = getDomainPO(newParentIdent);
+ checkDistinctRows(childIdent, childPO, newParentPO);
SessionUtils.doMultipleWithCommit(
() -> lockMetalakeAndDomains(childIdent, childPO, newParentPO),
() -> {
@@ -229,28 +480,34 @@ public class DomainRelService {
}
/**
- * Removes the parent edge of a domain, which makes it top-level.
+ * Locks the parent of a new sub-domain shared, in the caller's transaction,
after the caller
+ * locked the metalake, and returns it with its own parent set.
*
- * @param childIdent The identifier of the domain.
- * @return True if the edge is removed, false if the domain is already
top-level.
- * @throws IOException If the edge cannot be removed.
- * @throws NoSuchEntityException If the domain does not exist.
+ * @param parentIdent The identifier of the parent domain.
+ * @return The locked parent domain.
+ * @throws NoSuchEntityException If the parent domain does not exist.
*/
- @Monitored(
- metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
- baseMetricName = "deleteDomainRel")
- public boolean deleteDomainRel(NameIdentifier childIdent) throws IOException
{
- int[] deleted = new int[1];
- try {
- DomainPO childPO = getDomainPO(childIdent);
- SessionUtils.doMultipleWithCommit(
- () -> lockMetalakeAndDomains(childIdent, childPO, null),
- () -> deleted[0] = softDeleteDomainRel(childPO.domainId()));
- } catch (RuntimeException e) {
- ExceptionUtils.checkSQLException(e, Entity.EntityType.DOMAIN,
childIdent.toString());
- throw e;
- }
- return deleted[0] > 0;
+ DomainEntity lockAndGetParentOfNewSubDomain(NameIdentifier parentIdent) {
+ DomainPO lockedParentPO = lockDomain(getDomainPO(parentIdent), false);
+ SessionUtils.clearSessionCache();
+ return toEntityWithParent(lockedParentPO, parentIdent.namespace());
+ }
+
+ /**
+ * Inserts the parent edge of a new sub-domain, in the caller's transaction.
The caller holds the
+ * shared lock on the parent from {@link
#lockAndGetParentOfNewSubDomain(NameIdentifier)}.
+ *
+ * @param metalakeId The metalake ID.
+ * @param parentId The ID of the parent domain.
+ * @param childId The ID of the new sub-domain.
+ * @param auditInfo The audit info of the edge.
+ */
+ void insertDomainRelOfNewSubDomain(
+ Long metalakeId, Long parentId, Long childId, AuditInfo auditInfo) {
+ DomainRelPO domainRelPO =
+ DomainRelPO.initializeDomainRelPO(metalakeId, parentId, childId,
auditInfo);
+ SessionUtils.doWithoutCommit(
+ DomainRelMapper.class, mapper -> mapper.insertDomainRel(domainRelPO));
}
/**
@@ -276,23 +533,82 @@ public class DomainRelService {
}
}
- private void lockMetalakeAndDomains(
+ // Returns the locked child and the locked parent, which is null without a
parent.
+ private Pair<DomainPO, DomainPO> lockMetalakeAndDomains(
NameIdentifier childIdent, DomainPO childPO, @Nullable DomainPO
parentPO) {
MetalakeMetaService.getInstance()
.lockMetalakeForChildWrite(childIdent.namespace().level(0),
childPO.metalakeId());
+ Pair<DomainPO, DomainPO> locked;
if (parentPO == null) {
- lockDomain(childPO, true);
+ locked = Pair.of(lockDomain(childPO, true), null);
} else if (childPO.domainId() < parentPO.domainId()) {
- lockDomain(childPO, true);
- lockDomain(parentPO, false);
+ DomainPO lockedChildPO = lockDomain(childPO, true);
+ locked = Pair.of(lockedChildPO, lockDomain(parentPO, false));
} else {
- lockDomain(parentPO, false);
- lockDomain(childPO, true);
+ DomainPO lockedParentPO = lockDomain(parentPO, false);
+ locked = Pair.of(lockDomain(childPO, true), lockedParentPO);
}
+ SessionUtils.clearSessionCache();
+ return locked;
}
- private void lockDomain(DomainPO observedDomainPO, boolean exclusive) {
- OccWriteSupport.lockParentForChildWrite(
+ // Like lockMetalakeAndDomains, but the parent is locked by ID only: whether
it is still the
+ // parent is decided by the edge read afterwards, not by its name. Returns
null if the parent row
+ // is gone, which means that the edge changed.
+ @Nullable
+ private Pair<DomainPO, DomainPO> lockMetalakeDomainAndParent(
+ NameIdentifier identifier, DomainPO observedPO, @Nullable Long parentId)
{
+ MetalakeMetaService.getInstance()
+ .lockMetalakeForChildWrite(identifier.namespace().level(0),
observedPO.metalakeId());
+ if (parentId == null) {
+ return Pair.of(lockDomain(observedPO, true), null);
+ }
+ DomainPO lockedPO;
+ DomainPO lockedParentPO;
+ if (observedPO.domainId() < parentId) {
+ lockedPO = lockDomain(observedPO, true);
+ lockedParentPO = selectDomainForShare(parentId);
+ } else {
+ lockedParentPO = selectDomainForShare(parentId);
+ lockedPO = lockDomain(observedPO, true);
+ }
+ SessionUtils.clearSessionCache();
+ return lockedParentPO == null ? null : Pair.of(lockedPO, lockedParentPO);
+ }
+
+ @Nullable
+ private DomainPO selectDomainForShare(Long domainId) {
+ return SessionUtils.getWithoutCommit(
+ DomainMetaMapper.class, mapper ->
mapper.selectDomainMetaByIdForShare(domainId));
+ }
+
+ // The parent name is read under the lock on the domain, which every writer
of its edge holds.
+ private DomainEntity toEntityWithParent(DomainPO domainPO, Namespace
namespace) {
+ return DomainPO.fromDomainPO(
+ domainPO, namespace,
getParentDomainName(domainPO.domainId()).orElse(null));
+ }
+
+ private List<DomainEntity> listSubDomainsOf(DomainPO domainPO, Namespace
namespace) {
+ return SessionUtils.getWithoutCommit(
+ DomainMetaMapper.class,
+ mapper ->
+ mapper.listDomainPOsByParentIdentifier(namespace.level(0),
domainPO.domainName()))
+ .stream()
+ .filter(po -> po.domainId() != null)
+ .map(po -> DomainPO.fromDomainPO(po, namespace))
+ .collect(Collectors.toList());
+ }
+
+ // Reads the domain with its backing tag name, in the caller's transaction.
+ private DomainEntity readDomain(Long domainId, Namespace namespace) {
+ DomainPO domainPO =
+ SessionUtils.getWithoutCommit(
+ DomainMetaMapper.class, mapper ->
mapper.selectDomainMetaById(domainId));
+ return toEntityWithParent(domainPO, namespace);
+ }
+
+ private DomainPO lockDomain(DomainPO observedDomainPO, boolean exclusive) {
+ return OccWriteSupport.lockParentForChildWrite(
observedDomainPO.domainName(),
Entity.EntityType.DOMAIN,
() ->
@@ -335,6 +651,16 @@ public class DomainRelService {
return domainPO;
}
+ // Two names can resolve to the same row, for example on MySQL, whose
collation ignores trailing
+ // spaces, so the rows are compared as well as the names.
+ private static void checkDistinctRows(
+ NameIdentifier childIdent, DomainPO childPO, DomainPO parentPO) {
+ Preconditions.checkArgument(
+ !Objects.equals(childPO.domainId(), parentPO.domainId()),
+ "The domain %s cannot be its own parent",
+ childIdent);
+ }
+
private static void checkSameMetalakeAndDistinct(
NameIdentifier childIdent, NameIdentifier parentIdent) {
Preconditions.checkArgument(
@@ -347,4 +673,13 @@ public class DomainRelService {
"The domain %s cannot be its own parent",
childIdent);
}
+
+ // Rolls back a status change whose parent changed between the read and the
locks.
+ private static final class ParentChangedException extends RuntimeException {
+ private static final long serialVersionUID = 1L;
+
+ private ParentChangedException() {
+ super(null, null, false, false);
+ }
+ }
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/service/LiveEndpointService.java
b/core/src/main/java/org/apache/gravitino/storage/relational/service/LiveEndpointService.java
index 2b7d1ad9a9..a0c5dc90e0 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/service/LiveEndpointService.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/service/LiveEndpointService.java
@@ -26,9 +26,7 @@ import org.apache.gravitino.exceptions.NoSuchEntityException;
import org.apache.gravitino.meta.NamespacedEntityId;
import org.apache.gravitino.storage.relational.RelationalEntityStoreIdResolver;
import org.apache.gravitino.storage.relational.mapper.LiveEndpointMapper;
-import org.apache.gravitino.storage.relational.session.SqlSessions;
import org.apache.gravitino.storage.relational.utils.SessionUtils;
-import org.apache.ibatis.session.SqlSession;
/**
* Fences metadata-object endpoints while a relation write is in progress.
@@ -82,12 +80,7 @@ public final class LiveEndpointService {
// An earlier name lookup in this transaction may be in MyBatis's
first-level cache. The
// locking reads above must be followed by a database read of the current
name-to-ID chain.
- SqlSession session = SqlSessions.getSqlSession();
- try {
- session.clearCache();
- } finally {
- SqlSessions.closeSqlSession();
- }
+ SessionUtils.clearSessionCache();
NamespacedEntityId current =
new RelationalEntityStoreIdResolver().getEntityIds(identifier, type);
if (!Arrays.equals(current.fullIds(), observed.fullIds())) {
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/utils/SessionUtils.java
b/core/src/main/java/org/apache/gravitino/storage/relational/utils/SessionUtils.java
index 54539355bb..63e96d2567 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/utils/SessionUtils.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/utils/SessionUtils.java
@@ -23,6 +23,7 @@ import java.util.Arrays;
import java.util.function.Consumer;
import java.util.function.Function;
import org.apache.gravitino.storage.relational.session.SqlSessions;
+import org.apache.ibatis.session.SqlSession;
/**
* This class provides utility methods to perform database operations with
MyBatis mappers in the
@@ -107,6 +108,20 @@ public class SessionUtils {
}
}
+ /**
+ * Clears MyBatis's first-level cache of the current session. A read earlier
in the session, for
+ * example by a caller's outer transaction, may be cached; a read that must
see the database as it
+ * is now, such as one that follows a locking read, calls this first.
+ */
+ public static void clearSessionCache() {
+ SqlSession session = SqlSessions.getSqlSession();
+ try {
+ session.clearCache();
+ } finally {
+ SqlSessions.closeSqlSession();
+ }
+ }
+
public static void beginTransaction() {
SqlSessions.getSqlSession();
}
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
index 152cfa9e44..006acc1fbd 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/TestRelationalEntityStoreLockWaitTimeout.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/TestRelationalEntityStoreLockWaitTimeout.java
@@ -18,6 +18,7 @@
*/
package org.apache.gravitino.storage.relational;
+import com.google.common.collect.ImmutableMap;
import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
@@ -43,15 +44,18 @@ 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.domain.Domain;
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.DomainEntity;
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.NameIdentifierUtil;
import org.apache.gravitino.utils.NamespaceUtil;
import org.apache.ibatis.exceptions.PersistenceException;
import org.junit.jupiter.api.AfterEach;
@@ -78,6 +82,8 @@ public class TestRelationalEntityStoreLockWaitTimeout {
private static final NameIdentifier TABLE_IDENT =
NameIdentifier.of(METALAKE, CATALOG, SCHEMA, "t1");
+ private static final NameIdentifier DOMAIN_IDENT =
NameIdentifierUtil.ofDomain(METALAKE, "d1");
+
private RelationalEntityStore store;
private String dbPath;
private Object previousConfig;
@@ -344,7 +350,35 @@ public class TestRelationalEntityStoreLockWaitTimeout {
Pair.of(
"batchDelete",
() -> store.batchDelete(List.of(Pair.of(TABLE_IDENT,
Entity.EntityType.TABLE)), false)),
- Pair.of("batchPut", () -> store.batchPut(List.of(table), true)));
+ Pair.of("batchPut", () -> store.batchPut(List.of(table), true)),
+ Pair.of("insertDomain", () -> store.insertDomain(domain(), state ->
{})),
+ Pair.of(
+ "setDomainParentAndGet",
+ () ->
+ store.setDomainParentAndGet(
+ DOMAIN_IDENT,
+ NameIdentifierUtil.ofDomain(METALAKE, "parent"),
+ AUDIT_INFO,
+ state -> {})),
+ Pair.of(
+ "removeDomainParentAndGet",
+ () -> store.removeDomainParentAndGet(DOMAIN_IDENT, state -> {})),
+ Pair.of(
+ "updateDomainStatusAndGet",
+ () ->
+ store.updateDomainStatusAndGet(
+ DOMAIN_IDENT, Domain.Status.PUBLISHED, AUDIT_INFO, state
-> {})));
+ }
+
+ private static DomainEntity domain() {
+ return DomainEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName(DOMAIN_IDENT.name())
+ .withNamespace(DOMAIN_IDENT.namespace())
+ .withProperties(ImmutableMap.of())
+ .withStatus(Domain.Status.DRAFT)
+ .withAuditInfo(AUDIT_INFO)
+ .build();
}
private List<TagEntity> associateTags(NameIdentifier[] tags) throws
IOException {
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestDomainMetaService.java
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestDomainMetaService.java
index b3c83391ec..62169aea59 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestDomainMetaService.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestDomainMetaService.java
@@ -149,13 +149,31 @@ public class TestDomainMetaService extends
TestJDBCBackend {
DomainEntity finance = newDomain(METALAKE_NAME, "Finance", "comment");
domainMetaService.insertDomain(finance, false);
- DomainEntity replacement =
-
finance.toBuilder().withComment("replaced").withStatus(Domain.Status.PUBLISHED).build();
+ DomainEntity replacement =
finance.toBuilder().withComment("replaced").build();
domainMetaService.insertDomain(replacement, true);
Assertions.assertEquals(
replacement,
domainMetaService.getDomainByIdentifier(finance.nameIdentifier()));
Assertions.assertEquals(2L, selectDomainPO(finance.id()).currentVersion());
+ // A replacement keeps the status, which changes only under the status
locks.
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ domainMetaService.insertDomain(
+
replacement.toBuilder().withStatus(Domain.Status.PUBLISHED).build(), true));
+ Assertions.assertEquals(2L, selectDomainPO(finance.id()).currentVersion());
+
+ // A sub-domain is inserted only with its parent locked and checked.
+ DomainEntity trading = newDomain(METALAKE_NAME, "Trading", null);
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ domainMetaService.insertDomain(
+ trading.toBuilder().withParent("Finance").build(), false));
+ Assertions.assertThrows(
+ NoSuchEntityException.class,
+ () ->
domainMetaService.getDomainByIdentifier(trading.nameIdentifier()));
+
// An overwrite cannot reuse the ID of a deleted domain.
Assertions.assertTrue(domainMetaService.deleteDomain(finance.nameIdentifier()));
Assertions.assertThrows(
@@ -226,14 +244,14 @@ public class TestDomainMetaService extends
TestJDBCBackend {
Assertions.assertTrue(batch.contains(finance));
Assertions.assertTrue(batch.contains(markets));
- DomainEntity published =
+ DomainEntity updated =
backend.<DomainEntity>update(
finance.nameIdentifier(),
Entity.EntityType.DOMAIN,
- entity ->
entity.toBuilder().withStatus(Domain.Status.PUBLISHED).build());
- Assertions.assertEquals(Domain.Status.PUBLISHED, published.status());
+ entity -> entity.toBuilder().withComment("updated").build());
+ Assertions.assertEquals("updated", updated.comment());
Assertions.assertEquals(
- published, backend.get(finance.nameIdentifier(),
Entity.EntityType.DOMAIN));
+ updated, backend.get(finance.nameIdentifier(),
Entity.EntityType.DOMAIN));
Assertions.assertTrue(
backend.delete(finance.nameIdentifier(), Entity.EntityType.DOMAIN,
false));
@@ -305,7 +323,6 @@ public class TestDomainMetaService extends TestJDBCBackend {
.withName("Finance/Tax")
.withComment(null)
.withProperties(ImmutableMap.of("k2", "v2"))
- .withStatus(Domain.Status.PUBLISHED)
.withAuditInfo(updatedAudit)
.build());
@@ -319,10 +336,20 @@ public class TestDomainMetaService extends
TestJDBCBackend {
Assertions.assertEquals(finance.id(), loaded.id());
Assertions.assertNull(loaded.comment());
Assertions.assertEquals(ImmutableMap.of("k2", "v2"), loaded.properties());
- Assertions.assertEquals(Domain.Status.PUBLISHED, loaded.status());
Assertions.assertEquals("modifier", loaded.auditInfo().lastModifier());
Assertions.assertEquals(2L, selectDomainPO(finance.id()).currentVersion());
+ // The status is changed only with the status locks, not by an update.
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ domainMetaService.updateDomain(
+ loaded.nameIdentifier(),
+ entity ->
+ ((DomainEntity) entity)
+
.toBuilder().withStatus(Domain.Status.PUBLISHED).build()));
+ Assertions.assertEquals(2L, selectDomainPO(finance.id()).currentVersion());
+
// Renaming to a name in use fails.
Assertions.assertThrows(
EntityAlreadyExistsException.class,
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestDomainRelService.java
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestDomainRelService.java
index dd53110914..3e0bcca213 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestDomainRelService.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestDomainRelService.java
@@ -18,6 +18,9 @@
*/
package org.apache.gravitino.storage.relational.service;
+import static org.mockito.ArgumentMatchers.any;
+
+import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;
import java.io.IOException;
@@ -26,9 +29,12 @@ import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.time.Instant;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -37,19 +43,24 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
+import java.util.stream.Collectors;
import org.apache.gravitino.Entity;
import org.apache.gravitino.EntityAlreadyExistsException;
import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.SupportsDomainOperations.LockedDomainState;
import org.apache.gravitino.domain.Domain;
import org.apache.gravitino.exceptions.NoSuchEntityException;
import org.apache.gravitino.exceptions.NonEmptyEntityException;
+import org.apache.gravitino.exceptions.OptimisticLockException;
import org.apache.gravitino.json.JsonUtils;
import org.apache.gravitino.meta.AuditInfo;
import org.apache.gravitino.meta.BaseMetalake;
import org.apache.gravitino.meta.DomainEntity;
import org.apache.gravitino.storage.RandomIdGenerator;
import org.apache.gravitino.storage.relational.TestJDBCBackend;
+import org.apache.gravitino.storage.relational.mapper.DomainMetaMapper;
import org.apache.gravitino.storage.relational.mapper.DomainRelMapper;
+import org.apache.gravitino.storage.relational.po.DomainPO;
import org.apache.gravitino.storage.relational.po.DomainRelPO;
import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper;
import org.apache.gravitino.storage.relational.utils.SessionUtils;
@@ -67,6 +78,7 @@ import org.apache.ibatis.session.RowBounds;
import org.apache.ibatis.session.SqlSession;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.TestTemplate;
+import org.mockito.Mockito;
public class TestDomainRelService extends TestJDBCBackend {
@@ -272,31 +284,6 @@ public class TestDomainRelService extends TestJDBCBackend {
AUDIT_INFO));
}
- @TestTemplate
- public void testDeleteDomainRel() throws IOException {
- createAndInsertMakeLake(METALAKE_NAME);
- DomainEntity finance = insertDomain(METALAKE_NAME, "Finance");
- DomainEntity trading = insertDomain(METALAKE_NAME, "Trading");
-
-
Assertions.assertFalse(domainRelService.deleteDomainRel(trading.nameIdentifier()));
- domainRelService.insertDomainRel(
- trading.nameIdentifier(), finance.nameIdentifier(), AUDIT_INFO);
-
Assertions.assertTrue(domainRelService.deleteDomainRel(trading.nameIdentifier()));
-
Assertions.assertFalse(domainRelService.deleteDomainRel(trading.nameIdentifier()));
-
Assertions.assertFalse(domainRelService.getParentDomainName(trading.id()).isPresent());
-
Assertions.assertTrue(domainRelService.listSubDomains(finance.nameIdentifier()).isEmpty());
- Assertions.assertThrows(
- NoSuchEntityException.class,
- () ->
- domainRelService.deleteDomainRel(
- NameIdentifierUtil.ofDomain(METALAKE_NAME, "missing")));
-
- // A detached domain can be attached again.
- domainRelService.insertDomainRel(
- trading.nameIdentifier(), finance.nameIdentifier(), AUDIT_INFO);
- Assertions.assertEquals("Finance",
domainRelService.getParentDomainName(trading.id()).get());
- }
-
@TestTemplate
public void testDeleteDomainWithSubDomains() throws IOException {
createAndInsertMakeLake(METALAKE_NAME);
@@ -515,7 +502,11 @@ public class TestDomainRelService extends TestJDBCBackend {
try {
for (int i = 0; i < 3; i++) {
insertDomainRel(child, parent);
-
Assertions.assertTrue(domainRelService.deleteDomainRel(child.nameIdentifier()));
+ Assertions.assertFalse(
+ domainRelService
+ .removeDomainParentAndGet(child.nameIdentifier(), state
-> {})
+ .parent()
+ .isPresent());
}
insertDomainRel(child, parent);
Assertions.assertTrue(domainMetaService.deleteDomain(child.nameIdentifier()));
@@ -747,7 +738,11 @@ public class TestDomainRelService extends TestJDBCBackend {
() -> {
insertDomainRel(child, parent);
try {
-
Assertions.assertTrue(domainRelService.deleteDomainRel(child.nameIdentifier()));
+ Assertions.assertFalse(
+ domainRelService
+ .removeDomainParentAndGet(child.nameIdentifier(), state ->
{})
+ .parent()
+ .isPresent());
} catch (IOException e) {
throw new RuntimeException(e);
}
@@ -811,6 +806,629 @@ public class TestDomainRelService extends TestJDBCBackend
{
}
}
+ @TestTemplate
+ public void testInsertDomainWithCheck() throws IOException {
+ createAndInsertMakeLake(METALAKE_NAME);
+ DomainEntity finance = newDomain(METALAKE_NAME, "Finance", null);
+ List<LockedDomainState> states = new ArrayList<>();
+
+ domainMetaService.insertDomain(finance, states::add);
+ Assertions.assertEquals(
+ finance,
domainMetaService.getDomainByIdentifier(finance.nameIdentifier()));
+ Assertions.assertEquals(1, states.size());
+ Assertions.assertEquals(finance, states.get(0).domain());
+ Assertions.assertFalse(states.get(0).parent().isPresent());
+ Assertions.assertTrue(states.get(0).subDomains().isEmpty());
+
+ // A domain with a parent is inserted together with its edge, and the
check sees the parent.
+ DomainEntity trading = newDomain(METALAKE_NAME, "Trading", "Finance");
+ states.clear();
+ domainMetaService.insertDomain(trading, states::add);
+ Assertions.assertEquals(
+ trading,
domainMetaService.getDomainByIdentifier(trading.nameIdentifier()));
+ Assertions.assertEquals(finance.id(),
selectDomainRel(trading.id()).parentId());
+ LockedDomainState state = states.get(0);
+ Assertions.assertEquals(finance.id(), state.parent().get().id());
+ Assertions.assertFalse(state.parent().get().parent().isPresent());
+
+ // The check sees that the parent is itself a sub-domain.
+ DomainEntity desk = newDomain(METALAKE_NAME, "Desk", "Trading");
+ states.clear();
+ domainMetaService.insertDomain(desk, states::add);
+ Assertions.assertEquals("Finance",
states.get(0).parent().get().parent().get());
+
+ // A failing check inserts neither the domain nor its edge.
+ DomainEntity rejected = newDomain(METALAKE_NAME, "Rejected", "Finance");
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ domainMetaService.insertDomain(
+ rejected,
+ s -> {
+ throw new IllegalArgumentException("rejected");
+ }));
+ Assertions.assertThrows(
+ NoSuchEntityException.class,
+ () ->
domainMetaService.getDomainByIdentifier(rejected.nameIdentifier()));
+ Assertions.assertEquals(0, countDomainRelRows(rejected.id()));
+
+ DomainEntity orphan = newDomain(METALAKE_NAME, "Orphan", "missing");
+ Assertions.assertThrows(
+ NoSuchEntityException.class, () ->
domainMetaService.insertDomain(orphan, s -> {}));
+ Assertions.assertThrows(
+ NoSuchEntityException.class,
+ () ->
domainMetaService.getDomainByIdentifier(orphan.nameIdentifier()));
+
+ Assertions.assertThrows(
+ EntityAlreadyExistsException.class,
+ () -> domainMetaService.insertDomain(newDomain(METALAKE_NAME,
"Finance", null), s -> {}));
+ Assertions.assertThrows(
+ NoSuchEntityException.class,
+ () -> domainMetaService.insertDomain(newDomain("missing_metalake",
"X", null), s -> {}));
+ }
+
+ @TestTemplate
+ public void testInsertDomainWithCheckRejectsOuterTransaction() throws
IOException {
+ createAndInsertMakeLake(METALAKE_NAME);
+ DomainEntity finance = insertDomain(METALAKE_NAME, "Finance");
+ for (String parent : new String[] {null, finance.name()}) {
+ DomainEntity rejected = newDomain(METALAKE_NAME, "Rejected", parent);
+ DomainEntity accepted = newDomain(METALAKE_NAME, "Accepted" +
rejected.id(), null);
+ SessionUtils.doMultipleWithCommit(
+ () -> {
+ Assertions.assertThrows(
+ IllegalStateException.class,
+ () ->
+ domainMetaService.insertDomain(
+ rejected,
+ state -> {
+ throw new IllegalArgumentException("rejected");
+ }));
+ // Catching the rejection must not leave a domain for the outer
commit to persist.
+ // A separate write confirms that the outer transaction still
commits normally.
+ try {
+ domainMetaService.insertDomain(accepted, false);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ });
+ Assertions.assertEquals(
+ accepted,
domainMetaService.getDomainByIdentifier(accepted.nameIdentifier()));
+ Assertions.assertThrows(
+ NoSuchEntityException.class,
+ () ->
domainMetaService.getDomainByIdentifier(rejected.nameIdentifier()));
+ Assertions.assertEquals(0, countDomainRelRows(rejected.id()));
+ }
+ }
+
+ @TestTemplate
+ public void testSetDomainParent() throws IOException {
+ createAndInsertMakeLake(METALAKE_NAME);
+ String anotherMetalakeName = METALAKE_NAME + "_another";
+ createAndInsertMakeLake(anotherMetalakeName);
+ DomainEntity finance = insertDomain(METALAKE_NAME, "Finance");
+ DomainEntity markets = insertDomain(METALAKE_NAME, "Markets");
+ DomainEntity trading = insertDomain(METALAKE_NAME, "Trading");
+ DomainEntity desk = insertDomain(METALAKE_NAME, "Desk");
+ DomainEntity elsewhere = insertDomain(anotherMetalakeName, "Elsewhere");
+ insertDomainRel(desk, trading);
+ List<LockedDomainState> states = new ArrayList<>();
+
+ // Attaching a top-level domain inserts its edge; the check sees the
locked domains.
+ DomainEntity attached =
+ domainRelService.setDomainParentAndGet(
+ trading.nameIdentifier(), finance.nameIdentifier(), AUDIT_INFO,
states::add);
+ Assertions.assertEquals(trading.toBuilder().withParent("Finance").build(),
attached);
+ LockedDomainState state = states.get(0);
+ Assertions.assertEquals(trading.id(), state.domain().id());
+ Assertions.assertFalse(state.domain().parent().isPresent());
+ Assertions.assertEquals(finance.id(), state.parent().get().id());
+ Assertions.assertFalse(state.parent().get().parent().isPresent());
+ Assertions.assertEquals(
+ Lists.newArrayList(desk.id()),
+
state.subDomains().stream().map(DomainEntity::id).collect(Collectors.toList()));
+ Assertions.assertEquals(1L,
selectDomainRel(trading.id()).currentVersion());
+
+ // Moving it re-points the edge; the check sees the current parent.
+ states.clear();
+ Instant modifiedTime = Instant.now();
+ AuditInfo modifiedAuditInfo =
+ AuditInfo.builder()
+ .withCreator("mover")
+ .withCreateTime(modifiedTime)
+ .withLastModifier("mover")
+ .withLastModifiedTime(modifiedTime)
+ .build();
+ DomainEntity moved =
+ domainRelService.setDomainParentAndGet(
+ trading.nameIdentifier(), markets.nameIdentifier(),
modifiedAuditInfo, states::add);
+ Assertions.assertEquals("Markets", moved.parent().get());
+ Assertions.assertEquals("Finance", states.get(0).domain().parent().get());
+ Assertions.assertEquals(markets.id(), states.get(0).parent().get().id());
+ DomainRelPO domainRelPO = selectDomainRel(trading.id());
+ Assertions.assertEquals(2L, domainRelPO.currentVersion());
+ AuditInfo storedAuditInfo =
+ JsonUtils.anyFieldMapper().readValue(domainRelPO.auditInfo(),
AuditInfo.class);
+ Assertions.assertEquals(AUDIT_INFO.creator(), storedAuditInfo.creator());
+ Assertions.assertEquals("mover", storedAuditInfo.lastModifier());
+
+ // The check also sees that the new parent is itself a sub-domain.
+ states.clear();
+ domainRelService.setDomainParentAndGet(
+ finance.nameIdentifier(), desk.nameIdentifier(), AUDIT_INFO,
states::add);
+ Assertions.assertEquals("Trading",
states.get(0).parent().get().parent().get());
+ domainRelService.removeDomainParentAndGet(finance.nameIdentifier(), s ->
{});
+
+ // Setting the current parent changes nothing.
+ domainRelService.setDomainParentAndGet(
+ trading.nameIdentifier(), markets.nameIdentifier(), modifiedAuditInfo,
s -> {});
+ Assertions.assertEquals(2L,
selectDomainRel(trading.id()).currentVersion());
+
+ // A failing check changes nothing.
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ domainRelService.setDomainParentAndGet(
+ trading.nameIdentifier(),
+ finance.nameIdentifier(),
+ AUDIT_INFO,
+ s -> {
+ throw new IllegalArgumentException("rejected");
+ }));
+ Assertions.assertEquals("Markets",
domainRelService.getParentDomainName(trading.id()).get());
+
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ domainRelService.setDomainParentAndGet(
+ trading.nameIdentifier(), trading.nameIdentifier(),
AUDIT_INFO, s -> {}));
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ domainRelService.setDomainParentAndGet(
+ elsewhere.nameIdentifier(), finance.nameIdentifier(),
AUDIT_INFO, s -> {}));
+ // On MySQL, whose collation ignores trailing spaces, both names resolve
to the same row, which
+ // is refused as its own parent; elsewhere the second name does not exist.
No edge is written.
+ RuntimeException selfParent =
+ Assertions.assertThrows(
+ RuntimeException.class,
+ () ->
+ domainRelService.setDomainParentAndGet(
+ finance.nameIdentifier(),
+ NameIdentifierUtil.ofDomain(METALAKE_NAME, "Finance "),
+ AUDIT_INFO,
+ s -> {}));
+ Assertions.assertTrue(
+ selfParent instanceof IllegalArgumentException
+ || selfParent instanceof NoSuchEntityException,
+ String.valueOf(selfParent));
+
Assertions.assertFalse(domainRelService.getParentDomainName(finance.id()).isPresent());
+
+ NameIdentifier missing = NameIdentifierUtil.ofDomain(METALAKE_NAME,
"missing");
+ Assertions.assertThrows(
+ NoSuchEntityException.class,
+ () ->
+ domainRelService.setDomainParentAndGet(
+ missing, finance.nameIdentifier(), AUDIT_INFO, s -> {}));
+ Assertions.assertThrows(
+ NoSuchEntityException.class,
+ () ->
+ domainRelService.setDomainParentAndGet(
+ trading.nameIdentifier(), missing, AUDIT_INFO, s -> {}));
+ }
+
+ @TestTemplate
+ public void testRemoveDomainParent() throws IOException {
+ createAndInsertMakeLake(METALAKE_NAME);
+ DomainEntity finance = insertDomain(METALAKE_NAME, "Finance");
+ DomainEntity trading = insertDomain(METALAKE_NAME, "Trading");
+ insertDomainRel(trading, finance);
+ List<LockedDomainState> states = new ArrayList<>();
+
+ // A failing check keeps the edge.
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ domainRelService.removeDomainParentAndGet(
+ trading.nameIdentifier(),
+ s -> {
+ throw new IllegalArgumentException("rejected");
+ }));
+ Assertions.assertEquals("Finance",
domainRelService.getParentDomainName(trading.id()).get());
+
+ // The check sees the parent the edge is removed from.
+ Assertions.assertEquals(
+ trading,
domainRelService.removeDomainParentAndGet(trading.nameIdentifier(),
states::add));
+ Assertions.assertEquals("Finance", states.get(0).domain().parent().get());
+ Assertions.assertFalse(states.get(0).parent().isPresent());
+ Assertions.assertTrue(states.get(0).subDomains().isEmpty());
+
Assertions.assertFalse(domainRelService.getParentDomainName(trading.id()).isPresent());
+
Assertions.assertTrue(domainRelService.listSubDomains(finance.nameIdentifier()).isEmpty());
+
+ // Removing the parent of a top-level domain changes nothing.
+ states.clear();
+ Assertions.assertEquals(
+ trading,
domainRelService.removeDomainParentAndGet(trading.nameIdentifier(),
states::add));
+ Assertions.assertFalse(states.get(0).domain().parent().isPresent());
+ Assertions.assertThrows(
+ NoSuchEntityException.class,
+ () ->
+ domainRelService.removeDomainParentAndGet(
+ NameIdentifierUtil.ofDomain(METALAKE_NAME, "missing"), s ->
{}));
+
+ // A detached domain can be attached again.
+ insertDomainRel(trading, finance);
+ Assertions.assertEquals("Finance",
domainRelService.getParentDomainName(trading.id()).get());
+ }
+
+ @TestTemplate
+ public void testUpdateDomainStatus() throws IOException {
+ createAndInsertMakeLake(METALAKE_NAME);
+ DomainEntity finance = insertDomain(METALAKE_NAME, "Finance");
+ DomainEntity trading = insertDomain(METALAKE_NAME, "Trading");
+ insertDomainRel(trading, finance);
+ List<LockedDomainState> states = new ArrayList<>();
+ Instant modifiedTime = Instant.now();
+ AuditInfo modifiedAuditInfo =
+ AuditInfo.builder()
+ .withLastModifier("publisher")
+ .withLastModifiedTime(modifiedTime)
+ .build();
+
+ // A top-level domain has no parent; the check sees its sub-domains.
+ DomainEntity published =
+ domainRelService.updateDomainStatusAndGet(
+ finance.nameIdentifier(), Domain.Status.PUBLISHED,
modifiedAuditInfo, states::add);
+ Assertions.assertEquals(Domain.Status.PUBLISHED, published.status());
+ Assertions.assertEquals(AUDIT_INFO.creator(),
published.auditInfo().creator());
+ Assertions.assertEquals("publisher", published.auditInfo().lastModifier());
+ Assertions.assertEquals(modifiedTime,
published.auditInfo().lastModifiedTime());
+ Assertions.assertEquals(
+ published,
domainMetaService.getDomainByIdentifier(finance.nameIdentifier()));
+ Assertions.assertEquals(Domain.Status.DRAFT,
states.get(0).domain().status());
+ Assertions.assertFalse(states.get(0).parent().isPresent());
+ Assertions.assertEquals(trading.id(),
states.get(0).subDomains().get(0).id());
+ Assertions.assertEquals(Domain.Status.DRAFT,
states.get(0).subDomains().get(0).status());
+
+ // A sub-domain's check sees its parent.
+ states.clear();
+ DomainEntity publishedTrading =
+ domainRelService.updateDomainStatusAndGet(
+ trading.nameIdentifier(), Domain.Status.PUBLISHED,
modifiedAuditInfo, states::add);
+ Assertions.assertEquals("Finance", publishedTrading.parent().get());
+ Assertions.assertEquals("Finance", states.get(0).domain().parent().get());
+ Assertions.assertEquals(Domain.Status.PUBLISHED,
states.get(0).parent().get().status());
+
+ // Setting the current status writes nothing.
+ long version = selectDomainVersion(trading.id());
+ domainRelService.updateDomainStatusAndGet(
+ trading.nameIdentifier(), Domain.Status.PUBLISHED, modifiedAuditInfo,
s -> {});
+ Assertions.assertEquals(version, selectDomainVersion(trading.id()));
+
+ // A failing check changes nothing.
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ domainRelService.updateDomainStatusAndGet(
+ finance.nameIdentifier(),
+ Domain.Status.DRAFT,
+ modifiedAuditInfo,
+ s -> {
+ throw new IllegalArgumentException("rejected");
+ }));
+ Assertions.assertEquals(
+ Domain.Status.PUBLISHED,
+
domainMetaService.getDomainByIdentifier(finance.nameIdentifier()).status());
+
+ Assertions.assertThrows(
+ NoSuchEntityException.class,
+ () ->
+ domainRelService.updateDomainStatusAndGet(
+ NameIdentifierUtil.ofDomain(METALAKE_NAME, "missing"),
+ Domain.Status.PUBLISHED,
+ modifiedAuditInfo,
+ s -> {}));
+ }
+
+ @TestTemplate
+ public void testUpdateDomainStatusRetriesWhenParentChanges() throws
IOException {
+ createAndInsertMakeLake(METALAKE_NAME);
+ DomainEntity finance = insertDomain(METALAKE_NAME, "Finance");
+ DomainEntity markets = insertDomain(METALAKE_NAME, "Markets");
+ DomainEntity trading = insertDomain(METALAKE_NAME, "Trading");
+ insertDomainRel(trading, finance);
+ NameIdentifier tradingIdent = trading.nameIdentifier();
+ DomainPO observedPO = selectDomainPO(trading.id());
+ DomainRelPO observedRelPO = selectDomainRel(trading.id());
+
+ // The domain is moved after its parent was read: the attempt writes
nothing.
+ moveDomainRel(trading, markets);
+ Assertions.assertFalse(
+ domainRelService
+ .tryUpdateDomainStatusOnce(
+ tradingIdent,
+ Domain.Status.PUBLISHED,
+ AUDIT_INFO,
+ s -> Assertions.fail("the check must not run"),
+ observedPO,
+ observedRelPO)
+ .isPresent());
+ Assertions.assertEquals(
+ Domain.Status.DRAFT,
domainMetaService.getDomainByIdentifier(tradingIdent).status());
+
+ // The domain is detached after its parent was read.
+ domainRelService.removeDomainParentAndGet(tradingIdent, s -> {});
+ Assertions.assertFalse(
+ domainRelService
+ .tryUpdateDomainStatusOnce(
+ tradingIdent,
+ Domain.Status.PUBLISHED,
+ AUDIT_INFO,
+ s -> {},
+ observedPO,
+ observedRelPO)
+ .isPresent());
+
+ // The domain is attached after it was read as top-level.
+ insertDomainRel(trading, markets);
+ Assertions.assertFalse(
+ domainRelService
+ .tryUpdateDomainStatusOnce(
+ tradingIdent, Domain.Status.PUBLISHED, AUDIT_INFO, s -> {},
observedPO, null)
+ .isPresent());
+
+ // A fresh read locks the current parent, which the check sees.
+ List<LockedDomainState> states = new ArrayList<>();
+ DomainEntity published =
+ domainRelService.updateDomainStatusAndGet(
+ tradingIdent, Domain.Status.PUBLISHED, AUDIT_INFO, states::add);
+ Assertions.assertEquals(Domain.Status.PUBLISHED, published.status());
+ Assertions.assertEquals(markets.id(), states.get(0).parent().get().id());
+ }
+
+ @TestTemplate
+ public void testUpdateDomainStatusWhenObservedParentIsGone() throws
IOException {
+ createAndInsertMakeLake(METALAKE_NAME);
+ DomainEntity finance = insertDomain(METALAKE_NAME, "Finance");
+ DomainEntity trading = insertDomain(METALAKE_NAME, "Trading");
+ insertDomainRel(trading, finance);
+ DomainPO observedPO = selectDomainPO(trading.id());
+ DomainRelPO observedRelPO = selectDomainRel(trading.id());
+
+ // The domain is detached and its old parent deleted after the parent was
read.
+ domainRelService.removeDomainParentAndGet(trading.nameIdentifier(), s ->
{});
+
Assertions.assertTrue(domainMetaService.deleteDomain(finance.nameIdentifier()));
+ Assertions.assertFalse(
+ domainRelService
+ .tryUpdateDomainStatusOnce(
+ trading.nameIdentifier(),
+ Domain.Status.PUBLISHED,
+ AUDIT_INFO,
+ s -> Assertions.fail("the check must not run"),
+ observedPO,
+ observedRelPO)
+ .isPresent());
+ Assertions.assertEquals(
+ Domain.Status.DRAFT,
+
domainMetaService.getDomainByIdentifier(trading.nameIdentifier()).status());
+ }
+
+ @TestTemplate
+ public void testUpdateDomainStatusGivesUpAfterRetries() throws IOException {
+ createAndInsertMakeLake(METALAKE_NAME);
+ DomainEntity finance = insertDomain(METALAKE_NAME, "Finance");
+ DomainRelService service = Mockito.spy(domainRelService);
+ // Every attempt finds that the parent changed.
+ Mockito.doReturn(Optional.empty())
+ .when(service)
+ .tryUpdateDomainStatusOnce(any(), any(), any(), any(), any(), any());
+
+ Assertions.assertThrows(
+ OptimisticLockException.class,
+ () ->
+ service.updateDomainStatusAndGet(
+ finance.nameIdentifier(), Domain.Status.PUBLISHED, AUDIT_INFO,
s -> {}));
+ Mockito.verify(service, Mockito.times(3))
+ .tryUpdateDomainStatusOnce(any(), any(), any(), any(), any(), any());
+ }
+
+ @TestTemplate
+ public void testUpdateDomainStatusRejectsOuterTransaction() throws
IOException {
+ createAndInsertMakeLake(METALAKE_NAME);
+ DomainEntity finance = insertDomain(METALAKE_NAME, "Finance");
+
+ Assertions.assertThrows(
+ IllegalStateException.class,
+ () ->
+ SessionUtils.doMultipleWithCommit(
+ () -> {
+ try {
+ domainRelService.updateDomainStatusAndGet(
+ finance.nameIdentifier(), Domain.Status.PUBLISHED,
AUDIT_INFO, s -> {});
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }));
+ Assertions.assertEquals(
+ Domain.Status.DRAFT,
+
domainMetaService.getDomainByIdentifier(finance.nameIdentifier()).status());
+ }
+
+ @TestTemplate
+ public void testLockedOperationsWaitForDomainRows() throws Exception {
+ createAndInsertMakeLake(METALAKE_NAME);
+ DomainEntity finance = insertDomain(METALAKE_NAME, "Finance");
+ DomainEntity markets = insertDomain(METALAKE_NAME, "Markets");
+ DomainEntity trading = insertDomain(METALAKE_NAME, "Trading");
+ DomainEntity desk = insertDomain(METALAKE_NAME, "Desk");
+ insertDomainRel(trading, finance);
+
+ // Only the domain row is held, not the metalake row, so each wait is on
that domain row.
+ assertWaitsForDomainRow(
+ finance.id(),
+ () ->
+ domainRelService.updateDomainStatusAndGet(
+ trading.nameIdentifier(), Domain.Status.PUBLISHED, AUDIT_INFO,
s -> {}));
+ assertWaitsForDomainRow(
+ finance.id(),
+ () -> {
+ domainMetaService.insertDomain(newDomain(METALAKE_NAME, "Tax",
"Finance"), s -> {});
+ return null;
+ });
+ assertWaitsForDomainRow(
+ markets.id(),
+ () ->
+ domainRelService.setDomainParentAndGet(
+ desk.nameIdentifier(), markets.nameIdentifier(), AUDIT_INFO, s
-> {}));
+ assertWaitsForDomainRow(
+ desk.id(),
+ () ->
+ domainRelService.setDomainParentAndGet(
+ desk.nameIdentifier(), finance.nameIdentifier(), AUDIT_INFO, s
-> {}));
+
+ assertWaitsForDomainRow(
+ desk.id(), () ->
domainRelService.removeDomainParentAndGet(desk.nameIdentifier(), s -> {}));
+ assertWaitsForDomainRow(
+ markets.id(),
+ () ->
+ domainRelService.updateDomainStatusAndGet(
+ markets.nameIdentifier(), Domain.Status.PUBLISHED, AUDIT_INFO,
s -> {}));
+
+ Assertions.assertEquals(
+ Domain.Status.PUBLISHED,
+
domainMetaService.getDomainByIdentifier(trading.nameIdentifier()).status());
+ Assertions.assertEquals(
+ Domain.Status.PUBLISHED,
+
domainMetaService.getDomainByIdentifier(markets.nameIdentifier()).status());
+
Assertions.assertFalse(domainRelService.getParentDomainName(desk.id()).isPresent());
+ }
+
+ @TestTemplate
+ public void testLockedReadsBypassTheSessionCache() throws Exception {
+ createAndInsertMakeLake(METALAKE_NAME);
+ DomainEntity finance = insertDomain(METALAKE_NAME, "Finance");
+ DomainEntity markets = insertDomain(METALAKE_NAME, "Markets");
+ DomainEntity trading = insertDomain(METALAKE_NAME, "Trading");
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+
+ try {
+ SessionUtils.doMultipleWithCommit(
+ () -> {
+ // An outer transaction reads the edge before another transaction
attaches the domain;
+ // the empty result is now in this session's cache.
+ Assertions.assertNull(selectDomainRel(trading.id()));
+ try {
+ executor.submit(() -> insertDomainRel(trading, finance)).get(30,
TimeUnit.SECONDS);
+ // Under its locks, the change reads the committed edge and
re-points it, instead of
+ // inserting a second edge from the cached empty read.
+ DomainEntity moved =
+ domainRelService.setDomainParentAndGet(
+ trading.nameIdentifier(), markets.nameIdentifier(),
AUDIT_INFO, s -> {});
+ Assertions.assertEquals("Markets", moved.parent().get());
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ });
+ } finally {
+ executor.shutdownNow();
+ }
+ Assertions.assertEquals("Markets",
domainRelService.getParentDomainName(trading.id()).get());
+ Assertions.assertEquals(1, countDomainRelRows(trading.id()));
+ }
+
+ // On H2 the metalake row is locked exclusively, so there the publish waits
on the metalake row;
+ // testLockedOperationsWaitForDomainRows checks the parent row lock on every
database.
+ @TestTemplate
+ public void testPublishWaitsForConcurrentParentUnpublish() throws Exception {
+ createAndInsertMakeLake(METALAKE_NAME);
+ DomainEntity finance = insertDomain(METALAKE_NAME, "Finance");
+ DomainEntity trading = insertDomain(METALAKE_NAME, "Trading");
+ insertDomainRel(trading, finance);
+ domainRelService.updateDomainStatusAndGet(
+ finance.nameIdentifier(), Domain.Status.PUBLISHED, AUDIT_INFO, s ->
{});
+
+ CountDownLatch unpublishChecked = new CountDownLatch(1);
+ CountDownLatch allowUnpublishCommit = new CountDownLatch(1);
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+ Future<Throwable> unpublishResult =
+ executor.submit(
+ () -> {
+ try {
+ domainRelService.updateDomainStatusAndGet(
+ finance.nameIdentifier(),
+ Domain.Status.DRAFT,
+ AUDIT_INFO,
+ s -> {
+ unpublishChecked.countDown();
+ await(allowUnpublishCommit);
+ });
+ return null;
+ } catch (Throwable throwable) {
+ return throwable;
+ }
+ });
+
+ try {
+ Assertions.assertTrue(unpublishChecked.await(30, TimeUnit.SECONDS));
+ Future<Throwable> publishResult =
+ executor.submit(
+ () -> {
+ try {
+ domainRelService.updateDomainStatusAndGet(
+ trading.nameIdentifier(),
+ Domain.Status.PUBLISHED,
+ AUDIT_INFO,
+ s ->
+ Preconditions.checkArgument(
+ s.parent().get().status() ==
Domain.Status.PUBLISHED,
+ "The parent is not published"));
+ return null;
+ } catch (Throwable throwable) {
+ return throwable;
+ }
+ });
+ // The unpublish holds the parent row, so the publish waits and then
sees the draft parent.
+ Assertions.assertThrows(
+ TimeoutException.class, () -> publishResult.get(500,
TimeUnit.MILLISECONDS));
+
+ allowUnpublishCommit.countDown();
+ Assertions.assertNull(unpublishResult.get(30, TimeUnit.SECONDS));
+ Throwable publishFailure = publishResult.get(30, TimeUnit.SECONDS);
+ Assertions.assertTrue(
+ publishFailure instanceof IllegalArgumentException,
String.valueOf(publishFailure));
+ } finally {
+ allowUnpublishCommit.countDown();
+ executor.shutdownNow();
+ }
+
+ Assertions.assertEquals(
+ Domain.Status.DRAFT,
+
domainMetaService.getDomainByIdentifier(trading.nameIdentifier()).status());
+ }
+
+ // Holds the row of a domain in another transaction, checks that the
operation waits for it,
+ // and lets the operation finish once the row is released.
+ private void assertWaitsForDomainRow(long domainId, Callable<?> operation)
throws Exception {
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try (SqlSession holder =
+
SqlSessionFactoryHelper.getInstance().getSqlSessionFactory().openSession(false))
{
+ try (Statement statement = holder.getConnection().createStatement();
+ ResultSet rs =
+ statement.executeQuery(
+ String.format(
+ "SELECT domain_id FROM domain_meta WHERE domain_id = %d
FOR UPDATE",
+ domainId))) {
+ Assertions.assertTrue(rs.next());
+ }
+ Future<?> result = executor.submit(operation);
+ Assertions.assertThrows(TimeoutException.class, () -> result.get(500,
TimeUnit.MILLISECONDS));
+ holder.rollback(true);
+ result.get(30, TimeUnit.SECONDS);
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+
private void installReadProbe() {
var configuration =
SqlSessionFactoryHelper.getInstance().getSqlSessionFactory().getConfiguration();
@@ -861,6 +1479,32 @@ public class TestDomainRelService extends TestJDBCBackend
{
}
}
+ private DomainEntity newDomain(String metalake, String name, String parent) {
+ return DomainEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName(name)
+ .withNamespace(NamespaceUtil.ofDomain(metalake))
+ .withProperties(ImmutableMap.of("k1", "v1"))
+ .withStatus(Domain.Status.DRAFT)
+ .withParent(parent)
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ }
+
+ private void moveDomainRel(DomainEntity child, DomainEntity newParent)
throws IOException {
+ domainRelService.updateDomainRel(
+ child.nameIdentifier(), newParent.nameIdentifier(), AUDIT_INFO);
+ }
+
+ private DomainPO selectDomainPO(long domainId) {
+ return SessionUtils.getWithoutCommit(
+ DomainMetaMapper.class, mapper ->
mapper.selectDomainMetaById(domainId));
+ }
+
+ private long selectDomainVersion(long domainId) {
+ return selectDomainPO(domainId).currentVersion();
+ }
+
private DomainEntity insertDomain(String metalake, String name) throws
IOException {
Map<String, String> props = ImmutableMap.of("k1", "v1");
DomainEntity domain =
