This is an automated email from the ASF dual-hosted git repository.
yuqi1129 pushed a commit to branch branch-1.3
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/branch-1.3 by this push:
new cdba48b380 [Cherry-pick to branch-1.3] [#13296] fix(core): close
catalogs when force-dropping a metalake (#13297) (#13298)
cdba48b380 is described below
commit cdba48b3800c374ca258ef3a8491149e8fc73f1a
Author: Qi Yu <[email protected]>
AuthorDate: Fri Sep 18 14:04:03 2026 +0800
[Cherry-pick to branch-1.3] [#13296] fix(core): close catalogs when
force-dropping a metalake (#13297) (#13298)
### What changes were proposed in this pull request?
Backport to `branch-1.3` of the catalog-drop part of #12420 plus the
regression test from #13297:
- `MetalakeManager.dropMetalake(force=true)` now force-drops each child
catalog through `CatalogManager.dropCatalog` before deleting the
metalake entity, so every catalog is evicted from the catalog cache and
closed on the same path as an individual catalog drop. A disabled
metalake is briefly re-enabled first because `dropCatalog` requires
`metalake-in-use=true`.
- `MetalakeManager` gains a constructor taking `CatalogManager`;
`GravitinoEnv` creates `CatalogManager` before `MetalakeManager` and
passes it in. The 2-arg constructor is kept.
- Tests: `testForceDropMetalakeClosesCachedCatalogs` (real
`CatalogManager`, fails without the fix) and
`testForceDropMetalakeAfterDisableDropsLeftoverCatalogs` (disable then
force-drop).
The rest of #12420 (secrets) is not backported.
### Why are the changes needed?
On `branch-1.3`, dropping a metalake only deletes the metalake entity;
the cached catalog instances and their connection pools stay alive until
cache expiry or server restart.
Fix: #13296
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
`TestMetalakeManager` (8/8), `TestCatalogManager` (27/27),
`TestMetalakeNormalizeDispatcher` locally on `branch-1.3`.
---
.../java/org/apache/gravitino/GravitinoEnv.java | 16 ++--
.../apache/gravitino/metalake/MetalakeManager.java | 57 ++++++++++++
.../gravitino/metalake/TestMetalakeManager.java | 100 +++++++++++++++++++++
3 files changed, 166 insertions(+), 7 deletions(-)
diff --git a/core/src/main/java/org/apache/gravitino/GravitinoEnv.java
b/core/src/main/java/org/apache/gravitino/GravitinoEnv.java
index 1f3cbefb94..c0222c2089 100644
--- a/core/src/main/java/org/apache/gravitino/GravitinoEnv.java
+++ b/core/src/main/java/org/apache/gravitino/GravitinoEnv.java
@@ -746,22 +746,24 @@ public class GravitinoEnv {
// Tree lock
this.lockManager = new LockManager(config);
+ // Create and initialize Catalog related modules first so MetalakeManager
can force-drop
+ // child catalogs through CatalogManager. The operation chain is:
+ // CatalogEventDispatcher -> CatalogNormalizeDispatcher ->
CatalogHookDispatcher ->
+ // CatalogManager
+ // CatalogManager registers its own change-log listener with the entity
store (when the store
+ // supports it), so no poller wiring is needed here.
+ this.catalogManager = new CatalogManager(config, entityStore, idGenerator);
+
// Create and initialize metalake related modules, the operation chain is:
// MetalakeEventDispatcher -> MetalakeNormalizeDispatcher ->
MetalakeHookDispatcher ->
// MetalakeManager
- this.metalakeManager = new MetalakeManager(entityStore, idGenerator);
+ this.metalakeManager = new MetalakeManager(entityStore, idGenerator,
catalogManager);
this.internalMetalakeDispatcher = new
MetalakeNormalizeDispatcher(metalakeManager);
MetalakeHookDispatcher metalakeHookDispatcher = new
MetalakeHookDispatcher(metalakeManager);
MetalakeNormalizeDispatcher metalakeNormalizeDispatcher =
new MetalakeNormalizeDispatcher(metalakeHookDispatcher);
this.metalakeDispatcher = new MetalakeEventDispatcher(eventBus,
metalakeNormalizeDispatcher);
- // Create and initialize Catalog related modules, the operation chain is:
- // CatalogEventDispatcher -> CatalogNormalizeDispatcher ->
CatalogHookDispatcher ->
- // CatalogManager
- // CatalogManager registers its own change-log listener with the entity
store (when the store
- // supports it), so no poller wiring is needed here.
- this.catalogManager = new CatalogManager(config, entityStore, idGenerator);
this.internalCatalogDispatcher = new
CatalogNormalizeDispatcher(catalogManager);
CatalogHookDispatcher catalogHookDispatcher = new
CatalogHookDispatcher(catalogManager);
CatalogNormalizeDispatcher catalogNormalizeDispatcher =
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 1bc5dfd13b..b2a01b6a06 100644
--- a/core/src/main/java/org/apache/gravitino/metalake/MetalakeManager.java
+++ b/core/src/main/java/org/apache/gravitino/metalake/MetalakeManager.java
@@ -37,6 +37,7 @@ import org.apache.gravitino.GravitinoEnv;
import org.apache.gravitino.MetalakeChange;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.Namespace;
+import org.apache.gravitino.catalog.CatalogManager;
import org.apache.gravitino.exceptions.AlreadyExistsException;
import org.apache.gravitino.exceptions.MetalakeAlreadyExistsException;
import org.apache.gravitino.exceptions.MetalakeInUseException;
@@ -69,6 +70,8 @@ public class MetalakeManager implements MetalakeDispatcher,
Closeable {
private final IdGenerator idGenerator;
+ private final CatalogManager catalogManager;
+
@Override
public void close() {
// do nothing
@@ -81,8 +84,24 @@ public class MetalakeManager implements MetalakeDispatcher,
Closeable {
* @param idGenerator The IdGenerator to use for generating Metalake
identifiers.
*/
public MetalakeManager(EntityStore store, IdGenerator idGenerator) {
+ this(store, idGenerator, null);
+ }
+
+ /**
+ * Constructs a MetalakeManager instance.
+ *
+ * @param store The EntityStore to use for managing Metalakes.
+ * @param idGenerator The IdGenerator to use for generating Metalake
identifiers.
+ * @param catalogManager Used on force-drop to drop child catalogs via {@link
+ * CatalogManager#dropCatalog}, which evicts them from the catalog cache
and closes their
+ * resources (for example JDBC connection pools); may be null in tests
that do not exercise
+ * force-drop with catalogs.
+ */
+ public MetalakeManager(
+ EntityStore store, IdGenerator idGenerator, CatalogManager
catalogManager) {
this.store = store;
this.idGenerator = idGenerator;
+ this.catalogManager = catalogManager;
// preload all metalakes and put them into cache, this is useful when user
load schema/table
// directly without list/get metalake first.
@@ -340,6 +359,14 @@ public class MetalakeManager implements
MetalakeDispatcher, Closeable {
@Override
public boolean dropMetalake(NameIdentifier ident, boolean force)
throws NonEmptyEntityException, MetalakeInUseException {
+ // Force-drop child catalogs through CatalogManager.dropCatalog so each
one is evicted from
+ // the catalog cache and closed. Deleting only the metalake entity leaves
the cached catalog
+ // instances (and their source connections) alive until cache expiry. Do
this before the
+ // metalake root lock to avoid nesting tree locks.
+ if (force) {
+ dropCatalogsUnderMetalake(ident);
+ }
+
return TreeLockUtils.doWithRootTreeLock(
LockType.WRITE,
() -> {
@@ -367,6 +394,36 @@ public class MetalakeManager implements
MetalakeDispatcher, Closeable {
});
}
+ /**
+ * Force-drop child catalogs via {@link CatalogManager#dropCatalog} so they
are removed from the
+ * catalog cache and closed on the same path as a normal catalog force-drop.
+ *
+ * <p>Callers typically {@code disableMetalake} before force-drop. {@link
+ * CatalogManager#dropCatalog} requires catalog {@code
metalake-in-use=true}, so a disabled
+ * metalake is briefly re-enabled for child cleanup. The metalake entity is
deleted immediately
+ * afterward, so the temporary enable is not restored.
+ */
+ private void dropCatalogsUnderMetalake(NameIdentifier metalakeIdent) {
+ if (catalogManager == null) {
+ return;
+ }
+ try {
+ if (!metalakeInUse(store, metalakeIdent)) {
+ enableMetalake(metalakeIdent);
+ }
+ List<CatalogEntity> catalogs =
+ store.list(Namespace.of(metalakeIdent.name()), CatalogEntity.class,
EntityType.CATALOG);
+ for (CatalogEntity catalog : catalogs) {
+ catalogManager.dropCatalog(
+ NameIdentifier.of(metalakeIdent.name(), catalog.name()), true /*
force */);
+ }
+ } catch (NoSuchMetalakeException e) {
+ // Metalake is already gone; dropMetalake will return false.
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
@Override
public void enableMetalake(NameIdentifier ident) throws
NoSuchMetalakeException {
TreeLockUtils.doWithTreeLock(
diff --git
a/core/src/test/java/org/apache/gravitino/metalake/TestMetalakeManager.java
b/core/src/test/java/org/apache/gravitino/metalake/TestMetalakeManager.java
index c5be0774ea..d42786c8ee 100644
--- a/core/src/test/java/org/apache/gravitino/metalake/TestMetalakeManager.java
+++ b/core/src/test/java/org/apache/gravitino/metalake/TestMetalakeManager.java
@@ -21,29 +21,40 @@ package org.apache.gravitino.metalake;
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.awaitility.Awaitility.await;
import static org.mockito.Mockito.doReturn;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Sets;
import java.io.IOException;
+import java.time.Instant;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.TimeUnit;
import org.apache.commons.lang3.reflect.FieldUtils;
+import org.apache.gravitino.Catalog;
import org.apache.gravitino.Config;
+import org.apache.gravitino.Configs;
+import org.apache.gravitino.Entity;
import org.apache.gravitino.EntityStore;
import org.apache.gravitino.GravitinoEnv;
import org.apache.gravitino.MetalakeChange;
import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.Namespace;
import org.apache.gravitino.StringIdentifier;
import org.apache.gravitino.UserPrincipal;
import org.apache.gravitino.auth.AuthConstants;
+import org.apache.gravitino.catalog.CatalogManager;
import org.apache.gravitino.exceptions.MetalakeAlreadyExistsException;
import org.apache.gravitino.exceptions.NoSuchMetalakeException;
import org.apache.gravitino.lock.LockManager;
+import org.apache.gravitino.meta.AuditInfo;
import org.apache.gravitino.meta.BaseMetalake;
+import org.apache.gravitino.meta.CatalogEntity;
import org.apache.gravitino.storage.RandomIdGenerator;
import org.apache.gravitino.storage.memory.TestMemoryEntityStore;
+import
org.apache.gravitino.storage.memory.TestMemoryEntityStore.InMemoryEntityStore;
import org.apache.gravitino.utils.PrincipalUtils;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.Assertions;
@@ -239,6 +250,95 @@ public class TestMetalakeManager {
metalakeManager.dropMetalake(ident3, true);
}
+ @Test
+ public void testForceDropMetalakeClosesCachedCatalogs() throws Exception {
+ // Dropping a metalake must release the resources of its catalogs (for
example JDBC
+ // connection pools) exactly as dropping each catalog does, instead of
leaving the cached
+ // catalog instances alive until cache expiry.
+ Config catalogConfig = new Config(false) {};
+ catalogConfig.set(Configs.CATALOG_LOAD_ISOLATED, false);
+ InMemoryEntityStore store = new InMemoryEntityStore();
+ store.initialize(catalogConfig);
+ CatalogManager catalogManager =
+ new CatalogManager(catalogConfig, store, new RandomIdGenerator());
+ MetalakeManager manager = new MetalakeManager(store, new
RandomIdGenerator(), catalogManager);
+
+ NameIdentifier metalakeIdent =
NameIdentifier.of("force_drop_closes_catalogs_ml");
+ manager.createMetalake(metalakeIdent, "comment", ImmutableMap.of());
+ NameIdentifier catalogIdent = NameIdentifier.of(metalakeIdent.name(),
"cached_catalog");
+ catalogManager.createCatalog(
+ catalogIdent,
+ Catalog.Type.RELATIONAL,
+ "test",
+ "comment",
+ ImmutableMap.of(
+ "provider", "test", "key1", "value1", "key2", "value2", "key5-1",
"value3"));
+ // createCatalog caches the wrapper; loadCatalog keeps it warm the same
way a schema listing
+ // against the catalog would.
+ catalogManager.loadCatalog(catalogIdent);
+ CatalogManager.CatalogWrapper wrapper =
+ catalogManager.getCatalogCache().getIfPresent(catalogIdent);
+ Assertions.assertNotNull(wrapper);
+ Assertions.assertNotNull(wrapper.catalog());
+
+ Assertions.assertTrue(manager.dropMetalake(metalakeIdent, true));
+
+
Assertions.assertNull(catalogManager.getCatalogCache().getIfPresent(catalogIdent));
+ Assertions.assertFalse(store.exists(catalogIdent,
Entity.EntityType.CATALOG));
+ // The cache removal listener retires the wrapper asynchronously; once
cleaned up, the wrapper
+ // drops its catalog reference.
+ await().atMost(10, TimeUnit.SECONDS).until(() -> wrapper.catalog() ==
null);
+
+ catalogManager.close();
+ store.close();
+ }
+
+ @Test
+ public void testForceDropMetalakeAfterDisableDropsLeftoverCatalogs() throws
Exception {
+ // Mirrors IT tearDown: disableMetalake then dropMetalake(force=true)
while a catalog entity
+ // may still remain. CatalogManager.dropCatalog requires metalake-in-use
on the catalog.
+ InMemoryEntityStore store = new InMemoryEntityStore();
+ store.initialize(config);
+ CatalogManager catalogManager = Mockito.mock(CatalogManager.class);
+ Mockito.when(catalogManager.dropCatalog(Mockito.any(NameIdentifier.class),
Mockito.eq(true)))
+ .thenReturn(true);
+ // enableMetalake/disableMetalake propagate metalake-in-use to catalogs
through
+ // GravitinoEnv.catalogManager(), so point it at the mock for the duration
of this test.
+ Mockito.doNothing()
+ .when(catalogManager)
+ .setMetalakeInUseStatus(Mockito.any(NameIdentifier.class),
Mockito.anyBoolean());
+ Object previousCatalogManager =
+ FieldUtils.readField(GravitinoEnv.getInstance(), "catalogManager",
true);
+ FieldUtils.writeField(GravitinoEnv.getInstance(), "catalogManager",
catalogManager, true);
+
+ try {
+ MetalakeManager manager = new MetalakeManager(store, new
RandomIdGenerator(), catalogManager);
+
+ NameIdentifier metalakeIdent =
NameIdentifier.of("force_drop_leftover_ml");
+ manager.createMetalake(metalakeIdent, "comment", ImmutableMap.of());
+ store.put(
+ CatalogEntity.builder()
+ .withId(99L)
+ .withName("leftover_catalog")
+ .withNamespace(Namespace.of(metalakeIdent.name()))
+ .withType(Catalog.Type.RELATIONAL)
+ .withProvider("test")
+ .withAuditInfo(
+
AuditInfo.builder().withCreator("test").withCreateTime(Instant.now()).build())
+ .build(),
+ true);
+
+ manager.disableMetalake(metalakeIdent);
+ Assertions.assertTrue(manager.dropMetalake(metalakeIdent, true));
+ Mockito.verify(catalogManager)
+ .dropCatalog(NameIdentifier.of(metalakeIdent.name(),
"leftover_catalog"), true);
+ } finally {
+ FieldUtils.writeField(
+ GravitinoEnv.getInstance(), "catalogManager",
previousCatalogManager, true);
+ store.close();
+ }
+ }
+
private void testProperties(Map<String, String> expectedProps, Map<String,
String> testProps) {
expectedProps.forEach(
(k, v) -> {
