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 a246aa49bb [Cherry-pick to branch-1.3] [#11498] improvement(core):
Reuse catalog cache when resolving properties (#11499) (#11540)
a246aa49bb is described below
commit a246aa49bb994cd8c1f87424c303750d257551ab
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Tue Jun 9 22:09:30 2026 +0800
[Cherry-pick to branch-1.3] [#11498] improvement(core): Reuse catalog cache
when resolving properties (#11499) (#11540)
**Cherry-pick Information:**
- Original commit: 64433f98b1fb8651bd584a8d3ad2650b918f677e
- Target branch: `branch-1.3`
- Status: ✅ Clean cherry-pick (no conflicts)
Co-authored-by: Qi Yu <[email protected]>
---
.../apache/gravitino/catalog/CatalogManager.java | 20 +++---
.../gravitino/catalog/TestCatalogManager.java | 81 ++++++++++++++++++++++
2 files changed, 93 insertions(+), 8 deletions(-)
diff --git
a/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java
b/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java
index 91bd5b08b6..4df78ad21b 100644
--- a/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java
+++ b/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java
@@ -582,7 +582,15 @@ public class CatalogManager implements CatalogDispatcher,
Closeable {
// instance,
// we need to clean up the entity stored.
try {
- store.delete(ident, EntityType.CATALOG, true);
+ if (store.delete(ident, EntityType.CATALOG, true)) {
+ // This cleanup deletion writes a DROP record to the entity
change log. Mark it as
+ // a local mutation so the change-log poller consumes that
record's token instead
+ // of one meant for a later mutation of the same identifier.
Without this, the
+ // poller could treat a subsequent local change as remote
and spuriously
+ // invalidate (and asynchronously close) a cached catalog
wrapper that is still in
+ // use, causing a NullPointerException.
+ markLocalMutation(ident);
+ }
} catch (IOException e4) {
LOG.error("Failed to clean up catalog {}", ident, e4);
}
@@ -1145,13 +1153,9 @@ public class CatalogManager implements
CatalogDispatcher, Closeable {
* @return The resolved properties.
*/
private Map<String, String> getResolvedProperties(CatalogEntity entity) {
- Map<String, String> conf = entity.getProperties();
- String provider = entity.getProvider();
-
- try (IsolatedClassLoader classLoader = createClassLoader(provider, conf)) {
- BaseCatalog<?> catalog = createBaseCatalog(classLoader, entity);
- return classLoader.withClassLoader(cl -> catalog.properties(),
RuntimeException.class);
- }
+ CatalogWrapper catalogWrapper =
loadCatalogAndWrap(entity.nameIdentifier());
+ return catalogWrapper.classLoader.withClassLoader(
+ cl -> catalogWrapper.catalog.properties(), RuntimeException.class);
}
private BaseCatalog<?> createBaseCatalog(IsolatedClassLoader classLoader,
CatalogEntity entity) {
diff --git
a/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java
b/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java
index 4880d7f258..6d2ad36bf1 100644
--- a/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java
+++ b/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java
@@ -436,6 +436,7 @@ public class TestCatalogManager {
catalogManager.createCatalog(relIdent, Catalog.Type.RELATIONAL, provider,
"comment", props);
catalogManager.createCatalog(fileIdent, Catalog.Type.FILESET, provider,
"comment", props);
+ catalogManager.getCatalogCache().invalidateAll();
Catalog[] catalogs = catalogManager.listCatalogsInfo(relIdent.namespace());
Assertions.assertEquals(2, catalogs.length);
@@ -452,6 +453,17 @@ public class TestCatalogManager {
}
}
+ CatalogManager.CatalogWrapper relWrapper =
+ catalogManager.getCatalogCache().getIfPresent(relIdent);
+ CatalogManager.CatalogWrapper fileWrapper =
+ catalogManager.getCatalogCache().getIfPresent(fileIdent);
+ Assertions.assertNotNull(relWrapper);
+ Assertions.assertNotNull(fileWrapper);
+
+ catalogManager.listCatalogsInfo(relIdent.namespace());
+ Assertions.assertSame(relWrapper,
catalogManager.getCatalogCache().getIfPresent(relIdent));
+ Assertions.assertSame(fileWrapper,
catalogManager.getCatalogCache().getIfPresent(fileIdent));
+
// Test list under non-existed metalake
NameIdentifier ident2 = NameIdentifier.of("metalake1", "test1");
Namespace namespace = ident2.namespace();
@@ -760,6 +772,75 @@ public class TestCatalogManager {
manager.close();
}
+ @Test
+ void testFailedCreateCatalogCleanupMarksLocalMutation() throws Exception {
+ ChangeLogAwareEntityStore store = new ChangeLogAwareEntityStore();
+ store.initialize(config);
+ store.put(metalakeEntity, true);
+
+ CatalogManager manager = new CatalogManager(config, store, new
RandomIdGenerator());
+ NameIdentifier ident = NameIdentifier.of("metalake",
"failed_create_cleanup");
+
+ // A creation that fails validation (key1 is required but missing) stores
the entity and then
+ // rolls it back via store.delete(), which writes a DROP record to the
entity change log.
+ Map<String, String> invalidProps =
+ ImmutableMap.of(
+ "provider", "test", PROPERTY_KEY2, "value2", PROPERTY_KEY5_PREFIX
+ "1", "v");
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ manager.createCatalog(
+ ident, Catalog.Type.RELATIONAL, provider, "comment",
invalidProps));
+
+ // Recreate the same catalog successfully and load it into the cache.
+ Map<String, String> validProps =
+ ImmutableMap.of(
+ "provider",
+ "test",
+ PROPERTY_KEY1,
+ "value1",
+ PROPERTY_KEY2,
+ "value2",
+ PROPERTY_KEY5_PREFIX + "1",
+ "value3");
+ manager.createCatalog(ident, Catalog.Type.RELATIONAL, provider, "comment",
validProps);
+ Assertions.assertNotNull(manager.loadCatalogAndWrap(ident));
+ Assertions.assertNotNull(manager.getCatalogCache().getIfPresent(ident));
+
+ // Simulate a subsequent local mutation (e.g. disableCatalog) that writes
an ALTER record.
+ manager.markLocalMutation(ident);
+
+ // The poller delivers both records in one batch: the DROP from the
failed-create cleanup and
+ // the ALTER from the local mutation. The cleanup DROP must carry its own
local-mutation token
+ // (the fix); otherwise it consumes the ALTER's token, the ALTER is
treated as remote, and the
+ // in-use cached wrapper is spuriously invalidated and asynchronously
closed.
+ store
+ .listener
+ .get()
+ .onEntityChange(
+ List.of(
+ new EntityChangeRecord(
+ 1L,
+ "metalake",
+ "CATALOG",
+ "metalake.failed_create_cleanup",
+ OperateType.DROP,
+ 0L),
+ new EntityChangeRecord(
+ 2L,
+ "metalake",
+ "CATALOG",
+ "metalake.failed_create_cleanup",
+ OperateType.ALTER,
+ 0L)));
+
+ Assertions.assertNotNull(
+ manager.getCatalogCache().getIfPresent(ident),
+ "Cache should NOT be invalidated: the failed-create cleanup DROP must
be tracked as a "
+ + "local mutation so it does not steal the token meant for the
later ALTER");
+ manager.close();
+ }
+
@Test
public void testDropCatalogSkipsImportedSchemas() throws Exception {
NameIdentifier ident = NameIdentifier.of("metalake", "test41");
