This is an automated email from the ASF dual-hosted git repository.
mchades 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 db66a5d63b [#13884] feat(core): Add the domain manager (#14522)
db66a5d63b is described below
commit db66a5d63bd831c365e596c5f0b4f91cc18817d6
Author: Jerry Shao <[email protected]>
AuthorDate: Tue Nov 3 20:48:52 2026 +0800
[#13884] feat(core): Add the domain manager (#14522)
### What changes were proposed in this pull request?
Add `DomainDispatcher` and `DomainManager` for creating, listing,
loading, altering and deleting domains; managing sub-domains; and
publishing or unpublishing domains.
Hierarchy and status checks run through the entity store against the
locked domain state. Register the manager in `GravitinoEnv` and support
DOMAIN existence checks in `MetadataObjectUtil`.
### Why are the changes needed?
Provide the server-side domain operations for #13884 while keeping the
hierarchy one level deep and published sub-domains under published
parents.
Part of #13884. Backing-tag creation and binding, authorization, hooks,
events and draft visibility filtering follow in separate PRs. Creation
paths include TODOs for the backing-tag work.
### Does this PR introduce _any_ user-facing change?
No REST or client API changes. This adds internal server operations.
### How was this patch tested?
- H2-backed manager tests cover domain operations, hierarchy and status
rules, moves and no-op cases.
- Mock-store tests cover exception mapping and the checks passed to
locked store operations.
- `./gradlew compileJava compileTestJava spotlessCheck -PskipITs`
- `./gradlew :core:javadoc` (run separately)
- Related tests through `:core:coreUnitTest` and `:core:coreH2Test`.
MySQL and PostgreSQL tests run in CI.
---------
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../java/org/apache/gravitino/GravitinoEnv.java | 28 ++
.../apache/gravitino/domain/DomainDispatcher.java | 241 ++++++++++
.../org/apache/gravitino/domain/DomainManager.java | 517 ++++++++++++++++++++
.../apache/gravitino/utils/MetadataObjectUtil.java | 10 +
.../TestGravitinoEnvMetadataComponents.java | 6 +
.../apache/gravitino/domain/TestDomainManager.java | 521 +++++++++++++++++++++
.../domain/TestDomainManagerWithMockStore.java | 520 ++++++++++++++++++++
7 files changed, 1843 insertions(+)
diff --git a/core/src/main/java/org/apache/gravitino/GravitinoEnv.java
b/core/src/main/java/org/apache/gravitino/GravitinoEnv.java
index 3d17fb2b3e..bf2d0c04bf 100644
--- a/core/src/main/java/org/apache/gravitino/GravitinoEnv.java
+++ b/core/src/main/java/org/apache/gravitino/GravitinoEnv.java
@@ -60,6 +60,8 @@ import org.apache.gravitino.catalog.ViewDispatcher;
import org.apache.gravitino.catalog.ViewNormalizeDispatcher;
import org.apache.gravitino.catalog.ViewOperationDispatcher;
import org.apache.gravitino.credential.CredentialOperationDispatcher;
+import org.apache.gravitino.domain.DomainDispatcher;
+import org.apache.gravitino.domain.DomainManager;
import org.apache.gravitino.encryption.kms.KmsClientRegistry;
import org.apache.gravitino.hook.AccessControlHookDispatcher;
import org.apache.gravitino.hook.CatalogHookDispatcher;
@@ -185,6 +187,9 @@ public class GravitinoEnv {
private PolicyDispatcher policyDispatcher;
private PolicyDispatcher internalPolicyDispatcher;
+ private DomainDispatcher domainDispatcher;
+ private DomainDispatcher internalDomainDispatcher;
+
private AccessControlDispatcher accessControlDispatcher;
private AccessControlDispatcher internalAccessControlDispatcher;
@@ -701,6 +706,25 @@ public class GravitinoEnv {
return internalPolicyDispatcher;
}
+ /**
+ * Get the DomainDispatcher associated with the Gravitino environment.
+ *
+ * @return The DomainDispatcher instance.
+ */
+ public DomainDispatcher domainDispatcher() {
+ return domainDispatcher;
+ }
+
+ /**
+ * Get the internal DomainDispatcher associated with the Gravitino
environment. The internal
+ * dispatcher is the domain manager itself.
+ *
+ * @return The internal DomainDispatcher instance.
+ */
+ public DomainDispatcher internalDomainDispatcher() {
+ return internalDomainDispatcher;
+ }
+
/**
* Get the Owner dispatcher associated with the Gravitino environment.
*
@@ -935,6 +959,7 @@ public class GravitinoEnv {
this.internalTagDispatcher = new TagManager(idGenerator, entityStore);
this.internalPolicyDispatcher = new PolicyManager(idGenerator,
entityStore);
+ this.internalDomainDispatcher = new DomainManager(idGenerator,
entityStore);
return new MetadataOperations(
filesetOperationDispatcher,
@@ -1185,6 +1210,9 @@ public class GravitinoEnv {
PolicyHookDispatcher policyHookDispatcher = new
PolicyHookDispatcher(internalPolicyDispatcher);
this.policyDispatcher = new PolicyEventDispatcher(eventBus,
policyHookDispatcher);
+
+ // Domain hooks and events are added with domain authorization and
auditing.
+ this.domainDispatcher = internalDomainDispatcher;
}
private static final class MetadataOperations {
diff --git
a/core/src/main/java/org/apache/gravitino/domain/DomainDispatcher.java
b/core/src/main/java/org/apache/gravitino/domain/DomainDispatcher.java
new file mode 100644
index 0000000000..3e75233f96
--- /dev/null
+++ b/core/src/main/java/org/apache/gravitino/domain/DomainDispatcher.java
@@ -0,0 +1,241 @@
+/*
+ * 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.domain;
+
+import java.util.Arrays;
+import java.util.Map;
+import javax.annotation.Nullable;
+import org.apache.gravitino.annotation.Evolving;
+import org.apache.gravitino.exceptions.DomainAlreadyExistsException;
+import org.apache.gravitino.exceptions.NoSuchDomainException;
+import org.apache.gravitino.exceptions.NoSuchMetalakeException;
+import org.apache.gravitino.exceptions.NonEmptyDomainException;
+import org.apache.gravitino.exceptions.OptimisticLockException;
+import org.apache.gravitino.meta.DomainEntity;
+
+/**
+ * The domain operations of the server. Domains are one level deep: a domain
is either top-level or
+ * a sub-domain of one top-level domain. Creating a sub-domain, attaching a
domain and moving it
+ * check that:
+ *
+ * <ul>
+ * <li>a domain is not its own parent;
+ * <li>the parent is top-level, so a sub-domain cannot have sub-domains;
+ * <li>the attached domain has no sub-domains, so it does not bring a second
level with it;
+ * <li>parent and child are in the same metalake;
+ * <li>a domain has at most one parent, so attaching a sub-domain moves it;
+ * <li>a published domain is attached only under a published parent.
+ * </ul>
+ *
+ * <p>The first three keep the hierarchy one level deep, so it cannot have a
cycle.
+ *
+ * <p>A new domain is a draft. A sub-domain can be published only under a
published parent, and a
+ * domain can be unpublished only when none of its sub-domains is published.
+ *
+ * <p>The rules are checked against the domains as they are under the row
locks of the change, so a
+ * concurrent change cannot break them. A rule that does not hold fails the
operation with an {@link
+ * IllegalArgumentException}.
+ */
+@Evolving
+public interface DomainDispatcher {
+
+ /**
+ * Lists the names of the domains in a metalake, including sub-domains.
+ *
+ * @param metalake The name of the metalake.
+ * @return The names of the domains.
+ * @throws NoSuchMetalakeException If the metalake does not exist.
+ */
+ default String[] listDomains(String metalake) throws NoSuchMetalakeException
{
+ return
Arrays.stream(listDomainInfos(metalake)).map(DomainEntity::name).toArray(String[]::new);
+ }
+
+ /**
+ * Lists the domains in a metalake, including sub-domains.
+ *
+ * @param metalake The name of the metalake.
+ * @return The domains.
+ * @throws NoSuchMetalakeException If the metalake does not exist.
+ */
+ DomainEntity[] listDomainInfos(String metalake) throws
NoSuchMetalakeException;
+
+ /**
+ * Gets a domain.
+ *
+ * @param metalake The name of the metalake.
+ * @param name The name of the domain.
+ * @return The domain.
+ * @throws NoSuchMetalakeException If the metalake does not exist.
+ * @throws NoSuchDomainException If the domain does not exist.
+ */
+ DomainEntity getDomain(String metalake, String name) throws
NoSuchDomainException;
+
+ // TODO: The next PR of #13884 adds the explicit backing tag to createDomain
and
+ // createSubdomain, which every create request must name, and adds
updateBackingTag.
+
+ /**
+ * Creates a top-level domain in draft status.
+ *
+ * @param metalake The name of the metalake.
+ * @param name The name of the domain.
+ * @param comment The comment of the domain, or null.
+ * @param properties The properties of the domain, or null.
+ * @return The created domain.
+ * @throws NoSuchMetalakeException If the metalake does not exist.
+ * @throws DomainAlreadyExistsException If a domain with the same name
exists.
+ */
+ DomainEntity createDomain(
+ String metalake,
+ String name,
+ @Nullable String comment,
+ @Nullable Map<String, String> properties)
+ throws DomainAlreadyExistsException;
+
+ /**
+ * Creates a sub-domain in draft status under a top-level domain.
+ *
+ * <p>If the metalake is deleted after the initial existence check, creation
reports the parent
+ * domain as not found.
+ *
+ * @param metalake The name of the metalake.
+ * @param parent The name of the parent domain.
+ * @param name The name of the sub-domain.
+ * @param comment The comment of the sub-domain, or null.
+ * @param properties The properties of the sub-domain, or null.
+ * @return The created sub-domain.
+ * @throws NoSuchMetalakeException If the metalake does not exist.
+ * @throws NoSuchDomainException If the parent does not exist.
+ * @throws DomainAlreadyExistsException If a domain with the same name
exists.
+ * @throws IllegalArgumentException If the parent is the sub-domain itself
or a sub-domain.
+ */
+ DomainEntity createSubdomain(
+ String metalake,
+ String parent,
+ String name,
+ @Nullable String comment,
+ @Nullable Map<String, String> properties)
+ throws NoSuchDomainException, DomainAlreadyExistsException,
IllegalArgumentException;
+
+ /**
+ * Alters a domain. Its status and parent are not changed by this method.
+ *
+ * @param metalake The name of the metalake.
+ * @param name The name of the domain.
+ * @param changes The changes to apply.
+ * @return The altered domain.
+ * @throws NoSuchMetalakeException If the metalake does not exist.
+ * @throws NoSuchDomainException If the domain does not exist.
+ * @throws DomainAlreadyExistsException If the domain is renamed to a name
in use.
+ * @throws IllegalArgumentException If a change cannot be applied.
+ */
+ DomainEntity alterDomain(String metalake, String name, DomainChange...
changes)
+ throws NoSuchDomainException, DomainAlreadyExistsException,
IllegalArgumentException;
+
+ /**
+ * Deletes a domain, with its parent edge.
+ *
+ * @param metalake The name of the metalake.
+ * @param name The name of the domain.
+ * @return True if the domain is deleted, false if it does not exist.
+ * @throws NoSuchMetalakeException If the metalake does not exist.
+ * @throws NonEmptyDomainException If the domain has sub-domains.
+ */
+ boolean deleteDomain(String metalake, String name) throws
NonEmptyDomainException;
+
+ /**
+ * Publishes a domain. Publishing a published domain changes nothing.
+ *
+ * @param metalake The name of the metalake.
+ * @param name The name of the domain.
+ * @return The published domain.
+ * @throws NoSuchMetalakeException If the metalake does not exist.
+ * @throws NoSuchDomainException If the domain does not exist.
+ * @throws IllegalArgumentException If the domain is a sub-domain of a draft
parent.
+ * @throws OptimisticLockException If the parent keeps changing while the
status is changed.
+ */
+ DomainEntity publishDomain(String metalake, String name)
+ throws NoSuchDomainException, IllegalArgumentException;
+
+ /**
+ * Unpublishes a domain. Unpublishing a draft domain changes nothing.
+ *
+ * @param metalake The name of the metalake.
+ * @param name The name of the domain.
+ * @return The unpublished domain.
+ * @throws NoSuchMetalakeException If the metalake does not exist.
+ * @throws NoSuchDomainException If the domain does not exist.
+ * @throws IllegalArgumentException If the domain has published sub-domains.
+ * @throws OptimisticLockException If the parent keeps changing while the
status is changed.
+ */
+ DomainEntity unpublishDomain(String metalake, String name)
+ throws NoSuchDomainException, IllegalArgumentException;
+
+ /**
+ * Lists the names of the sub-domains of a domain.
+ *
+ * @param metalake The name of the metalake.
+ * @param name The name of the domain.
+ * @return The names of the sub-domains.
+ * @throws NoSuchMetalakeException If the metalake does not exist.
+ * @throws NoSuchDomainException If the domain does not exist.
+ */
+ default String[] listSubdomains(String metalake, String name) throws
NoSuchDomainException {
+ return Arrays.stream(listSubdomainInfos(metalake, name))
+ .map(DomainEntity::name)
+ .toArray(String[]::new);
+ }
+
+ /**
+ * Lists the sub-domains of a domain.
+ *
+ * @param metalake The name of the metalake.
+ * @param name The name of the domain.
+ * @return The sub-domains.
+ * @throws NoSuchMetalakeException If the metalake does not exist.
+ * @throws NoSuchDomainException If the domain does not exist.
+ */
+ DomainEntity[] listSubdomainInfos(String metalake, String name) throws
NoSuchDomainException;
+
+ /**
+ * Attaches a domain under a parent, or moves a sub-domain to another
parent. Attaching a domain
+ * to its current parent changes nothing.
+ *
+ * @param metalake The name of the metalake.
+ * @param name The name of the domain.
+ * @param parent The name of the new parent.
+ * @return The attached domain.
+ * @throws NoSuchMetalakeException If the metalake does not exist.
+ * @throws NoSuchDomainException If the domain or the parent does not exist.
+ * @throws IllegalArgumentException If the domain cannot be attached under
the parent.
+ */
+ DomainEntity attachDomain(String metalake, String name, String parent)
+ throws NoSuchDomainException, IllegalArgumentException;
+
+ /**
+ * Detaches a domain from its current parent, making it top-level. Detaching
a top-level domain
+ * changes nothing.
+ *
+ * @param metalake The name of the metalake.
+ * @param name The name of the domain.
+ * @return The detached domain.
+ * @throws NoSuchMetalakeException If the metalake does not exist.
+ * @throws NoSuchDomainException If the domain does not exist.
+ */
+ DomainEntity detachDomain(String metalake, String name) throws
NoSuchDomainException;
+}
diff --git a/core/src/main/java/org/apache/gravitino/domain/DomainManager.java
b/core/src/main/java/org/apache/gravitino/domain/DomainManager.java
new file mode 100644
index 0000000000..2fc4e0ef7e
--- /dev/null
+++ b/core/src/main/java/org/apache/gravitino/domain/DomainManager.java
@@ -0,0 +1,517 @@
+/*
+ * 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.domain;
+
+import static org.apache.gravitino.metalake.MetalakeManager.checkMetalake;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import java.io.IOException;
+import java.time.Instant;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.stream.Collectors;
+import javax.annotation.Nullable;
+import org.apache.gravitino.Audit;
+import org.apache.gravitino.Entity;
+import org.apache.gravitino.EntityAlreadyExistsException;
+import org.apache.gravitino.EntityStore;
+import org.apache.gravitino.SupportsDomainOperations;
+import org.apache.gravitino.SupportsDomainOperations.LockedDomainState;
+import org.apache.gravitino.exceptions.DomainAlreadyExistsException;
+import org.apache.gravitino.exceptions.NoSuchDomainException;
+import org.apache.gravitino.exceptions.NoSuchEntityException;
+import org.apache.gravitino.exceptions.NoSuchMetalakeException;
+import org.apache.gravitino.exceptions.NonEmptyDomainException;
+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.storage.IdGenerator;
+import org.apache.gravitino.utils.NameIdentifierUtil;
+import org.apache.gravitino.utils.NamespaceUtil;
+import org.apache.gravitino.utils.PrincipalUtils;
+import org.apache.gravitino.utils.RenameUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * The domain manager. It keeps the domain hierarchy one level deep and the
status of each domain
+ * consistent with its parent and sub-domains (see {@link DomainDispatcher}).
+ *
+ * <p>The rules that involve other domains are checked by the entity store's
domain operations,
+ * against the domains as they are under the row locks of the change, so that
a concurrent change
+ * cannot break them. Name, comment and property changes involve no other
domain and use the store's
+ * update, as other entities do.
+ */
+public class DomainManager implements DomainDispatcher {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(DomainManager.class);
+
+ private final IdGenerator idGenerator;
+ private final EntityStore entityStore;
+
+ /**
+ * Creates a domain manager.
+ *
+ * @param idGenerator The ID generator of new domains.
+ * @param entityStore The entity store, which must support the domain
operations.
+ */
+ public DomainManager(IdGenerator idGenerator, EntityStore entityStore) {
+ this.idGenerator = idGenerator;
+ this.entityStore = entityStore;
+ }
+
+ @Override
+ public DomainEntity[] listDomainInfos(String metalake) {
+ checkMetalake(NameIdentifierUtil.ofMetalake(metalake), entityStore);
+ try {
+ return entityStore
+ .list(NamespaceUtil.ofDomain(metalake), DomainEntity.class,
Entity.EntityType.DOMAIN)
+ .toArray(new DomainEntity[0]);
+ } catch (IOException ioe) {
+ LOG.error("Failed to list domains under metalake {}", metalake, ioe);
+ throw new RuntimeException(ioe);
+ }
+ }
+
+ @Override
+ public DomainEntity getDomain(String metalake, String name) throws
NoSuchDomainException {
+ checkMetalake(NameIdentifierUtil.ofMetalake(metalake), entityStore);
+ try {
+ return entityStore.get(
+ NameIdentifierUtil.ofDomain(metalake, name),
+ Entity.EntityType.DOMAIN,
+ DomainEntity.class);
+ } catch (NoSuchEntityException e) {
+ throw noSuchDomain(e, metalake, name);
+ } catch (IOException ioe) {
+ LOG.error("Failed to get domain {} under metalake {}", name, metalake,
ioe);
+ throw new RuntimeException(ioe);
+ }
+ }
+
+ @Override
+ public DomainEntity createDomain(
+ String metalake,
+ String name,
+ @Nullable String comment,
+ @Nullable Map<String, String> properties)
+ throws DomainAlreadyExistsException {
+ return insertDomain(metalake, null, name, comment, properties);
+ }
+
+ @Override
+ public DomainEntity createSubdomain(
+ String metalake,
+ String parent,
+ String name,
+ @Nullable String comment,
+ @Nullable Map<String, String> properties)
+ throws NoSuchDomainException, DomainAlreadyExistsException,
IllegalArgumentException {
+ checkParentName(name, parent);
+ return insertDomain(metalake, parent, name, comment, properties);
+ }
+
+ @Override
+ public DomainEntity alterDomain(String metalake, String name,
DomainChange... changes)
+ throws NoSuchDomainException, DomainAlreadyExistsException,
IllegalArgumentException {
+ checkMetalake(NameIdentifierUtil.ofMetalake(metalake), entityStore);
+ try {
+ return entityStore.update(
+ NameIdentifierUtil.ofDomain(metalake, name),
+ DomainEntity.class,
+ Entity.EntityType.DOMAIN,
+ domain -> applyChanges(domain, changes));
+ } catch (NoSuchEntityException e) {
+ throw noSuchDomain(e, metalake, name);
+ } catch (EntityAlreadyExistsException e) {
+ String newName =
+ RenameUtils.getLastRenameName(
+ name,
+ changes,
+ DomainChange.RenameDomain.class,
+ DomainChange.RenameDomain::getNewName);
+ throw new DomainAlreadyExistsException(
+ e, "Domain %s under metalake %s already exists", newName, metalake);
+ } catch (OptimisticLockException ole) {
+ LOG.warn(
+ "Failed to alter domain {} under metalake {} because of a concurrent
operation",
+ name,
+ metalake,
+ ole);
+ throw ole;
+ } catch (IOException ioe) {
+ LOG.error("Failed to alter domain {} under metalake {}", name, metalake,
ioe);
+ throw new RuntimeException(ioe);
+ }
+ }
+
+ @Override
+ public boolean deleteDomain(String metalake, String name) throws
NonEmptyDomainException {
+ checkMetalake(NameIdentifierUtil.ofMetalake(metalake), entityStore);
+ try {
+ // In one transaction, the store refuses a domain with sub-domains, and
removes the domain
+ // with its own parent edge, its owner and the privileges on it. Its
backing tag stays.
+ return entityStore.delete(
+ NameIdentifierUtil.ofDomain(metalake, name),
Entity.EntityType.DOMAIN);
+ } catch (NonEmptyEntityException e) {
+ throw new NonEmptyDomainException(
+ e, "Domain %s under metalake %s has sub-domains, remove them first",
name, metalake);
+ } catch (OptimisticLockException ole) {
+ LOG.warn(
+ "Failed to delete domain {} under metalake {} because of a
concurrent operation",
+ name,
+ metalake,
+ ole);
+ throw ole;
+ } catch (IOException ioe) {
+ LOG.error("Failed to delete domain {} under metalake {}", name,
metalake, ioe);
+ throw new RuntimeException(ioe);
+ }
+ }
+
+ @Override
+ public DomainEntity publishDomain(String metalake, String name)
+ throws NoSuchDomainException, IllegalArgumentException {
+ return updateStatus(metalake, name, Domain.Status.PUBLISHED);
+ }
+
+ @Override
+ public DomainEntity unpublishDomain(String metalake, String name)
+ throws NoSuchDomainException, IllegalArgumentException {
+ return updateStatus(metalake, name, Domain.Status.DRAFT);
+ }
+
+ @Override
+ public DomainEntity[] listSubdomainInfos(String metalake, String name)
+ throws NoSuchDomainException {
+ checkMetalake(NameIdentifierUtil.ofMetalake(metalake), entityStore);
+ try {
+ return domainOperations()
+ .listSubDomains(NameIdentifierUtil.ofDomain(metalake, name))
+ .toArray(new DomainEntity[0]);
+ } catch (NoSuchEntityException e) {
+ throw noSuchDomain(e, metalake, name);
+ } catch (IOException ioe) {
+ LOG.error("Failed to list sub-domains of domain {} under metalake {}",
name, metalake, ioe);
+ throw new RuntimeException(ioe);
+ }
+ }
+
+ @Override
+ public DomainEntity attachDomain(String metalake, String name, String parent)
+ throws NoSuchDomainException, IllegalArgumentException {
+ checkParentName(name, parent);
+ checkMetalake(NameIdentifierUtil.ofMetalake(metalake), entityStore);
+ try {
+ return domainOperations()
+ .setDomainParentAndGet(
+ NameIdentifierUtil.ofDomain(metalake, name),
+ NameIdentifierUtil.ofDomain(metalake, parent),
+ newAuditInfo(),
+ DomainManager::checkAttach);
+ } catch (NoSuchEntityException e) {
+ throw new NoSuchDomainException(
+ e,
+ "Domain %s or parent %s under metalake %s does not exist: %s",
+ name,
+ parent,
+ metalake,
+ e.getMessage());
+ } catch (IOException ioe) {
+ LOG.error("Failed to attach domain {} to {} under metalake {}", name,
parent, metalake, ioe);
+ throw new RuntimeException(ioe);
+ }
+ }
+
+ @Override
+ public DomainEntity detachDomain(String metalake, String name) throws
NoSuchDomainException {
+ checkMetalake(NameIdentifierUtil.ofMetalake(metalake), entityStore);
+ try {
+ // Detaching breaks no rule of the hierarchy or the status, so nothing
is checked.
+ return domainOperations()
+ .removeDomainParentAndGet(NameIdentifierUtil.ofDomain(metalake,
name), state -> {});
+ } catch (NoSuchEntityException e) {
+ throw noSuchDomain(e, metalake, name);
+ } catch (IOException ioe) {
+ LOG.error("Failed to detach domain {} under metalake {}", name,
metalake, ioe);
+ throw new RuntimeException(ioe);
+ }
+ }
+
+ /**
+ * Checks a new sub-domain against its locked parent: the parent must be
top-level, so that a
+ * sub-domain has no sub-domains. A new domain has no sub-domains of its own
and is a draft, so
+ * the other rules of an attach hold for it.
+ *
+ * @param state The locked state of the insert.
+ */
+ @VisibleForTesting
+ static void checkNewDomain(LockedDomainState state) {
+ state.parent().ifPresent(parent -> checkParentIsTopLevel(state.domain(),
parent));
+ }
+
+ /**
+ * Checks an attach or a move against the locked domains. Attaching a domain
to its current parent
+ * changes nothing and is not checked.
+ *
+ * @param state The locked state of the attach.
+ */
+ @VisibleForTesting
+ static void checkAttach(LockedDomainState state) {
+ DomainEntity domain = state.domain();
+ DomainEntity parent =
+ state.parent().orElseThrow(() -> new IllegalStateException("An attach
needs a parent"));
+ // Two different names can find the same row, such as names that differ
only in trailing
+ // spaces on MySQL. The store refuses that before it locks; this checks
the locked rows again,
+ // so the rule does not depend on one store implementation.
+ Preconditions.checkArgument(
+ !Objects.equals(domain.id(), parent.id()),
+ "Domain %s cannot be its own parent",
+ domain.name());
+ if (domain.parent().filter(parent.name()::equals).isPresent()) {
+ return;
+ }
+
+ // A sub-domain cannot be a parent, or the hierarchy would be two levels
deep.
+ checkParentIsTopLevel(domain, parent);
+ // A domain with sub-domains cannot become a sub-domain, for the same
reason.
+ Preconditions.checkArgument(
+ state.subDomains().isEmpty(),
+ "Domain %s has sub-domains %s and cannot be attached to another
domain",
+ domain.name(),
+ names(state.subDomains()));
+ // A published sub-domain needs a published parent, as when it is
published.
+ Preconditions.checkArgument(
+ domain.status() != Domain.Status.PUBLISHED || parent.status() ==
Domain.Status.PUBLISHED,
+ "Published domain %s cannot be attached to draft domain %s, publish %s
first",
+ domain.name(),
+ parent.name(),
+ parent.name());
+ }
+
+ /**
+ * Checks a publish against the locked domains: a sub-domain needs a
published parent.
+ *
+ * @param state The locked state of the status change.
+ */
+ @VisibleForTesting
+ static void checkPublish(LockedDomainState state) {
+ if (state.domain().status() == Domain.Status.PUBLISHED) {
+ return;
+ }
+ state
+ .parent()
+ .ifPresent(
+ parent ->
+ Preconditions.checkArgument(
+ parent.status() == Domain.Status.PUBLISHED,
+ "Domain %s cannot be published while its parent %s is a
draft",
+ state.domain().name(),
+ parent.name()));
+ }
+
+ /**
+ * Checks an unpublish against the locked domains: no sub-domain may be
published.
+ *
+ * @param state The locked state of the status change.
+ */
+ @VisibleForTesting
+ static void checkUnpublish(LockedDomainState state) {
+ if (state.domain().status() == Domain.Status.DRAFT) {
+ return;
+ }
+ List<DomainEntity> publishedSubDomains =
+ state.subDomains().stream()
+ .filter(subDomain -> subDomain.status() == Domain.Status.PUBLISHED)
+ .collect(Collectors.toList());
+ Preconditions.checkArgument(
+ publishedSubDomains.isEmpty(),
+ "Domain %s cannot be unpublished while its sub-domains %s are
published",
+ state.domain().name(),
+ names(publishedSubDomains));
+ }
+
+ private DomainEntity insertDomain(
+ String metalake,
+ @Nullable String parent,
+ String name,
+ @Nullable String comment,
+ @Nullable Map<String, String> properties) {
+ checkMetalake(NameIdentifierUtil.ofMetalake(metalake), entityStore);
+ // TODO: The next PR of #13884 sets the backing tag here: it creates a new
tag or binds an
+ // existing (tag, value), in the same transaction as the domain, under the
binding rules.
+ DomainEntity domain =
+ DomainEntity.builder()
+ .withId(idGenerator.nextId())
+ .withName(name)
+ .withNamespace(NamespaceUtil.ofDomain(metalake))
+ .withComment(comment)
+ .withProperties(properties == null ? Collections.emptyMap() :
properties)
+ .withStatus(Domain.Status.DRAFT)
+ .withParent(parent)
+ .withAuditInfo(
+ AuditInfo.builder()
+
.withCreator(PrincipalUtils.getCurrentPrincipal().getName())
+ .withCreateTime(Instant.now())
+ .build())
+ .build();
+
+ try {
+ // Return the parent's name as stored, read under its lock, rather than
the requested one:
+ // on MySQL, a name with trailing spaces finds the same parent row.
+ String[] storedParent = new String[] {parent};
+ domainOperations()
+ .insertDomain(
+ domain,
+ state -> {
+ checkNewDomain(state);
+ state.parent().ifPresent(lockedParent -> storedParent[0] =
lockedParent.name());
+ });
+ return Objects.equals(storedParent[0], parent)
+ ? domain
+ : domain.toBuilder().withParent(storedParent[0]).build();
+ } catch (EntityAlreadyExistsException e) {
+ throw new DomainAlreadyExistsException(
+ e, "Domain %s under metalake %s already exists", name, metalake);
+ } catch (NoSuchEntityException e) {
+ if (parent == null) {
+ throw new NoSuchMetalakeException(e, "Metalake %s does not exist",
metalake);
+ }
+ // A missing metalake is reported by the pre-check. If it is deleted
concurrently after
+ // that check, the requested parent domain is also gone; report it as
not found.
+ throw new NoSuchDomainException(
+ e,
+ "Parent domain %s under metalake %s does not exist: %s",
+ parent,
+ metalake,
+ e.getMessage());
+ } catch (IOException ioe) {
+ LOG.error("Failed to create domain {} under metalake {}", name,
metalake, ioe);
+ throw new RuntimeException(ioe);
+ }
+ }
+
+ private DomainEntity updateStatus(String metalake, String name,
Domain.Status status) {
+ checkMetalake(NameIdentifierUtil.ofMetalake(metalake), entityStore);
+ try {
+ return domainOperations()
+ .updateDomainStatusAndGet(
+ NameIdentifierUtil.ofDomain(metalake, name),
+ status,
+ newAuditInfo(),
+ status == Domain.Status.PUBLISHED
+ ? DomainManager::checkPublish
+ : DomainManager::checkUnpublish);
+ } catch (NoSuchEntityException e) {
+ throw noSuchDomain(e, metalake, name);
+ } catch (IOException ioe) {
+ LOG.error(
+ "Failed to change the status of domain {} under metalake {} to {}",
+ name,
+ metalake,
+ status,
+ ioe);
+ throw new RuntimeException(ioe);
+ }
+ }
+
+ private SupportsDomainOperations domainOperations() {
+ return entityStore.domainOperations();
+ }
+
+ private static DomainEntity applyChanges(DomainEntity domain,
DomainChange... changes) {
+ String newName = domain.name();
+ String newComment = domain.comment();
+ Map<String, String> newProperties =
+ domain.properties() == null ? new HashMap<>() : new
HashMap<>(domain.properties());
+
+ for (DomainChange change : changes) {
+ if (change instanceof DomainChange.RenameDomain) {
+ newName = ((DomainChange.RenameDomain) change).getNewName();
+ } else if (change instanceof DomainChange.UpdateDomainComment) {
+ newComment = ((DomainChange.UpdateDomainComment)
change).getNewComment();
+ } else if (change instanceof DomainChange.SetProperty) {
+ DomainChange.SetProperty setProperty = (DomainChange.SetProperty)
change;
+ newProperties.put(setProperty.getProperty(), setProperty.getValue());
+ } else if (change instanceof DomainChange.RemoveProperty) {
+ newProperties.remove(((DomainChange.RemoveProperty)
change).getProperty());
+ } else {
+ throw new IllegalArgumentException("Unsupported domain change: " +
change);
+ }
+ }
+
+ Audit audit = domain.auditInfo();
+ return domain.toBuilder()
+ .withName(newName)
+ .withComment(newComment)
+ .withProperties(newProperties)
+ .withAuditInfo(
+ AuditInfo.builder()
+ .withCreator(audit.creator())
+ .withCreateTime(audit.createTime())
+
.withLastModifier(PrincipalUtils.getCurrentPrincipal().getName())
+ .withLastModifiedTime(Instant.now())
+ .build())
+ .build();
+ }
+
+ // The audit info of a change: a new parent edge takes its creator, a
changed row its modifier.
+ private static AuditInfo newAuditInfo() {
+ String user = PrincipalUtils.getCurrentPrincipal().getName();
+ Instant now = Instant.now();
+ return AuditInfo.builder()
+ .withCreator(user)
+ .withCreateTime(now)
+ .withLastModifier(user)
+ .withLastModifiedTime(now)
+ .build();
+ }
+
+ // A domain cannot be its own parent. Parent and child are always in the
same metalake, since
+ // both are looked up in the given one.
+ private static void checkParentName(String name, @Nullable String parent) {
+ Preconditions.checkArgument(parent != null, "The parent domain name must
not be null");
+ Preconditions.checkArgument(
+ !Objects.equals(name, parent), "Domain %s cannot be its own parent",
name);
+ }
+
+ private static void checkParentIsTopLevel(DomainEntity domain, DomainEntity
parent) {
+ Preconditions.checkArgument(
+ !parent.parent().isPresent(),
+ "Domain %s cannot be a sub-domain of %s, which is itself a sub-domain
of %s",
+ domain.name(),
+ parent.name(),
+ parent.parent().orElse(null));
+ }
+
+ private static List<String> names(List<DomainEntity> domains) {
+ return
domains.stream().map(DomainEntity::name).collect(Collectors.toList());
+ }
+
+ private static NoSuchDomainException noSuchDomain(
+ NoSuchEntityException e, String metalake, String name) {
+ return new NoSuchDomainException(
+ e, "Domain %s under metalake %s does not exist", name, metalake);
+ }
+}
diff --git
a/core/src/main/java/org/apache/gravitino/utils/MetadataObjectUtil.java
b/core/src/main/java/org/apache/gravitino/utils/MetadataObjectUtil.java
index 0413866c69..65968cc218 100644
--- a/core/src/main/java/org/apache/gravitino/utils/MetadataObjectUtil.java
+++ b/core/src/main/java/org/apache/gravitino/utils/MetadataObjectUtil.java
@@ -37,6 +37,7 @@ import org.apache.gravitino.MetadataObjects;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.authorization.AuthorizationUtils;
import org.apache.gravitino.exceptions.IllegalMetadataObjectException;
+import org.apache.gravitino.exceptions.NoSuchDomainException;
import org.apache.gravitino.exceptions.NoSuchJobException;
import org.apache.gravitino.exceptions.NoSuchJobTemplateException;
import org.apache.gravitino.exceptions.NoSuchMetadataObjectException;
@@ -350,6 +351,15 @@ public class MetadataObjectUtil {
}
break;
+ case DOMAIN:
+ NameIdentifierUtil.checkDomain(identifier);
+ try {
+ env.internalDomainDispatcher().getDomain(metalake,
object.fullName());
+ } catch (NoSuchDomainException e) {
+ throw exceptionToThrowSupplier.get();
+ }
+ break;
+
default:
throw new IllegalArgumentException(
String.format("Doesn't support the type %s", object.type()));
diff --git
a/core/src/test/java/org/apache/gravitino/TestGravitinoEnvMetadataComponents.java
b/core/src/test/java/org/apache/gravitino/TestGravitinoEnvMetadataComponents.java
index 9d2e182905..9416976216 100644
---
a/core/src/test/java/org/apache/gravitino/TestGravitinoEnvMetadataComponents.java
+++
b/core/src/test/java/org/apache/gravitino/TestGravitinoEnvMetadataComponents.java
@@ -56,6 +56,7 @@ import org.apache.gravitino.catalog.TopicNormalizeDispatcher;
import org.apache.gravitino.catalog.TopicOperationDispatcher;
import org.apache.gravitino.catalog.ViewNormalizeDispatcher;
import org.apache.gravitino.catalog.ViewOperationDispatcher;
+import org.apache.gravitino.domain.DomainManager;
import org.apache.gravitino.hook.FilesetHookDispatcher;
import org.apache.gravitino.hook.FunctionHookDispatcher;
import org.apache.gravitino.hook.ModelHookDispatcher;
@@ -156,6 +157,7 @@ class TestGravitinoEnvMetadataComponents {
assertNotNull(env.secretPropertyOperationDispatcher());
assertNotNull(env.internalTagDispatcher());
assertNotNull(env.internalPolicyDispatcher());
+ assertInstanceOf(DomainManager.class, env.internalDomainDispatcher());
assertInstanceOf(StatisticManager.class,
env.internalStatisticDispatcher());
assertNotNull(env.lockManager());
assertNotNull(env.metricsSystem());
@@ -175,6 +177,7 @@ class TestGravitinoEnvMetadataComponents {
assertNull(env.viewDispatcher());
assertNull(env.tagDispatcher());
assertNull(env.policyDispatcher());
+ assertNull(env.domainDispatcher());
assertNull(env.statisticDispatcher());
assertNull(env.internalAccessControlDispatcher());
assertNull(env.internalOwnerDispatcher());
@@ -238,6 +241,9 @@ class TestGravitinoEnvMetadataComponents {
env.initializeFullComponents(config);
+ assertInstanceOf(DomainManager.class, env.internalDomainDispatcher());
+ assertSame(env.internalDomainDispatcher(), env.domainDispatcher());
+
assertDispatcherChain(
env.filesetDispatcher(),
FilesetEventDispatcher.class,
diff --git
a/core/src/test/java/org/apache/gravitino/domain/TestDomainManager.java
b/core/src/test/java/org/apache/gravitino/domain/TestDomainManager.java
new file mode 100644
index 0000000000..7a0463ef82
--- /dev/null
+++ b/core/src/test/java/org/apache/gravitino/domain/TestDomainManager.java
@@ -0,0 +1,521 @@
+/*
+ * 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.domain;
+
+import static
org.apache.gravitino.Configs.DEFAULT_ENTITY_CHANGE_LOG_POLL_BATCH_SIZE;
+import static
org.apache.gravitino.Configs.DEFAULT_ENTITY_CHANGE_LOG_REPLAY_MAX_ENTRIES;
+import static
org.apache.gravitino.Configs.DEFAULT_ENTITY_CHANGE_LOG_REPLAY_WINDOW_SECS;
+import static org.apache.gravitino.Configs.DEFAULT_ENTITY_RELATIONAL_STORE;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS;
+import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_POLL_BATCH_SIZE;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_REPLAY_MAX_ENTRIES;
+import static
org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_REPLAY_WINDOW_SECS;
+import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_RETENTION_SECS;
+import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_DRIVER;
+import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_MAX_CONNECTIONS;
+import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_MAX_IDLE_CONNECTIONS;
+import static org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_URL;
+import static
org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_WAIT_MILLISECONDS;
+import static org.apache.gravitino.Configs.ENTITY_RELATIONAL_STORE;
+import static org.apache.gravitino.Configs.ENTITY_STORE;
+import static org.apache.gravitino.Configs.RELATIONAL_ENTITY_STORE;
+import static org.apache.gravitino.Configs.STORE_DELETE_AFTER_TIME;
+import static org.apache.gravitino.Configs.STORE_TRANSACTION_MAX_SKEW_TIME;
+import static org.apache.gravitino.Configs.TREE_LOCK_CLEAN_INTERVAL;
+import static org.apache.gravitino.Configs.TREE_LOCK_MAX_NODE_IN_MEMORY;
+import static org.apache.gravitino.Configs.TREE_LOCK_MIN_NODE_IN_MEMORY;
+import static org.apache.gravitino.Configs.VERSION_RETENTION_COUNT;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
+import java.io.File;
+import java.io.IOException;
+import java.time.Instant;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import org.apache.commons.io.FileUtils;
+import org.apache.commons.lang3.reflect.FieldUtils;
+import org.apache.gravitino.Config;
+import org.apache.gravitino.Configs;
+import org.apache.gravitino.EntityStore;
+import org.apache.gravitino.EntityStoreFactory;
+import org.apache.gravitino.GravitinoEnv;
+import org.apache.gravitino.MetadataObject;
+import org.apache.gravitino.MetadataObjects;
+import org.apache.gravitino.SupportsDomainOperations.LockedDomainState;
+import org.apache.gravitino.exceptions.DomainAlreadyExistsException;
+import org.apache.gravitino.exceptions.NoSuchDomainException;
+import org.apache.gravitino.exceptions.NoSuchMetadataObjectException;
+import org.apache.gravitino.exceptions.NoSuchMetalakeException;
+import org.apache.gravitino.exceptions.NonEmptyDomainException;
+import org.apache.gravitino.meta.AuditInfo;
+import org.apache.gravitino.meta.BaseMetalake;
+import org.apache.gravitino.meta.DomainEntity;
+import org.apache.gravitino.meta.SchemaVersion;
+import org.apache.gravitino.storage.IdGenerator;
+import org.apache.gravitino.storage.RandomIdGenerator;
+import org.apache.gravitino.utils.MetadataObjectUtil;
+import org.apache.gravitino.utils.NamespaceUtil;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+public class TestDomainManager {
+
+ private static final String METALAKE = "metalake_for_domain_test";
+ private static final String JDBC_STORE_PATH =
+ "/tmp/gravitino_jdbc_entityStore_" +
UUID.randomUUID().toString().replace("-", "");
+ private static final String DB_DIR = JDBC_STORE_PATH + "/testdb";
+
+ private static EntityStore entityStore;
+ private static DomainManager domainManager;
+
+ @BeforeAll
+ public static void setUp() throws IllegalAccessException, IOException {
+ IdGenerator idGenerator = new RandomIdGenerator();
+ Config config = mockConfig();
+ entityStore = EntityStoreFactory.createEntityStore(config);
+ entityStore.initialize(config);
+ domainManager = new DomainManager(idGenerator, entityStore);
+
+ FieldUtils.writeField(
+ GravitinoEnv.getInstance(), "internalDomainDispatcher", domainManager,
true);
+
+ BaseMetalake metalake =
+ BaseMetalake.builder()
+ .withId(idGenerator.nextId())
+ .withName(METALAKE)
+ .withVersion(SchemaVersion.V_0_1)
+ .withComment("Test metalake")
+ .withAuditInfo(
+
AuditInfo.builder().withCreator("test").withCreateTime(Instant.now()).build())
+ .build();
+ entityStore.put(metalake, false /* overwritten */);
+ }
+
+ @AfterAll
+ public static void tearDown() throws IOException, IllegalAccessException {
+ FieldUtils.writeField(GravitinoEnv.getInstance(),
"internalDomainDispatcher", null, true);
+ if (entityStore != null) {
+ entityStore.close();
+ entityStore = null;
+ }
+ FileUtils.deleteDirectory(new File(JDBC_STORE_PATH));
+ }
+
+ @AfterEach
+ public void cleanUp() {
+ DomainEntity[] domains = domainManager.listDomainInfos(METALAKE);
+ // Sub-domains first, since a domain with sub-domains cannot be deleted.
+ Arrays.stream(domains)
+ .filter(domain -> domain.parent().isPresent())
+ .forEach(domain -> domainManager.deleteDomain(METALAKE,
domain.name()));
+ Arrays.stream(domains)
+ .filter(domain -> !domain.parent().isPresent())
+ .forEach(domain -> domainManager.deleteDomain(METALAKE,
domain.name()));
+ }
+
+ @Test
+ public void testCreateGetAndListDomains() {
+ Map<String, String> properties = ImmutableMap.of("k1", "v1");
+ DomainEntity finance = domainManager.createDomain(METALAKE, "Finance",
"comment", properties);
+ Assertions.assertEquals("Finance", finance.name());
+ Assertions.assertEquals(NamespaceUtil.ofDomain(METALAKE),
finance.namespace());
+ Assertions.assertEquals("comment", finance.comment());
+ Assertions.assertEquals(properties, finance.properties());
+ Assertions.assertEquals(Domain.Status.DRAFT, finance.status());
+ Assertions.assertFalse(finance.parent().isPresent());
+ Assertions.assertFalse(finance.backingTag().isPresent());
+ Assertions.assertNotNull(finance.auditInfo().creator());
+ Assertions.assertEquals(finance, domainManager.getDomain(METALAKE,
"Finance"));
+
+ DomainEntity markets = domainManager.createDomain(METALAKE, "Markets",
null, null);
+ Assertions.assertNull(markets.comment());
+ Assertions.assertTrue(markets.properties().isEmpty());
+
+ Assertions.assertEquals(
+ ImmutableSet.of("Finance", "Markets"),
+ ImmutableSet.copyOf(domainManager.listDomains(METALAKE)));
+ Assertions.assertEquals(
+ ImmutableSet.of(finance, markets),
+ ImmutableSet.copyOf(domainManager.listDomainInfos(METALAKE)));
+
+ Assertions.assertThrows(
+ DomainAlreadyExistsException.class,
+ () -> domainManager.createDomain(METALAKE, "Finance", null, null));
+ Assertions.assertThrows(
+ NoSuchDomainException.class, () -> domainManager.getDomain(METALAKE,
"missing"));
+ Assertions.assertThrows(
+ NoSuchMetalakeException.class,
+ () -> domainManager.createDomain("missing_metalake", "Finance", null,
null));
+ Assertions.assertThrows(
+ NoSuchMetalakeException.class, () ->
domainManager.listDomains("missing_metalake"));
+ }
+
+ @Test
+ public void testCreateSubdomain() {
+ domainManager.createDomain(METALAKE, "Finance", null, null);
+ DomainEntity trading =
+ domainManager.createSubdomain(METALAKE, "Finance", "Trading",
"comment", null);
+ Assertions.assertEquals("Finance", trading.parent().get());
+ Assertions.assertEquals(Domain.Status.DRAFT, trading.status());
+ Assertions.assertEquals(trading, domainManager.getDomain(METALAKE,
"Trading"));
+ Assertions.assertArrayEquals(
+ new String[] {"Trading"}, domainManager.listSubdomains(METALAKE,
"Finance"));
+ Assertions.assertArrayEquals(
+ new DomainEntity[] {trading},
domainManager.listSubdomainInfos(METALAKE, "Finance"));
+ Assertions.assertEquals(0, domainManager.listSubdomains(METALAKE,
"Trading").length);
+
+ // A sub-domain cannot have sub-domains.
+ IllegalArgumentException e =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> domainManager.createSubdomain(METALAKE, "Trading", "Desk",
null, null));
+ Assertions.assertTrue(e.getMessage().contains("itself a sub-domain"),
e.getMessage());
+ Assertions.assertThrows(
+ NoSuchDomainException.class, () -> domainManager.getDomain(METALAKE,
"Desk"));
+
+ Assertions.assertThrows(
+ NoSuchDomainException.class,
+ () -> domainManager.createSubdomain(METALAKE, "missing", "Desk", null,
null));
+ // A domain cannot be its own parent.
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> domainManager.createSubdomain(METALAKE, "Finance", "Finance",
null, null));
+ Assertions.assertThrows(
+ DomainAlreadyExistsException.class,
+ () -> domainManager.createSubdomain(METALAKE, "Finance", "Trading",
null, null));
+ Assertions.assertThrows(
+ NoSuchDomainException.class, () ->
domainManager.listSubdomains(METALAKE, "missing"));
+ }
+
+ @Test
+ public void testAlterDomain() {
+ domainManager.createDomain(METALAKE, "Finance", "comment",
ImmutableMap.of("k1", "v1"));
+ domainManager.createSubdomain(METALAKE, "Finance", "Trading", null, null);
+ domainManager.createDomain(METALAKE, "Markets", null, null);
+
+ DomainEntity altered =
+ domainManager.alterDomain(
+ METALAKE,
+ "Finance",
+ DomainChange.rename("Finance2"),
+ DomainChange.updateComment("new comment"),
+ DomainChange.setProperty("k2", "v2"),
+ DomainChange.removeProperty("k1"));
+ Assertions.assertEquals("Finance2", altered.name());
+ Assertions.assertEquals("new comment", altered.comment());
+ Assertions.assertEquals(ImmutableMap.of("k2", "v2"), altered.properties());
+ Assertions.assertEquals(Domain.Status.DRAFT, altered.status());
+ Assertions.assertNotNull(altered.auditInfo().lastModifier());
+ Assertions.assertEquals(altered, domainManager.getDomain(METALAKE,
"Finance2"));
+ Assertions.assertThrows(
+ NoSuchDomainException.class, () -> domainManager.getDomain(METALAKE,
"Finance"));
+ // The sub-domain follows the renamed parent, and a sub-domain keeps its
parent when altered.
+ Assertions.assertEquals(
+ "Finance2", domainManager.getDomain(METALAKE,
"Trading").parent().get());
+ Assertions.assertEquals(
+ "Finance2",
+ domainManager
+ .alterDomain(METALAKE, "Trading", DomainChange.updateComment("c"))
+ .parent()
+ .get());
+
+ Assertions.assertThrows(
+ DomainAlreadyExistsException.class,
+ () -> domainManager.alterDomain(METALAKE, "Finance2",
DomainChange.rename("Markets")));
+ Assertions.assertThrows(
+ NoSuchDomainException.class,
+ () -> domainManager.alterDomain(METALAKE, "missing",
DomainChange.updateComment("c")));
+ }
+
+ @Test
+ public void testDeleteDomain() {
+ domainManager.createDomain(METALAKE, "Finance", null, null);
+ domainManager.createSubdomain(METALAKE, "Finance", "Trading", null, null);
+
+ Assertions.assertThrows(
+ NonEmptyDomainException.class, () ->
domainManager.deleteDomain(METALAKE, "Finance"));
+ Assertions.assertNotNull(domainManager.getDomain(METALAKE, "Finance"));
+
+ Assertions.assertTrue(domainManager.deleteDomain(METALAKE, "Trading"));
+ Assertions.assertEquals(0, domainManager.listSubdomains(METALAKE,
"Finance").length);
+ Assertions.assertTrue(domainManager.deleteDomain(METALAKE, "Finance"));
+ Assertions.assertFalse(domainManager.deleteDomain(METALAKE, "Finance"));
+ Assertions.assertEquals(0, domainManager.listDomains(METALAKE).length);
+ }
+
+ @Test
+ public void testPublishAndUnpublish() {
+ domainManager.createDomain(METALAKE, "Finance", null, null);
+ domainManager.createSubdomain(METALAKE, "Finance", "Trading", null, null);
+
+ // A sub-domain needs a published parent.
+ Assertions.assertThrows(
+ IllegalArgumentException.class, () ->
domainManager.publishDomain(METALAKE, "Trading"));
+ Assertions.assertEquals(
+ Domain.Status.DRAFT, domainManager.getDomain(METALAKE,
"Trading").status());
+
+ DomainEntity finance = domainManager.publishDomain(METALAKE, "Finance");
+ Assertions.assertEquals(Domain.Status.PUBLISHED, finance.status());
+ Assertions.assertNotNull(finance.auditInfo().lastModifier());
+ // Publishing a published domain changes nothing.
+ Assertions.assertEquals(finance, domainManager.publishDomain(METALAKE,
"Finance"));
+ DomainEntity trading = domainManager.publishDomain(METALAKE, "Trading");
+ Assertions.assertEquals(Domain.Status.PUBLISHED, trading.status());
+ Assertions.assertEquals("Finance", trading.parent().get());
+
+ // A domain with published sub-domains cannot be unpublished.
+ IllegalArgumentException e =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> domainManager.unpublishDomain(METALAKE, "Finance"));
+ Assertions.assertTrue(e.getMessage().contains("[Trading]"),
e.getMessage());
+
+ Assertions.assertEquals(
+ Domain.Status.DRAFT, domainManager.unpublishDomain(METALAKE,
"Trading").status());
+ Assertions.assertEquals(
+ Domain.Status.DRAFT, domainManager.unpublishDomain(METALAKE,
"Finance").status());
+ // Unpublishing a draft domain changes nothing.
+ Assertions.assertEquals(
+ Domain.Status.DRAFT, domainManager.unpublishDomain(METALAKE,
"Finance").status());
+
+ Assertions.assertThrows(
+ NoSuchDomainException.class, () ->
domainManager.publishDomain(METALAKE, "missing"));
+ Assertions.assertThrows(
+ NoSuchDomainException.class, () ->
domainManager.unpublishDomain(METALAKE, "missing"));
+ }
+
+ @Test
+ public void testAttachMoveAndDetach() {
+ domainManager.createDomain(METALAKE, "Finance", null, null);
+ domainManager.createDomain(METALAKE, "Markets", null, null);
+ domainManager.createDomain(METALAKE, "Trading", null, null);
+
+ DomainEntity attached = domainManager.attachDomain(METALAKE, "Trading",
"Finance");
+ Assertions.assertEquals("Finance", attached.parent().get());
+ Assertions.assertArrayEquals(
+ new String[] {"Trading"}, domainManager.listSubdomains(METALAKE,
"Finance"));
+ // Attaching a domain to its current parent changes nothing.
+ Assertions.assertEquals(attached, domainManager.attachDomain(METALAKE,
"Trading", "Finance"));
+
+ // A domain has one parent, so attaching a sub-domain moves it.
+ DomainEntity moved = domainManager.attachDomain(METALAKE, "Trading",
"Markets");
+ Assertions.assertEquals("Markets", moved.parent().get());
+ Assertions.assertEquals(0, domainManager.listSubdomains(METALAKE,
"Finance").length);
+ Assertions.assertArrayEquals(
+ new String[] {"Trading"}, domainManager.listSubdomains(METALAKE,
"Markets"));
+
+ // A domain cannot be its own parent.
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> domainManager.attachDomain(METALAKE, "Finance", "Finance"));
+ // The parent must be top-level.
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> domainManager.attachDomain(METALAKE, "Finance", "Trading"));
+ // A domain with sub-domains cannot be attached.
+ IllegalArgumentException e =
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> domainManager.attachDomain(METALAKE, "Markets", "Finance"));
+ Assertions.assertTrue(e.getMessage().contains("[Trading]"),
e.getMessage());
+ Assertions.assertFalse(domainManager.getDomain(METALAKE,
"Markets").parent().isPresent());
+ // A sub-domain cannot be moved under another sub-domain.
+ domainManager.createSubdomain(METALAKE, "Finance", "Desk", null, null);
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> domainManager.attachDomain(METALAKE, "Trading", "Desk"));
+ Assertions.assertEquals("Markets", domainManager.getDomain(METALAKE,
"Trading").parent().get());
+
+ Assertions.assertThrows(
+ NoSuchDomainException.class,
+ () -> domainManager.attachDomain(METALAKE, "Trading", "missing"));
+ Assertions.assertThrows(
+ NoSuchDomainException.class,
+ () -> domainManager.attachDomain(METALAKE, "missing", "Finance"));
+
+ DomainEntity detached = domainManager.detachDomain(METALAKE, "Trading");
+ Assertions.assertFalse(detached.parent().isPresent());
+ Assertions.assertEquals(0, domainManager.listSubdomains(METALAKE,
"Markets").length);
+ // Detaching a top-level domain changes nothing.
+ Assertions.assertEquals(detached, domainManager.detachDomain(METALAKE,
"Trading"));
+ Assertions.assertThrows(
+ NoSuchDomainException.class, () ->
domainManager.detachDomain(METALAKE, "missing"));
+ }
+
+ @Test
+ public void testAttachChecksStatus() {
+ domainManager.createDomain(METALAKE, "Finance", null, null);
+ domainManager.createDomain(METALAKE, "Trading", null, null);
+ domainManager.publishDomain(METALAKE, "Trading");
+
+ // A published domain needs a published parent.
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> domainManager.attachDomain(METALAKE, "Trading", "Finance"));
+ Assertions.assertFalse(domainManager.getDomain(METALAKE,
"Trading").parent().isPresent());
+
+ domainManager.publishDomain(METALAKE, "Finance");
+ Assertions.assertEquals(
+ "Finance", domainManager.attachDomain(METALAKE, "Trading",
"Finance").parent().get());
+
+ // A draft domain can be attached under a draft or a published parent.
+ domainManager.createDomain(METALAKE, "Draft", null, null);
+ domainManager.createDomain(METALAKE, "Desk", null, null);
+ Assertions.assertEquals(
+ "Draft", domainManager.attachDomain(METALAKE, "Desk",
"Draft").parent().get());
+ Assertions.assertEquals(
+ "Finance", domainManager.attachDomain(METALAKE, "Desk",
"Finance").parent().get());
+
+ // A published sub-domain cannot be moved under a draft parent.
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> domainManager.attachDomain(METALAKE, "Trading", "Draft"));
+ Assertions.assertEquals("Finance", domainManager.getDomain(METALAKE,
"Trading").parent().get());
+ Assertions.assertEquals(0, domainManager.listSubdomains(METALAKE,
"Draft").length);
+ }
+
+ @Test
+ public void testChecksSkipUnchangedState() {
+ DomainEntity draftParent = domain("Finance", null, Domain.Status.DRAFT);
+ DomainEntity publishedChild = domain("Trading", "Finance",
Domain.Status.PUBLISHED);
+ DomainEntity grandParent = domain("Markets", "Other", Domain.Status.DRAFT);
+
+ // Attaching a domain to its current parent is not checked.
+ DomainManager.checkAttach(
+ new LockedDomainState(publishedChild, draftParent,
Collections.emptyList()));
+ // A change of parent is.
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ DomainManager.checkAttach(
+ new LockedDomainState(
+ domain("Trading", null, Domain.Status.PUBLISHED),
+ draftParent,
+ Collections.emptyList())));
+ // So is a move from another parent.
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ DomainManager.checkAttach(
+ new LockedDomainState(
+ domain("Trading", "Markets", Domain.Status.PUBLISHED),
+ draftParent,
+ Collections.emptyList())));
+ // Two names that find the same row are the domain itself.
+ DomainEntity finance = domain("Finance", null, Domain.Status.DRAFT);
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ DomainManager.checkAttach(
+ new LockedDomainState(finance, finance,
Collections.emptyList())));
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ DomainManager.checkNewDomain(
+ new LockedDomainState(
+ domain("Desk", "Markets", Domain.Status.DRAFT),
+ grandParent,
+ Collections.emptyList())));
+
+ // Publishing a published domain and unpublishing a draft one are not
checked.
+ DomainManager.checkPublish(
+ new LockedDomainState(publishedChild, draftParent,
Collections.emptyList()));
+ List<DomainEntity> publishedSubDomains =
Collections.singletonList(publishedChild);
+ DomainManager.checkUnpublish(new LockedDomainState(draftParent, null,
publishedSubDomains));
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ DomainManager.checkUnpublish(
+ new LockedDomainState(
+ domain("Finance", null, Domain.Status.PUBLISHED), null,
publishedSubDomains)));
+ }
+
+ @Test
+ public void testCheckDomainMetadataObject() {
+ domainManager.createDomain(METALAKE, "Finance", null, null);
+ MetadataObjectUtil.checkMetadataObject(
+ METALAKE, MetadataObjects.of(null, "Finance",
MetadataObject.Type.DOMAIN));
+ Assertions.assertThrows(
+ NoSuchMetadataObjectException.class,
+ () ->
+ MetadataObjectUtil.checkMetadataObject(
+ METALAKE, MetadataObjects.of(null, "missing",
MetadataObject.Type.DOMAIN)));
+ }
+
+ private static DomainEntity domain(String name, String parent, Domain.Status
status) {
+ return DomainEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName(name)
+ .withNamespace(NamespaceUtil.ofDomain(METALAKE))
+ .withProperties(Collections.emptyMap())
+ .withStatus(status)
+ .withParent(parent)
+ .withAuditInfo(
+
AuditInfo.builder().withCreator("test").withCreateTime(Instant.now()).build())
+ .build();
+ }
+
+ private static Config mockConfig() {
+ Config config = Mockito.mock(Config.class);
+
+ Mockito.when(config.get(ENTITY_STORE)).thenReturn(RELATIONAL_ENTITY_STORE);
+
Mockito.when(config.get(ENTITY_RELATIONAL_STORE)).thenReturn(DEFAULT_ENTITY_RELATIONAL_STORE);
+ Mockito.when(config.get(ENTITY_RELATIONAL_JDBC_BACKEND_URL))
+ .thenReturn(
+ String.format(
+
"jdbc:h2:file:%s;DB_CLOSE_DELAY=-1;MODE=MYSQL;LOCK_TIMEOUT=30000", DB_DIR));
+
Mockito.when(config.get(ENTITY_RELATIONAL_JDBC_BACKEND_DRIVER)).thenReturn("org.h2.Driver");
+
Mockito.when(config.get(ENTITY_RELATIONAL_JDBC_BACKEND_MAX_CONNECTIONS)).thenReturn(100);
+
Mockito.when(config.get(ENTITY_RELATIONAL_JDBC_BACKEND_MAX_IDLE_CONNECTIONS)).thenReturn(10);
+
Mockito.when(config.get(ENTITY_RELATIONAL_JDBC_BACKEND_WAIT_MILLISECONDS)).thenReturn(1000L);
+
Mockito.when(config.get(STORE_TRANSACTION_MAX_SKEW_TIME)).thenReturn(1000L);
+ Mockito.when(config.get(STORE_DELETE_AFTER_TIME)).thenReturn(20 * 60 *
1000L);
+
Mockito.when(config.get(ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS)).thenReturn(3L);
+ Mockito.when(config.get(ENTITY_CHANGE_LOG_POLL_BATCH_SIZE))
+ .thenReturn(DEFAULT_ENTITY_CHANGE_LOG_POLL_BATCH_SIZE);
+ Mockito.when(config.get(ENTITY_CHANGE_LOG_REPLAY_WINDOW_SECS))
+ .thenReturn(DEFAULT_ENTITY_CHANGE_LOG_REPLAY_WINDOW_SECS);
+ Mockito.when(config.get(ENTITY_CHANGE_LOG_REPLAY_MAX_ENTRIES))
+ .thenReturn(DEFAULT_ENTITY_CHANGE_LOG_REPLAY_MAX_ENTRIES);
+ Mockito.when(config.get(ENTITY_CHANGE_LOG_RETENTION_SECS)).thenReturn(24 *
60 * 60L);
+
Mockito.when(config.get(ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS)).thenReturn(60
* 60L);
+ Mockito.when(config.get(VERSION_RETENTION_COUNT)).thenReturn(1L);
+ Mockito.when(config.get(Configs.CACHE_ENABLED)).thenReturn(true);
+ Mockito.when(config.get(Configs.CACHE_MAX_ENTRIES)).thenReturn(10_000);
+
Mockito.when(config.get(Configs.CACHE_EXPIRATION_TIME)).thenReturn(3_600_000L);
+ Mockito.when(config.get(Configs.CACHE_WEIGHER_ENABLED)).thenReturn(true);
+ Mockito.when(config.get(Configs.CACHE_STATS_ENABLED)).thenReturn(false);
+
Mockito.when(config.get(Configs.CACHE_IMPLEMENTATION)).thenReturn("caffeine");
+ Mockito.when(config.get(Configs.CACHE_LOCK_SEGMENTS)).thenReturn(16);
+
+ Mockito.doReturn(100000L).when(config).get(TREE_LOCK_MAX_NODE_IN_MEMORY);
+ Mockito.doReturn(1000L).when(config).get(TREE_LOCK_MIN_NODE_IN_MEMORY);
+ Mockito.doReturn(36000L).when(config).get(TREE_LOCK_CLEAN_INTERVAL);
+ return config;
+ }
+}
diff --git
a/core/src/test/java/org/apache/gravitino/domain/TestDomainManagerWithMockStore.java
b/core/src/test/java/org/apache/gravitino/domain/TestDomainManagerWithMockStore.java
new file mode 100644
index 0000000000..56e8a179bd
--- /dev/null
+++
b/core/src/test/java/org/apache/gravitino/domain/TestDomainManagerWithMockStore.java
@@ -0,0 +1,520 @@
+/*
+ * 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.domain;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+
+import java.io.IOException;
+import java.time.Instant;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Consumer;
+import java.util.function.Function;
+import org.apache.gravitino.Entity;
+import org.apache.gravitino.EntityAlreadyExistsException;
+import org.apache.gravitino.EntityStore;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.SupportsDomainOperations;
+import org.apache.gravitino.SupportsDomainOperations.LockedDomainState;
+import org.apache.gravitino.exceptions.DomainAlreadyExistsException;
+import org.apache.gravitino.exceptions.NoSuchDomainException;
+import org.apache.gravitino.exceptions.NoSuchEntityException;
+import org.apache.gravitino.exceptions.NoSuchMetalakeException;
+import org.apache.gravitino.exceptions.NonEmptyDomainException;
+import org.apache.gravitino.exceptions.NonEmptyEntityException;
+import org.apache.gravitino.exceptions.OptimisticLockException;
+import org.apache.gravitino.meta.AuditInfo;
+import org.apache.gravitino.meta.BaseMetalake;
+import org.apache.gravitino.meta.DomainEntity;
+import org.apache.gravitino.meta.SchemaVersion;
+import org.apache.gravitino.storage.IdGenerator;
+import org.apache.gravitino.utils.NameIdentifierUtil;
+import org.apache.gravitino.utils.NamespaceUtil;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
+import org.mockito.invocation.InvocationOnMock;
+
+/**
+ * Tests {@link DomainManager} against a mocked entity store: how store
exceptions map to domain
+ * exceptions, and which check each store operation gets.
+ */
+public class TestDomainManagerWithMockStore {
+
+ private static final String METALAKE = "metalake";
+ private static final NameIdentifier FINANCE =
NameIdentifierUtil.ofDomain(METALAKE, "Finance");
+ private static final NameIdentifier TRADING =
NameIdentifierUtil.ofDomain(METALAKE, "Trading");
+
+ private EntityStore entityStore;
+ private SupportsDomainOperations domainOperations;
+ private DomainManager domainManager;
+
+ @BeforeEach
+ public void setUp() throws IOException {
+ entityStore = Mockito.mock(EntityStore.class);
+ domainOperations = Mockito.mock(SupportsDomainOperations.class);
+ Mockito.when(entityStore.domainOperations()).thenReturn(domainOperations);
+ BaseMetalake metalake =
+ BaseMetalake.builder()
+ .withId(1L)
+ .withName(METALAKE)
+ .withVersion(SchemaVersion.V_0_1)
+ .withAuditInfo(auditInfo())
+ .build();
+ Mockito.when(
+ entityStore.get(
+ NameIdentifierUtil.ofMetalake(METALAKE),
+ Entity.EntityType.METALAKE,
+ BaseMetalake.class))
+ .thenReturn(metalake);
+
+ IdGenerator idGenerator = Mockito.mock(IdGenerator.class);
+ Mockito.when(idGenerator.nextId()).thenReturn(100L);
+ domainManager = new DomainManager(idGenerator, entityStore);
+ }
+
+ @Test
+ public void testMissingMetalake() throws IOException {
+ Mockito.when(
+ entityStore.get(
+ NameIdentifierUtil.ofMetalake("missing"),
+ Entity.EntityType.METALAKE,
+ BaseMetalake.class))
+ .thenThrow(new NoSuchEntityException("missing"));
+
+ Assertions.assertThrows(
+ NoSuchMetalakeException.class, () ->
domainManager.listDomainInfos("missing"));
+ Assertions.assertThrows(
+ NoSuchMetalakeException.class, () ->
domainManager.getDomain("missing", "Finance"));
+ Assertions.assertThrows(
+ NoSuchMetalakeException.class,
+ () -> domainManager.createDomain("missing", "Finance", null, null));
+ Assertions.assertThrows(
+ NoSuchMetalakeException.class,
+ () -> domainManager.createSubdomain("missing", "Finance", "Trading",
null, null));
+ Assertions.assertThrows(
+ NoSuchMetalakeException.class,
+ () -> domainManager.alterDomain("missing", "Finance",
DomainChange.updateComment("c")));
+ Assertions.assertThrows(
+ NoSuchMetalakeException.class, () ->
domainManager.deleteDomain("missing", "Finance"));
+ Assertions.assertThrows(
+ NoSuchMetalakeException.class, () ->
domainManager.publishDomain("missing", "Finance"));
+ Assertions.assertThrows(
+ NoSuchMetalakeException.class, () ->
domainManager.unpublishDomain("missing", "Finance"));
+ Assertions.assertThrows(
+ NoSuchMetalakeException.class,
+ () -> domainManager.listSubdomainInfos("missing", "Finance"));
+ Assertions.assertThrows(
+ NoSuchMetalakeException.class,
+ () -> domainManager.attachDomain("missing", "Trading", "Finance"));
+ Assertions.assertThrows(
+ NoSuchMetalakeException.class, () ->
domainManager.detachDomain("missing", "Trading"));
+ Mockito.verifyNoInteractions(domainOperations);
+ Mockito.verify(entityStore, Mockito.never()).update(any(), any(), any(),
any());
+ Mockito.verify(entityStore, Mockito.never()).delete(any(), any());
+ }
+
+ @Test
+ public void testListAndGetDomain() throws IOException {
+ Mockito.when(
+ entityStore.list(
+ NamespaceUtil.ofDomain(METALAKE), DomainEntity.class,
Entity.EntityType.DOMAIN))
+ .thenThrow(new IOException("list failed"));
+ assertIOExceptionWrapped(() -> domainManager.listDomainInfos(METALAKE));
+
+ Mockito.when(entityStore.get(FINANCE, Entity.EntityType.DOMAIN,
DomainEntity.class))
+ .thenThrow(new NoSuchEntityException("missing"));
+ Mockito.when(entityStore.get(TRADING, Entity.EntityType.DOMAIN,
DomainEntity.class))
+ .thenThrow(new IOException("get failed"));
+ Assertions.assertThrows(
+ NoSuchDomainException.class, () -> domainManager.getDomain(METALAKE,
"Finance"));
+ assertIOExceptionWrapped(() -> domainManager.getDomain(METALAKE,
"Trading"));
+ }
+
+ @Test
+ public void testCreateDomain() throws IOException {
+ ArgumentCaptor<DomainEntity> domainCaptor =
ArgumentCaptor.forClass(DomainEntity.class);
+ ArgumentCaptor<Consumer<LockedDomainState>> checkCaptor = checkCaptor();
+ DomainEntity created = domainManager.createDomain(METALAKE, "Finance",
"comment", null);
+ Mockito.verify(domainOperations).insertDomain(domainCaptor.capture(),
checkCaptor.capture());
+
+ DomainEntity inserted = domainCaptor.getValue();
+ Assertions.assertSame(inserted, created);
+ Assertions.assertEquals(100L, inserted.id());
+ Assertions.assertEquals(FINANCE, inserted.nameIdentifier());
+ Assertions.assertEquals("comment", inserted.comment());
+ Assertions.assertTrue(inserted.properties().isEmpty());
+ Assertions.assertEquals(Domain.Status.DRAFT, inserted.status());
+ Assertions.assertFalse(inserted.parent().isPresent());
+ assertIsNewDomainCheck(checkCaptor.getValue());
+
+ Mockito.doThrow(new EntityAlreadyExistsException("exists"))
+ .when(domainOperations)
+ .insertDomain(any(), any());
+ Assertions.assertThrows(
+ DomainAlreadyExistsException.class,
+ () -> domainManager.createDomain(METALAKE, "Finance", null, null));
+
+ // The metalake was dropped after it was checked.
+ Mockito.doThrow(new NoSuchEntityException("missing"))
+ .when(domainOperations)
+ .insertDomain(any(), any());
+ Assertions.assertThrows(
+ NoSuchMetalakeException.class,
+ () -> domainManager.createDomain(METALAKE, "Finance", null, null));
+
+ Mockito.doThrow(new IOException("insert failed"))
+ .when(domainOperations)
+ .insertDomain(any(), any());
+ assertIOExceptionWrapped(() -> domainManager.createDomain(METALAKE,
"Finance", null, null));
+ }
+
+ @Test
+ public void testCreateSubdomain() throws IOException {
+ ArgumentCaptor<DomainEntity> domainCaptor =
ArgumentCaptor.forClass(DomainEntity.class);
+ ArgumentCaptor<Consumer<LockedDomainState>> checkCaptor = checkCaptor();
+ domainManager.createSubdomain(METALAKE, "Finance", "Trading", null, null);
+ Mockito.verify(domainOperations).insertDomain(domainCaptor.capture(),
checkCaptor.capture());
+ Assertions.assertEquals(TRADING, domainCaptor.getValue().nameIdentifier());
+ Assertions.assertEquals("Finance", domainCaptor.getValue().parent().get());
+ assertIsNewDomainCheck(checkCaptor.getValue());
+
+ // The sub-domain is returned with the parent's stored name, which the
check reads under the
+ // parent's lock, even when the request names it differently.
+ Mockito.doAnswer(
+ invocation -> {
+ DomainEntity newDomain = invocation.getArgument(0);
+ Consumer<LockedDomainState> check = invocation.getArgument(1);
+ DomainEntity storedParent = domain("Finance", null,
Domain.Status.DRAFT);
+ check.accept(state(newDomain, storedParent));
+ return null;
+ })
+ .when(domainOperations)
+ .insertDomain(any(), any());
+ DomainEntity desk = domainManager.createSubdomain(METALAKE, "Finance ",
"Desk", null, null);
+ Assertions.assertEquals("Finance", desk.parent().get());
+ Assertions.assertEquals(NameIdentifierUtil.ofDomain(METALAKE, "Desk"),
desk.nameIdentifier());
+
+ // A check that throws reaches the caller unchanged.
+ Mockito.doAnswer(TestDomainManagerWithMockStore::runInsertCheck)
+ .when(domainOperations)
+ .insertDomain(any(), any());
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> domainManager.createSubdomain(METALAKE, "Trading", "Desk", null,
null));
+
+ Mockito.doThrow(new NoSuchEntityException("missing"))
+ .when(domainOperations)
+ .insertDomain(any(), any());
+ Assertions.assertThrows(
+ NoSuchDomainException.class,
+ () -> domainManager.createSubdomain(METALAKE, "missing", "Trading",
null, null));
+
+ Mockito.doThrow(new EntityAlreadyExistsException("exists"))
+ .when(domainOperations)
+ .insertDomain(any(), any());
+ Assertions.assertThrows(
+ DomainAlreadyExistsException.class,
+ () -> domainManager.createSubdomain(METALAKE, "Finance", "Trading",
null, null));
+
+ Mockito.doThrow(new IOException("insert failed"))
+ .when(domainOperations)
+ .insertDomain(any(), any());
+ assertIOExceptionWrapped(
+ () -> domainManager.createSubdomain(METALAKE, "Finance", "Trading",
null, null));
+
+ // A domain as its own parent, and a null parent, are refused before the
store is called.
+ Mockito.clearInvocations(domainOperations);
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> domainManager.createSubdomain(METALAKE, "Finance", "Finance",
null, null));
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> domainManager.createSubdomain(METALAKE, null, "Trading", null,
null));
+ Mockito.verifyNoInteractions(domainOperations);
+ }
+
+ @Test
+ public void testAlterDomain() throws IOException {
+ Mockito.when(entityStore.update(eq(FINANCE), eq(DomainEntity.class),
any(), any()))
+ .thenThrow(new NoSuchEntityException("missing"))
+ .thenThrow(new EntityAlreadyExistsException("exists"))
+ .thenThrow(new IOException("update failed"));
+
+ Assertions.assertThrows(
+ NoSuchDomainException.class,
+ () -> domainManager.alterDomain(METALAKE, "Finance",
DomainChange.updateComment("c")));
+ DomainAlreadyExistsException e =
+ Assertions.assertThrows(
+ DomainAlreadyExistsException.class,
+ () ->
+ domainManager.alterDomain(
+ METALAKE,
+ "Finance",
+ DomainChange.rename("Markets"),
+ DomainChange.rename("Trading")));
+ // The message names the last new name.
+ Assertions.assertTrue(e.getMessage().contains("Domain Trading"),
e.getMessage());
+ assertIOExceptionWrapped(
+ () -> domainManager.alterDomain(METALAKE, "Finance",
DomainChange.updateComment("c")));
+ }
+
+ @Test
+ public void testAlterDomainWithNullProperties() throws IOException {
+ DomainEntity stored =
+ domain("Finance", null,
Domain.Status.DRAFT).toBuilder().withProperties(null).build();
+ Mockito.when(entityStore.update(eq(FINANCE), eq(DomainEntity.class),
any(), any()))
+ .thenAnswer(
+ invocation -> {
+ Function<DomainEntity, DomainEntity> updater =
invocation.getArgument(3);
+ return updater.apply(stored);
+ });
+
+ DomainEntity updated =
+ domainManager.alterDomain(METALAKE, "Finance",
DomainChange.updateComment("new comment"));
+ Assertions.assertEquals("new comment", updated.comment());
+ Assertions.assertTrue(updated.properties().isEmpty());
+ Assertions.assertEquals(
+ Map.of("key", "value"),
+ domainManager
+ .alterDomain(METALAKE, "Finance", DomainChange.setProperty("key",
"value"))
+ .properties());
+ Assertions.assertTrue(
+ domainManager
+ .alterDomain(METALAKE, "Finance",
DomainChange.removeProperty("key"))
+ .properties()
+ .isEmpty());
+ Assertions.assertNull(stored.properties());
+ }
+
+ @Test
+ public void testAlterAndDeletePropagateOptimisticLockException() throws
IOException {
+ OptimisticLockException conflict = new OptimisticLockException("Concurrent
operation");
+ Mockito.when(entityStore.update(eq(FINANCE), eq(DomainEntity.class),
any(), any()))
+ .thenThrow(conflict);
+ Mockito.when(entityStore.delete(FINANCE,
Entity.EntityType.DOMAIN)).thenThrow(conflict);
+
+ Assertions.assertSame(
+ conflict,
+ Assertions.assertThrows(
+ OptimisticLockException.class,
+ () -> domainManager.alterDomain(METALAKE, "Finance",
DomainChange.updateComment("c"))));
+ Assertions.assertSame(
+ conflict,
+ Assertions.assertThrows(
+ OptimisticLockException.class, () ->
domainManager.deleteDomain(METALAKE, "Finance")));
+ }
+
+ @Test
+ public void testDeleteDomain() throws IOException {
+ Mockito.when(entityStore.delete(FINANCE, Entity.EntityType.DOMAIN))
+ .thenReturn(true)
+ .thenReturn(false)
+ .thenThrow(new NonEmptyEntityException("has sub-domains"))
+ .thenThrow(new IOException("delete failed"));
+
+ Assertions.assertTrue(domainManager.deleteDomain(METALAKE, "Finance"));
+ Assertions.assertFalse(domainManager.deleteDomain(METALAKE, "Finance"));
+ Assertions.assertThrows(
+ NonEmptyDomainException.class, () ->
domainManager.deleteDomain(METALAKE, "Finance"));
+ assertIOExceptionWrapped(() -> domainManager.deleteDomain(METALAKE,
"Finance"));
+ }
+
+ @Test
+ public void testPublishAndUnpublish() throws IOException {
+ ArgumentCaptor<Consumer<LockedDomainState>> publishCheck = checkCaptor();
+ domainManager.publishDomain(METALAKE, "Trading");
+ Mockito.verify(domainOperations)
+ .updateDomainStatusAndGet(
+ eq(TRADING), eq(Domain.Status.PUBLISHED), any(),
publishCheck.capture());
+ assertIsPublishCheck(publishCheck.getValue());
+
+ ArgumentCaptor<Consumer<LockedDomainState>> unpublishCheck = checkCaptor();
+ ArgumentCaptor<AuditInfo> auditCaptor =
ArgumentCaptor.forClass(AuditInfo.class);
+ domainManager.unpublishDomain(METALAKE, "Finance");
+ Mockito.verify(domainOperations)
+ .updateDomainStatusAndGet(
+ eq(FINANCE), eq(Domain.Status.DRAFT), auditCaptor.capture(),
unpublishCheck.capture());
+ assertIsUnpublishCheck(unpublishCheck.getValue());
+ Assertions.assertNotNull(auditCaptor.getValue().lastModifier());
+ Assertions.assertNotNull(auditCaptor.getValue().lastModifiedTime());
+
+ Mockito.when(domainOperations.updateDomainStatusAndGet(any(), any(),
any(), any()))
+ .thenThrow(new NoSuchEntityException("missing"))
+ .thenThrow(new NoSuchEntityException("missing"))
+ .thenThrow(new IOException("update failed"))
+ .thenThrow(new IOException("update failed"));
+ Assertions.assertThrows(
+ NoSuchDomainException.class, () ->
domainManager.publishDomain(METALAKE, "Finance"));
+ Assertions.assertThrows(
+ NoSuchDomainException.class, () ->
domainManager.unpublishDomain(METALAKE, "Finance"));
+ assertIOExceptionWrapped(() -> domainManager.publishDomain(METALAKE,
"Finance"));
+ assertIOExceptionWrapped(() -> domainManager.unpublishDomain(METALAKE,
"Finance"));
+ }
+
+ @Test
+ public void testListSubdomains() throws IOException {
+ DomainEntity trading = domain("Trading", "Finance", Domain.Status.DRAFT);
+ Mockito.when(domainOperations.listSubDomains(FINANCE))
+ .thenReturn(Collections.singletonList(trading))
+ .thenThrow(new NoSuchEntityException("missing"))
+ .thenThrow(new IOException("list failed"));
+
+ Assertions.assertArrayEquals(
+ new DomainEntity[] {trading},
domainManager.listSubdomainInfos(METALAKE, "Finance"));
+ Assertions.assertThrows(
+ NoSuchDomainException.class, () ->
domainManager.listSubdomainInfos(METALAKE, "Finance"));
+ assertIOExceptionWrapped(() -> domainManager.listSubdomainInfos(METALAKE,
"Finance"));
+ }
+
+ @Test
+ public void testAttachDomain() throws IOException {
+ ArgumentCaptor<Consumer<LockedDomainState>> checkCaptor = checkCaptor();
+ ArgumentCaptor<AuditInfo> auditCaptor =
ArgumentCaptor.forClass(AuditInfo.class);
+ domainManager.attachDomain(METALAKE, "Trading", "Finance");
+ Mockito.verify(domainOperations)
+ .setDomainParentAndGet(
+ eq(TRADING), eq(FINANCE), auditCaptor.capture(),
checkCaptor.capture());
+ assertIsAttachCheck(checkCaptor.getValue());
+ // A new edge takes the creator, a moved one the modifier, so both are set.
+ Assertions.assertNotNull(auditCaptor.getValue().creator());
+ Assertions.assertNotNull(auditCaptor.getValue().lastModifier());
+
+ Mockito.when(domainOperations.setDomainParentAndGet(any(), any(), any(),
any()))
+ .thenThrow(new NoSuchEntityException("missing"))
+ .thenThrow(new IOException("update failed"));
+ Assertions.assertThrows(
+ NoSuchDomainException.class,
+ () -> domainManager.attachDomain(METALAKE, "Trading", "Finance"));
+ assertIOExceptionWrapped(() -> domainManager.attachDomain(METALAKE,
"Trading", "Finance"));
+
+ // A domain as its own parent, and a null parent, are refused before the
store is called.
+ Mockito.clearInvocations(domainOperations);
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> domainManager.attachDomain(METALAKE, "Finance", "Finance"));
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> domainManager.attachDomain(METALAKE, "Trading", null));
+ Mockito.verifyNoInteractions(domainOperations);
+ }
+
+ @Test
+ public void testDetachDomain() throws IOException {
+ DomainEntity trading = domain("Trading", null, Domain.Status.DRAFT);
+ Mockito.when(domainOperations.removeDomainParentAndGet(eq(TRADING), any()))
+ .thenReturn(trading)
+ .thenThrow(new NoSuchEntityException("missing"))
+ .thenThrow(new IOException("update failed"));
+
+ Assertions.assertSame(trading, domainManager.detachDomain(METALAKE,
"Trading"));
+ Assertions.assertThrows(
+ NoSuchDomainException.class, () ->
domainManager.detachDomain(METALAKE, "Trading"));
+ assertIOExceptionWrapped(() -> domainManager.detachDomain(METALAKE,
"Trading"));
+ }
+
+ // The insert check refuses a parent that is itself a sub-domain.
+ private static void assertIsNewDomainCheck(Consumer<LockedDomainState>
check) {
+ DomainEntity desk = domain("Desk", "Trading", Domain.Status.DRAFT);
+ check.accept(state(desk, domain("Trading", null, Domain.Status.DRAFT)));
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> check.accept(state(desk, domain("Trading", "Finance",
Domain.Status.DRAFT))));
+ }
+
+ // The publish check refuses a draft parent, which the unpublish check would
accept.
+ private static void assertIsPublishCheck(Consumer<LockedDomainState> check) {
+ DomainEntity trading = domain("Trading", "Finance", Domain.Status.DRAFT);
+ check.accept(state(trading, domain("Finance", null,
Domain.Status.PUBLISHED)));
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> check.accept(state(trading, domain("Finance", null,
Domain.Status.DRAFT))));
+ }
+
+ // The unpublish check refuses published sub-domains, which the publish
check would accept.
+ private static void assertIsUnpublishCheck(Consumer<LockedDomainState>
check) {
+ DomainEntity finance = domain("Finance", null, Domain.Status.PUBLISHED);
+ check.accept(
+ new LockedDomainState(
+ finance, null, List.of(domain("Trading", "Finance",
Domain.Status.DRAFT))));
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ check.accept(
+ new LockedDomainState(
+ finance,
+ null,
+ List.of(domain("Trading", "Finance",
Domain.Status.PUBLISHED)))));
+ }
+
+ // The attach check refuses a domain with sub-domains, which the insert
check would accept.
+ private static void assertIsAttachCheck(Consumer<LockedDomainState> check) {
+ DomainEntity trading = domain("Trading", null, Domain.Status.DRAFT);
+ DomainEntity finance = domain("Finance", null, Domain.Status.DRAFT);
+ check.accept(state(trading, finance));
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ check.accept(
+ new LockedDomainState(
+ trading, finance, List.of(domain("Desk", "Trading",
Domain.Status.DRAFT)))));
+ }
+
+ private static Object runInsertCheck(InvocationOnMock invocation) {
+ DomainEntity domain = invocation.getArgument(0);
+ Consumer<LockedDomainState> check = invocation.getArgument(1);
+ // The parent of the new domain is itself a sub-domain.
+ check.accept(state(domain, domain(domain.parent().get(), "Finance",
Domain.Status.PUBLISHED)));
+ return null;
+ }
+
+ @SuppressWarnings("unchecked")
+ private static ArgumentCaptor<Consumer<LockedDomainState>> checkCaptor() {
+ return ArgumentCaptor.forClass(Consumer.class);
+ }
+
+ private static void assertIOExceptionWrapped(Runnable call) {
+ RuntimeException e = Assertions.assertThrows(RuntimeException.class,
call::run);
+ Assertions.assertInstanceOf(IOException.class, e.getCause());
+ }
+
+ private static LockedDomainState state(DomainEntity domain, DomainEntity
parent) {
+ return new LockedDomainState(domain, parent, Collections.emptyList());
+ }
+
+ private static DomainEntity domain(String name, String parent, Domain.Status
status) {
+ return DomainEntity.builder()
+ .withId((long) name.hashCode())
+ .withName(name)
+ .withNamespace(NamespaceUtil.ofDomain(METALAKE))
+ .withProperties(Collections.emptyMap())
+ .withStatus(status)
+ .withParent(parent)
+ .withAuditInfo(auditInfo())
+ .build();
+ }
+
+ private static AuditInfo auditInfo() {
+ return
AuditInfo.builder().withCreator("test").withCreateTime(Instant.now()).build();
+ }
+}
