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 933401bda9 [#12504] fix(core): Fix flaky
TestCatalogManager.testCatalogCacheRemoveListener (#12514)
933401bda9 is described below
commit 933401bda9da09fcfa1a5bf4048a92bce1ab8460
Author: Qi Yu <[email protected]>
AuthorDate: Wed Aug 26 17:16:29 2026 +0800
[#12504] fix(core): Fix flaky
TestCatalogManager.testCatalogCacheRemoveListener (#12514)
### What changes were proposed in this pull request?
1. `testCatalogCacheRemoveListener` now runs against its own
`CatalogManager` and `InMemoryEntityStore` instead of the `static`
instance shared by the whole test class.
2. `reset()` (`@BeforeEach`/`@AfterEach`) additionally invalidates the
shared catalog cache, so cache entries no longer leak from one test
method to the next.
3. `CatalogManager.removalListeners` becomes a `CopyOnWriteArrayList`.
### Why are the changes needed?
The shared `CatalogManager` is created once in `@BeforeAll` and
`reset()` only cleared `entityStore`, so its catalog cache accumulated
entries across test methods. Instrumenting the test showed 4 leftover
entries present when it starts:
```
cache size at test start = 4 keys=[metalake.test1, metalake.catalog_rel,
metalake.catalog_file, metalake.test51]
```
Caffeine delivers removal notifications asynchronously, and
`CatalogManager` has no API to unregister a listener, so those leftover
removals could reach the listener this test registers, producing the
reported `expected: <1> but was: <5>` (4 leftovers + its own).
`removalListeners` was a plain `ArrayList` iterated on cache executor
threads while `addCatalogCacheRemoveListener` may append concurrently —
unsafe regardless of the flaky test.
Fix: #12504
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
- `./gradlew :core:test --tests
"org.apache.gravitino.catalog.TestCatalogManager" -PskipITs` passes.
- Temporarily annotated the class with
`@TestMethodOrder(MethodOrderer.Random.class)` and re-ran it 3 times
with `--rerun-tasks`; all green (annotation not included in the commit).
---
.../apache/gravitino/catalog/CatalogManager.java | 5 +-
.../gravitino/catalog/TestCatalogManager.java | 63 +++++++++++++---------
2 files changed, 43 insertions(+), 25 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 b6090c9671..c5a205fb24 100644
--- a/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java
+++ b/core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java
@@ -53,6 +53,7 @@ import java.util.Properties;
import java.util.ServiceLoader;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
@@ -367,7 +368,9 @@ public class CatalogManager implements CatalogDispatcher,
Closeable {
private final SecretManager secretManager;
- private final List<Consumer<NameIdentifier>> removalListeners =
Lists.newArrayList();
+ // Copy-on-write: listeners may be registered while the cache's removal
listener (running on a
+ // cache executor thread) is iterating this list.
+ private final List<Consumer<NameIdentifier>> removalListeners = new
CopyOnWriteArrayList<>();
private final ConcurrentHashMap<NameIdentifier, AtomicInteger>
localMutationCounts =
new ConcurrentHashMap<>();
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 23b9140a36..b6caa8cbd9 100644
--- a/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java
+++ b/core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java
@@ -145,6 +145,9 @@ public class TestCatalogManager {
void reset() throws IOException {
((InMemoryEntityStore) entityStore).clear();
entityStore.put(metalakeEntity, true);
+ // The shared CatalogManager is created once in @BeforeAll, so its cache
would otherwise keep
+ // entries created by previously executed test methods and make tests
order-dependent.
+ catalogManager.getCatalogCache().invalidateAll();
}
@AfterAll
@@ -1443,37 +1446,49 @@ public class TestCatalogManager {
}
@Test
- public void testCatalogCacheRemoveListener() {
+ public void testCatalogCacheRemoveListener() throws IOException {
NameIdentifier ident = NameIdentifier.of(metalake, "catalog");
Map<String, String> props =
ImmutableMap.of(
PROPERTY_KEY1, "value1", PROPERTY_KEY2, "value2",
PROPERTY_KEY5_PREFIX + "1", "value3");
- // Create a catalog
- catalogManager.createCatalog(ident, Catalog.Type.RELATIONAL, provider,
"comment", props);
+ // Use a dedicated CatalogManager (and entity store) instead of the shared
static one: the
+ // shared instance keeps cache entries and cache removal listeners
registered by other test
+ // methods, which would make the assertions below depend on the test
execution order.
+ EntityStore store = new InMemoryEntityStore();
+ store.initialize(config);
+ store.put(metalakeEntity, true);
- // Load the catalog to add it to the cache
- catalogManager.loadCatalog(ident);
-
Assertions.assertNotNull(catalogManager.getCatalogCache().getIfPresent(ident));
+ try (CatalogManager manager =
+ new CatalogManager(config, store, new RandomIdGenerator(), new
SecretManager(config))) {
+ // Create a catalog
+ manager.createCatalog(ident, Catalog.Type.RELATIONAL, provider,
"comment", props);
+
+ // Load the catalog to add it to the cache
+ manager.loadCatalog(ident);
+ Assertions.assertNotNull(manager.getCatalogCache().getIfPresent(ident));
+
+ // Add a listener to track removed catalogs
+ Set<NameIdentifier> removedCatalogs = Sets.newConcurrentHashSet();
+ manager.addCatalogCacheRemoveListener(removedCatalogs::add);
- // Add a listener to track removed catalogs
- Set<NameIdentifier> removedCatalogs = Sets.newConcurrentHashSet();
- catalogManager.addCatalogCacheRemoveListener(removedCatalogs::add);
-
- // Invalidate the cache to trigger the removal listener
- catalogManager.getCatalogCache().invalidate(ident);
-
- // Wait for the async eviction to complete
- await()
- .atMost(Duration.ofSeconds(5))
- .untilAsserted(
- () -> {
- Assertions.assertTrue(
- removedCatalogs.contains(ident),
- "Listener should be notified of catalog removal");
- Assertions.assertEquals(
- 1, removedCatalogs.size(), "Only one catalog should be
removed");
- });
+ // Invalidate the cache to trigger the removal listener
+ manager.getCatalogCache().invalidate(ident);
+
+ // Wait for the async eviction to complete
+ await()
+ .atMost(Duration.ofSeconds(5))
+ .untilAsserted(
+ () -> {
+ Assertions.assertTrue(
+ removedCatalogs.contains(ident),
+ "Listener should be notified of catalog removal");
+ Assertions.assertEquals(
+ 1, removedCatalogs.size(), "Only one catalog should be
removed");
+ });
+ } finally {
+ store.close();
+ }
}
private void testProperties(Map<String, String> expectedProps, Map<String,
String> testProps) {
