This is an automated email from the ASF dual-hosted git repository.
yuqi1129 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new 2fd8dd9592 [#13853] fix(core): Preserve concurrent catalogs on
non-forced metalake drops (#14317)
2fd8dd9592 is described below
commit 2fd8dd95928bdbd978a0d11f118153f670db9358
Author: Qi Yu <[email protected]>
AuthorDate: Wed Oct 28 17:59:24 2026 +0800
[#13853] fix(core): Preserve concurrent catalogs on non-forced metalake
drops (#14317)
### What changes were proposed in this pull request?
Pass `force` to the metalake store delete so non-forced drops use the
existing transactional emptiness check. Translate a store-level
non-empty rejection to `NonEmptyMetalakeException`, preserving its cause
and the existing REST error contract.
Add manager-level regression coverage for a catalog committed after the
initial empty listing, existing catalogs, and repeated deletion of an
empty metalake, plus REST error coverage.
### Why are the changes needed?
Another server can create a catalog between the manager's empty listing
and deletion. The unconditional cascade flag previously allowed a
non-forced request to delete that catalog. The non-cascade store path
locks the metalake row and rolls back when catalogs exist.
Part of #13853. This PR addresses only the non-forced metalake drop
task; the epic remains open.
### Does this PR introduce _any_ user-facing change?
A non-forced drop refuses to delete a metalake when a catalog appears
after the initial check, using the existing non-empty conflict response.
Forced drops retain their existing behavior. No API, configuration, or
schema changes.
### How was this patch tested?
- 68 targeted tests passed, including the new manager cases on H2, MySQL
8 and PostgreSQL 13, existing metalake manager/dispatcher/event tests,
H2 metalake persistence tests, and REST tests.
- Spotless checks passed. No unfiltered core suite was run.
- Reverting only the cascade flag makes the new H2 race test fail;
restoring the committed code passes all three manager H2 cases again.
- A fresh-context review found no Blocker or Important issue.
---
.../apache/gravitino/metalake/MetalakeManager.java | 11 +-
.../metalake/TestMetalakeManagerRelational.java | 137 +++++++++++++++++++++
.../server/web/rest/TestMetalakeOperations.java | 22 ++++
3 files changed, 169 insertions(+), 1 deletion(-)
diff --git
a/core/src/main/java/org/apache/gravitino/metalake/MetalakeManager.java
b/core/src/main/java/org/apache/gravitino/metalake/MetalakeManager.java
index 59230a6a97..ce468b9f4e 100644
--- a/core/src/main/java/org/apache/gravitino/metalake/MetalakeManager.java
+++ b/core/src/main/java/org/apache/gravitino/metalake/MetalakeManager.java
@@ -396,7 +396,16 @@ public class MetalakeManager implements
MetalakeDispatcher, Closeable {
"Metalake %s has catalogs, please drop them first or use
force option", ident);
}
- return store.delete(ident, EntityType.METALAKE, true);
+ // The list above is only an early check: another server may
create a catalog
+ // before this delete. Let the store recheck under its metalake
row lock.
+ return store.delete(ident, EntityType.METALAKE, force);
+ } catch (NonEmptyEntityException e) {
+ NonEmptyMetalakeException failure =
+ new NonEmptyMetalakeException(
+ "Metalake %s has catalogs, please drop them first or use
force option",
+ ident);
+ failure.initCause(e);
+ throw failure;
} catch (NoSuchMetalakeException | NoSuchEntityException e) {
// Another server may have completed the drop after the initial
existence check.
// Dropping an already-removed metalake remains an idempotent
false result.
diff --git
a/core/src/test/java/org/apache/gravitino/metalake/TestMetalakeManagerRelational.java
b/core/src/test/java/org/apache/gravitino/metalake/TestMetalakeManagerRelational.java
new file mode 100644
index 0000000000..fb183cd2ba
--- /dev/null
+++
b/core/src/test/java/org/apache/gravitino/metalake/TestMetalakeManagerRelational.java
@@ -0,0 +1,137 @@
+/*
+ * 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.metalake;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.List;
+import org.apache.commons.lang3.reflect.FieldUtils;
+import org.apache.gravitino.Config;
+import org.apache.gravitino.Configs;
+import org.apache.gravitino.Entity.EntityType;
+import org.apache.gravitino.GravitinoEnv;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.Namespace;
+import org.apache.gravitino.cache.NoOpsCache;
+import org.apache.gravitino.exceptions.NonEmptyEntityException;
+import org.apache.gravitino.exceptions.NonEmptyMetalakeException;
+import org.apache.gravitino.lock.LockManager;
+import org.apache.gravitino.meta.BaseMetalake;
+import org.apache.gravitino.meta.CatalogEntity;
+import org.apache.gravitino.storage.RandomIdGenerator;
+import org.apache.gravitino.storage.relational.RelationalEntityStore;
+import org.apache.gravitino.storage.relational.TestJDBCBackend;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.mockito.Mockito;
+
+/** Tests metalake deletion through the manager against each relational
backend. */
+public class TestMetalakeManagerRelational extends TestJDBCBackend {
+
+ private static final String METALAKE = "metalake_drop_race";
+
+ private LockManager previousLockManager;
+ private RelationalEntityStore store;
+ private MetalakeManager manager;
+
+ @BeforeEach
+ void setUpManager() throws IllegalAccessException {
+ Config config = Mockito.mock(Config.class);
+
Mockito.when(config.get(Configs.TREE_LOCK_MAX_NODE_IN_MEMORY)).thenReturn(100000L);
+
Mockito.when(config.get(Configs.TREE_LOCK_MIN_NODE_IN_MEMORY)).thenReturn(1000L);
+
Mockito.when(config.get(Configs.TREE_LOCK_CLEAN_INTERVAL)).thenReturn(36000L);
+ previousLockManager = GravitinoEnv.getInstance().lockManager();
+ FieldUtils.writeField(GravitinoEnv.getInstance(), "lockManager", new
LockManager(config), true);
+
+ store = Mockito.spy(new RelationalEntityStore());
+ FieldUtils.writeField(store, "backend", backend, true);
+ FieldUtils.writeField(store, "cache", new NoOpsCache(config), true);
+ manager = new MetalakeManager(store, new RandomIdGenerator());
+ }
+
+ @AfterEach
+ void restoreLockManager() throws IllegalAccessException {
+ FieldUtils.writeField(GravitinoEnv.getInstance(), "lockManager",
previousLockManager, true);
+ }
+
+ @TestTemplate
+ void testNonForceDropPreservesCatalogCreatedAfterEmptyCheck() throws
Exception {
+ BaseMetalake metalake = createAndInsertMakeLake(METALAKE);
+ NameIdentifier ident = metalake.nameIdentifier();
+ manager.disableMetalake(ident);
+ CatalogEntity catalog =
+ createCatalog(
+ RandomIdGenerator.INSTANCE.nextId(),
+ Namespace.of(METALAKE),
+ "late_catalog",
+ AUDIT_INFO);
+
+ Mockito.doAnswer(
+ invocation -> {
+ List<?> observed = (List<?>) invocation.callRealMethod();
+ assertTrue(observed.isEmpty());
+ // Model a second server committing after the manager's empty
snapshot. It does
+ // not share the first server's tree lock or entity cache.
+ backend.insert(catalog, false);
+ return observed;
+ })
+ .when(store)
+ .list(Namespace.of(METALAKE), CatalogEntity.class, EntityType.CATALOG);
+
+ NonEmptyMetalakeException failure =
+ assertThrows(NonEmptyMetalakeException.class, () ->
manager.dropMetalake(ident, false));
+ assertTrue(failure.getMessage().contains(METALAKE));
+ assertTrue(failure.getMessage().contains("force"));
+ assertInstanceOf(NonEmptyEntityException.class, failure.getCause());
+ assertEquals(metalake.id(), backend.<BaseMetalake>get(ident,
EntityType.METALAKE).id());
+ assertEquals(catalog, backend.<CatalogEntity>get(catalog.nameIdentifier(),
EntityType.CATALOG));
+ assertFalse(MetalakeManager.metalakeInUse(store, ident));
+ }
+
+ @TestTemplate
+ void testNonForceDropRejectsExistingCatalog() throws Exception {
+ BaseMetalake metalake = createAndInsertMakeLake(METALAKE);
+ manager.disableMetalake(metalake.nameIdentifier());
+ CatalogEntity catalog = createAndInsertCatalog(METALAKE,
"existing_catalog");
+
+ assertThrows(
+ NonEmptyMetalakeException.class,
+ () -> manager.dropMetalake(metalake.nameIdentifier(), false));
+ assertEquals(catalog, backend.<CatalogEntity>get(catalog.nameIdentifier(),
EntityType.CATALOG));
+ assertEquals(
+ metalake.id(),
+ backend.<BaseMetalake>get(metalake.nameIdentifier(),
EntityType.METALAKE).id());
+ }
+
+ @TestTemplate
+ void testNonForceDropEmptyMetalakeAndRepeat() throws Exception {
+ BaseMetalake metalake = createAndInsertMakeLake(METALAKE);
+ NameIdentifier ident = metalake.nameIdentifier();
+ manager.disableMetalake(ident);
+
+ assertTrue(manager.dropMetalake(ident, false));
+ assertFalse(backend.exists(ident, EntityType.METALAKE));
+ assertFalse(manager.dropMetalake(ident, false));
+ }
+}
diff --git
a/server/src/test/java/org/apache/gravitino/server/web/rest/TestMetalakeOperations.java
b/server/src/test/java/org/apache/gravitino/server/web/rest/TestMetalakeOperations.java
index 6eeb2b7669..be125c3f34 100644
---
a/server/src/test/java/org/apache/gravitino/server/web/rest/TestMetalakeOperations.java
+++
b/server/src/test/java/org/apache/gravitino/server/web/rest/TestMetalakeOperations.java
@@ -62,6 +62,7 @@ import
org.apache.gravitino.dto.responses.MetalakeListResponse;
import org.apache.gravitino.dto.responses.MetalakeResponse;
import org.apache.gravitino.exceptions.MetalakeAlreadyExistsException;
import org.apache.gravitino.exceptions.NoSuchMetalakeException;
+import org.apache.gravitino.exceptions.NonEmptyMetalakeException;
import org.apache.gravitino.lock.LockManager;
import org.apache.gravitino.meta.AuditInfo;
import org.apache.gravitino.meta.BaseMetalake;
@@ -550,6 +551,27 @@ public class TestMetalakeOperations extends
BaseOperationsTest {
errorResponse.getMessage().contains("Failed to operate object [test]
operation [DROP]"));
}
+ @Test
+ public void testDropNonEmptyMetalake() {
+ doThrow(
+ new NonEmptyMetalakeException(
+ "Metalake test has catalogs, please drop them first or use
force option"))
+ .when(metalakeManager)
+ .dropMetalake(any(), anyBoolean());
+
+ Response response =
+ target("/metalakes/test")
+ .request(MediaType.APPLICATION_JSON_TYPE)
+ .accept("application/vnd.gravitino.v1+json")
+ .delete();
+
+ Assertions.assertEquals(Response.Status.CONFLICT.getStatusCode(),
response.getStatus());
+ ErrorResponse error = response.readEntity(ErrorResponse.class);
+ Assertions.assertEquals(ErrorConstants.NON_EMPTY_CODE, error.getCode());
+ Assertions.assertEquals(NonEmptyMetalakeException.class.getSimpleName(),
error.getType());
+ Assertions.assertTrue(error.getMessage().contains("Metalake test has
catalogs"));
+ }
+
@Test
public void testExceptionMapper() {
Response resp = target("/test/jsonProcessingException").request().get();
