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 d71f7121db [#13580] fix(core): Avoid deadlock between cache eviction
and segment locks (#13702)
d71f7121db is described below
commit d71f7121dbc355f0a0c9f6404a4740b04d58931e
Author: Qi Yu <[email protected]>
AuthorDate: Fri Oct 9 20:06:24 2026 +0800
[#13580] fix(core): Avoid deadlock between cache eviction and segment locks
(#13702)
### What changes were proposed in this pull request?
- Replace segment-lock acquisition in eviction/expiration callbacks with
a per-cache index lock. Its critical sections contain only quiet cache
lookups and radix-tree updates, never segment locks or mutating Caffeine
operations.
- Guard index removal against reinsertion, including explicit/cascading
invalidation, and serialize index replacement during clear. Reuse the
existing maintenance executor without adding a cleanup thread or queue.
- Use `removeIndexEntryIfAbsent` at cleanup call sites while retaining
the deprecated `invalidateExpiredItem` hook through a delegating bridge,
preserving existing subclass overrides.
- Clarify oversized-entry behavior: entries already rejected by
maintenance are not indexed; asynchronous maintenance can leave them
temporarily indexed until the SIZE notification.
### Why are the changes needed?
Caffeine may invoke a removal callback while holding its eviction lock,
including CallerRunsPolicy and submission-failure fallback. A writer
holding a segment lock can simultaneously wait for eviction maintenance
inside put, so a callback acquiring that segment completes a deadlock. A
separate cleanup queue avoids the cycle but can accumulate behind a slow
segment holder. The independent index lock avoids both problems.
Unconditional late cleanup can also unindex a reinserted key, preventing
later ancestor invalidation from finding it.
Fix: #13580
### Does this PR introduce _any_ user-facing change?
No configuration changes or breaking API changes; existing protected
subclass hooks remain supported. Ordinary reads are unchanged. Index
writes share a short per-cache critical section, and size() remains an
estimate that may include entries awaiting asynchronous removal
notifications.
### How was this patch tested?
- Cache tests cover foreign-thread inline eviction while segments remain
locked, stale notifications after reinsertion and clear, ancestor
invalidation, eight concurrent writers with the production executor, and
exactly 2,000 unrelated puts while another thread holds a segment.
- Added production-weigher tests for inline rejection and deferred
eviction, plus compatibility tests for legacy
BaseEntityCache/CaffeineEntityCache overrides.
- Passed 87 cache tests and 2556 core test cases (2 skipped, no
failures). Core spotlessCheck and javadoc passed; existing unrelated
javadoc warnings remain. Docker tests were not run.
- Mutation checks: restoring segment-lock acquisition makes the
strengthened held-segment test time out; removing the put presence guard
makes the inline weighted-rejection test fail. Restored the final code
afterwards.
- Independent fresh-context review of 693a09440 found no new
Blocker/Important findings.
---
.../apache/gravitino/cache/BaseEntityCache.java | 23 +-
.../gravitino/cache/CaffeineEntityCache.java | 84 ++++--
.../org/apache/gravitino/cache/NoOpsCache.java | 1 +
.../gravitino/cache/TestBaseEntityCache.java | 14 +-
.../cache/TestCaffeineEntityCacheEviction.java | 336 +++++++++++++++++++++
5 files changed, 435 insertions(+), 23 deletions(-)
diff --git a/core/src/main/java/org/apache/gravitino/cache/BaseEntityCache.java
b/core/src/main/java/org/apache/gravitino/cache/BaseEntityCache.java
index 23143bf331..89ada1ad2f 100644
--- a/core/src/main/java/org/apache/gravitino/cache/BaseEntityCache.java
+++ b/core/src/main/java/org/apache/gravitino/cache/BaseEntityCache.java
@@ -182,9 +182,28 @@ public abstract class BaseEntityCache implements
EntityCache {
}
/**
- * Removes an expired entity from the data cache.
+ * Removes a key from the implementation's bookkeeping only if it is absent
from the underlying
+ * cache. This includes eviction, expiration, and explicit invalidation.
This may be called while
+ * the underlying cache holds an internal lock. Implementations must not
acquire locks held across
+ * underlying cache operations, and must tolerate the key having been added
back before the call.
*
- * @param key The expired entity key to remove.
+ * <p>The default implementation delegates to the legacy hook so existing
subclass overrides
+ * remain effective.
+ *
+ * @param key The cache key to remove from bookkeeping if absent.
+ */
+ protected void removeIndexEntryIfAbsent(EntityCacheKey key) {
+ invalidateExpiredItem(key);
+ }
+
+ /**
+ * Legacy hook for removing an absent key from the implementation's
bookkeeping.
+ *
+ * @param key The evicted, expired, or explicitly invalidated entity key.
+ * @deprecated Use {@link #removeIndexEntryIfAbsent(EntityCacheKey)}.
Existing overrides continue
+ * to receive calls through its default implementation and must obey its
locking and
+ * reinsertion contract.
*/
+ @Deprecated
protected abstract void invalidateExpiredItem(EntityCacheKey key);
}
diff --git
a/core/src/main/java/org/apache/gravitino/cache/CaffeineEntityCache.java
b/core/src/main/java/org/apache/gravitino/cache/CaffeineEntityCache.java
index 0e117a3562..8ac859458d 100644
--- a/core/src/main/java/org/apache/gravitino/cache/CaffeineEntityCache.java
+++ b/core/src/main/java/org/apache/gravitino/cache/CaffeineEntityCache.java
@@ -32,6 +32,7 @@ import
com.googlecode.concurrenttrees.radix.node.concrete.DefaultCharArrayNodeFa
import java.util.List;
import java.util.Optional;
import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
@@ -66,6 +67,17 @@ public class CaffeineEntityCache extends BaseEntityCache {
private static final int CACHE_CLEANUP_QUEUE_CAPACITY = 100;
private static final int CACHE_MONITOR_PERIOD_MINUTES = 5;
private static final int CACHE_MONITOR_INITIAL_DELAY_MINUTES = 0;
+
+ /**
+ * Runs Caffeine's maintenance work and removal notifications.
+ *
+ * <p>The removal listener can still run while Caffeine holds its eviction
lock: when this queue
+ * is full, {@link ThreadPoolExecutor.CallerRunsPolicy} runs the task on the
submitting thread,
+ * which may be the one performing eviction; Caffeine also runs a task
inline if an executor
+ * throws on submission. Meanwhile a writer that holds a segment lock can
block on the eviction
+ * lock in {@code cacheData.put}, because Caffeine cleans up synchronously
when its write buffer
+ * is full. The removal listener must therefore never wait for a segment
lock.
+ */
private static final ExecutorService CLEANUP_EXECUTOR =
new ThreadPoolExecutor(
CACHE_CLEANUP_CORE_THREADS,
@@ -103,6 +115,14 @@ public class CaffeineEntityCache extends BaseEntityCache {
*/
private volatile RadixTree<EntityCacheKey> cacheIndex;
+ /**
+ * Serializes index updates with eviction callbacks. Never take a segment
lock or perform a
+ * mutating Caffeine operation while holding this lock: callbacks can hold
Caffeine's eviction
+ * lock while waiting for it. Only the lock-free policy lookup and
radix-tree operations belong
+ * inside this critical section.
+ */
+ private final Object indexLock = new Object();
+
private ScheduledExecutorService scheduler;
/**
@@ -111,6 +131,17 @@ public class CaffeineEntityCache extends BaseEntityCache {
* @param cacheConfig the cache configuration
*/
public CaffeineEntityCache(Config cacheConfig) {
+ this(cacheConfig, CLEANUP_EXECUTOR);
+ }
+
+ /**
+ * Constructs a new {@link CaffeineEntityCache} with the given executors.
+ *
+ * @param cacheConfig the cache configuration
+ * @param maintenanceExecutor the executor Caffeine uses for maintenance and
removal notifications
+ */
+ @VisibleForTesting
+ CaffeineEntityCache(Config cacheConfig, Executor maintenanceExecutor) {
super(cacheConfig);
this.cacheIndex = new ConcurrentRadixTree<>(new
DefaultCharArrayNodeFactory());
@@ -121,21 +152,20 @@ public class CaffeineEntityCache extends BaseEntityCache {
Caffeine<EntityCacheKey, Entity> cacheDataBuilder =
newBaseBuilder(cacheConfig);
cacheDataBuilder
- .executor(CLEANUP_EXECUTOR)
+ .executor(maintenanceExecutor)
.removalListener(
(key, value, cause) -> {
LOG.debug("Removed entity cache entry, key={}, cause={}", key,
cause);
if (cause == RemovalCause.EXPLICIT || cause ==
RemovalCause.REPLACED) {
return;
}
+ // A callback may hold Caffeine's eviction lock. The independent
index lock is safe
+ // because none of its holders can wait for that eviction lock
or a segment lock.
try {
- invalidateExpiredItem(key);
+ removeIndexEntryIfAbsent(key);
} catch (Throwable t) {
LOG.error(
- "Failed to remove entity key={} from cache asynchronously,
cause={}",
- key,
- cause,
- t);
+ "Failed to remove entity key={} from the cache index,
cause={}", key, cause, t);
}
});
@@ -202,7 +232,9 @@ public class CaffeineEntityCache extends BaseEntityCache {
segmentedLock.withGlobalLock(
() -> {
cacheData.invalidateAll();
- cacheIndex = new ConcurrentRadixTree<>(new
DefaultCharArrayNodeFactory());
+ synchronized (indexLock) {
+ cacheIndex = new ConcurrentRadixTree<>(new
DefaultCharArrayNodeFactory());
+ }
});
}
@@ -216,9 +248,15 @@ public class CaffeineEntityCache extends BaseEntityCache {
entityCacheKey,
() -> {
cacheData.put(entityCacheKey, entity);
- // If the entry was rejected (e.g. it exceeds the maximum weight),
skip indexing it.
- if (cacheData.policy().getIfPresentQuietly(entityCacheKey) != null) {
- cacheIndex.put(entityCacheKey.toString(), entityCacheKey);
+ synchronized (indexLock) {
+ // Check and index atomically against removal callbacks. Skip
entries already removed,
+ // including oversized entries rejected by synchronous
maintenance. With asynchronous
+ // maintenance an oversized entry may still be present and get
indexed; the later SIZE
+ // notification removes its index entry. This is not a synchronous
admission check.
+ // The quiet lookup does not schedule maintenance or acquire
Caffeine's eviction lock.
+ if (cacheData.policy().getIfPresentQuietly(entityCacheKey) !=
null) {
+ cacheIndex.put(entityCacheKey.toString(), entityCacheKey);
+ }
}
});
}
@@ -252,18 +290,24 @@ public class CaffeineEntityCache extends BaseEntityCache {
}
/**
- * Removes the expired entity from the cache. This method is a hook method
for the Cache, when an
- * entry expires, it will call this method.
+ * Removes a key from the prefix index only if it is still absent from
Caffeine.
+ *
+ * <p>The independent index lock serializes the presence check and removal
with indexing a put. No
+ * holder of that lock takes a segment lock or calls a mutating Caffeine
operation, so this method
+ * is safe even when a removal callback holds Caffeine's eviction lock.
*
- * @param key The key of the expired entity
+ * @param key The key of the evicted, expired, or invalidated entity
+ * @deprecated Use {@link #removeIndexEntryIfAbsent(EntityCacheKey)}. The
legacy hook remains the
+ * implementation target of the default bridge to preserve existing
subclass overrides.
*/
+ @Deprecated
@Override
protected void invalidateExpiredItem(EntityCacheKey key) {
- segmentedLock.withLock(
- key,
- () -> {
- cacheIndex.remove(key.toString());
- });
+ synchronized (indexLock) {
+ if (cacheData.policy().getIfPresentQuietly(key) == null) {
+ cacheIndex.remove(key.toString());
+ }
+ }
}
/**
@@ -292,7 +336,7 @@ public class CaffeineEntityCache extends BaseEntityCache {
*/
private void invalidateHierarchy(EntityCacheKey key) {
cacheData.invalidate(key);
- cacheIndex.remove(key.toString());
+ removeIndexEntryIfAbsent(key);
String identifier = key.identifier().toString();
invalidateDescendants(identifier + NAME_LEVEL_BOUNDARY);
@@ -311,7 +355,7 @@ public class CaffeineEntityCache extends BaseEntityCache {
Lists.newArrayList(cacheIndex.getValuesForKeysStartingWith(keyPrefix));
for (EntityCacheKey childKey : childKeys) {
cacheData.invalidate(childKey);
- cacheIndex.remove(childKey.toString());
+ removeIndexEntryIfAbsent(childKey);
}
}
diff --git a/core/src/main/java/org/apache/gravitino/cache/NoOpsCache.java
b/core/src/main/java/org/apache/gravitino/cache/NoOpsCache.java
index 0f839737be..b5aa0114ab 100644
--- a/core/src/main/java/org/apache/gravitino/cache/NoOpsCache.java
+++ b/core/src/main/java/org/apache/gravitino/cache/NoOpsCache.java
@@ -44,6 +44,7 @@ public class NoOpsCache extends BaseEntityCache {
}
/** {@inheritDoc} */
+ @Deprecated
@Override
protected void invalidateExpiredItem(EntityCacheKey key) {
// do nothing
diff --git
a/core/src/test/java/org/apache/gravitino/cache/TestBaseEntityCache.java
b/core/src/test/java/org/apache/gravitino/cache/TestBaseEntityCache.java
index e2e04deaf9..68a88902da 100644
--- a/core/src/test/java/org/apache/gravitino/cache/TestBaseEntityCache.java
+++ b/core/src/test/java/org/apache/gravitino/cache/TestBaseEntityCache.java
@@ -112,10 +112,19 @@ public class TestBaseEntityCache {
Assertions.assertThrows(IllegalArgumentException.class, () ->
cache.put(null));
}
+ @Test
+ void testIndexRemovalDelegatesToLegacySubclassHook() {
+ EntityCacheKey key =
+ EntityCacheKey.of(NameIdentifier.of("metalake"),
Entity.EntityType.METALAKE);
+ cache.removeIndexEntryIfAbsent(key);
+ Assertions.assertEquals(List.of(key), cache.removedKeys);
+ }
+
/** A minimal {@link BaseEntityCache} that records everything handed down to
it. */
private static class RecordingCache extends BaseEntityCache {
private final List<Entity> cached = new ArrayList<>();
private final List<Entity> keyChanges = new ArrayList<>();
+ private final List<EntityCacheKey> removedKeys = new ArrayList<>();
RecordingCache(Config config) {
super(config);
@@ -131,8 +140,11 @@ public class TestBaseEntityCache {
keyChanges.add(entity);
}
+ @Deprecated
@Override
- protected void invalidateExpiredItem(EntityCacheKey key) {}
+ protected void invalidateExpiredItem(EntityCacheKey key) {
+ removedKeys.add(key);
+ }
@Override
public <E extends Entity & HasIdentifier> Optional<E> getIfPresent(
diff --git
a/core/src/test/java/org/apache/gravitino/cache/TestCaffeineEntityCacheEviction.java
b/core/src/test/java/org/apache/gravitino/cache/TestCaffeineEntityCacheEviction.java
new file mode 100644
index 0000000000..dbe708698c
--- /dev/null
+++
b/core/src/test/java/org/apache/gravitino/cache/TestCaffeineEntityCacheEviction.java
@@ -0,0 +1,336 @@
+/*
+ * 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.cache;
+
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Queue;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Lock;
+import org.apache.commons.lang3.reflect.FieldUtils;
+import org.apache.gravitino.Config;
+import org.apache.gravitino.Configs;
+import org.apache.gravitino.Entity;
+import org.apache.gravitino.Namespace;
+import org.apache.gravitino.meta.BaseMetalake;
+import org.apache.gravitino.meta.CatalogEntity;
+import org.apache.gravitino.meta.SchemaEntity;
+import org.apache.gravitino.utils.TestUtil;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests that Caffeine eviction never waits for an entity segment lock, and
that the prefix index
+ * still tracks evicted keys correctly.
+ *
+ * <p>Caffeine runs maintenance and removal notifications inline when its
executor rejects a task,
+ * and does so while holding its eviction lock. A writer that holds a segment
lock can at the same
+ * time wait for that eviction lock inside {@code cacheData.put}, so the
removal listener must not
+ * take a segment lock on the thread Caffeine calls it on.
+ */
+public class TestCaffeineEntityCacheEviction {
+
+ @Test
+ void testEvictionCompletesWhileSegmentLocksAreHeld() throws Exception {
+ ExecutorService evictor = Executors.newSingleThreadExecutor();
+ try {
+ // A direct executor reproduces the rejected-task path: Caffeine
performs eviction and calls
+ // the removal listener inline, under its eviction lock.
+ CaffeineEntityCache cache = new
CaffeineEntityCache(countBoundedConfig(1), Runnable::run);
+ BaseMetalake first = TestUtil.getTestMetalake(1L, "metalake1", "first");
+ BaseMetalake second = TestUtil.getTestMetalake(2L, "metalake2",
"second");
+ EntityCacheKey firstKey = metalakeKey(first);
+ EntityCacheKey secondKey = metalakeKey(second);
+ cache.put(first);
+
+ // Hold both segment locks, so whichever entry Caffeine evicts, its
listener would block here
+ // if it took the segment lock.
+ cache.withCacheLock(
+ firstKey,
+ () ->
+ cache.withCacheLock(
+ secondKey,
+ () -> {
+ Future<?> eviction =
+ evictor.submit(
+ () -> {
+ // Bypass the segment lock to model Caffeine
evicting on a thread that
+ // does not own the evicted key's segment.
+ cache.getCacheData().put(secondKey, second);
+ cache.getCacheData().cleanUp();
+ });
+ Assertions.assertDoesNotThrow(
+ () -> eviction.get(5, TimeUnit.SECONDS),
+ "eviction must not wait for a segment lock while it
holds Caffeine's"
+ + " eviction lock");
+ }));
+
+ Assertions.assertEquals(1, cache.getCacheData().estimatedSize());
+ // Inline removal cleans up the index even while the segment locks
remain held.
+ Awaitility.await()
+ .atMost(5, TimeUnit.SECONDS)
+ .untilAsserted(
+ () ->
+ Assertions.assertEquals(
+
cache.getCacheData().policy().getIfPresentQuietly(firstKey) == null ? 0 : 1,
+ cache.size()));
+ } finally {
+ evictor.shutdownNow();
+ }
+ }
+
+ @Test
+ void testStaleEvictionDoesNotUnindexReinsertedEntry() {
+ // Defer Caffeine's removal notifications so one can arrive after the key
was put back.
+ Queue<Runnable> deferredNotifications = new ArrayDeque<>();
+ CaffeineEntityCache cache =
+ new CaffeineEntityCache(countBoundedConfig(10),
deferredNotifications::add);
+ BaseMetalake metalake = TestUtil.getTestMetalake(1L, "metalake1",
"metalake");
+ CatalogEntity catalog =
+ TestUtil.getTestCatalogEntity(2L, "catalog1",
Namespace.of("metalake1"), "hive", "child");
+ EntityCacheKey catalogKey =
+ EntityCacheKey.of(catalog.nameIdentifier(), Entity.EntityType.CATALOG);
+ cache.put(catalog);
+ runAll(deferredNotifications);
+
+ // Evict everything; the notification for the catalog stays queued.
+ cache.getCacheData().policy().eviction().get().setMaximum(0);
+ cache.getCacheData().cleanUp();
+
Assertions.assertNull(cache.getCacheData().policy().getIfPresentQuietly(catalogKey));
+ cache.getCacheData().policy().eviction().get().setMaximum(10);
+
+ // The catalog is cached again before the stale notification is processed.
+ cache.put(catalog);
+ runAll(deferredNotifications);
+
+ Assertions.assertEquals(1, cache.size(), "the re-cached catalog must stay
indexed");
+ // A cascading invalidation finds descendants only through the index.
+ cache.put(metalake);
+ cache.invalidate(metalake.nameIdentifier(), Entity.EntityType.METALAKE);
+ Assertions.assertNull(
+ cache.getCacheData().policy().getIfPresentQuietly(catalogKey),
+ "invalidating the metalake must also drop its cached catalog");
+ }
+
+ @Test
+ void testHeldSegmentDoesNotAccumulateEvictedKeys() throws Exception {
+ CaffeineEntityCache cache = new
CaffeineEntityCache(countBoundedConfig(64), Runnable::run);
+ BaseMetalake held = TestUtil.getTestMetalake(1L, "held", "slow read");
+ EntityCacheKey heldKey = metalakeKey(held);
+ SegmentedLock locks = (SegmentedLock) FieldUtils.readField(cache,
"segmentedLock", true);
+ Lock heldSegment = locks.getSegmentLock(heldKey);
+ cache.put(held);
+
+ ExecutorService writer = Executors.newSingleThreadExecutor();
+ try {
+ cache.withCacheLock(
+ heldKey,
+ () -> {
+ // The worker evicts the held key on a different thread, so a
callback cannot
+ // reacquire this segment reentrantly. It must finish before this
thread releases it.
+ Future<?> eviction =
+ writer.submit(
+ () -> {
+
cache.getCacheData().policy().eviction().get().setMaximum(0);
+ cache.getCacheData().cleanUp();
+ Assertions.assertEquals(0, cache.size());
+
cache.getCacheData().policy().eviction().get().setMaximum(64);
+ int inserted = 0;
+ for (long id = 100L; inserted < 2_000; id++) {
+ BaseMetalake entity =
+ TestUtil.getTestMetalake(id, "entry" + id,
"eviction");
+ // The put itself must not wait for the segment
deliberately held here.
+ if (locks.getSegmentLock(metalakeKey(entity)) ==
heldSegment) {
+ continue;
+ }
+ cache.put(entity);
+ cache.getCacheData().cleanUp();
+
Assertions.assertEquals(cache.getCacheData().asMap().size(), cache.size());
+ Assertions.assertTrue(
+ cache.size() <= 64, "evicted keys must not
accumulate");
+ inserted++;
+ }
+ });
+ Assertions.assertDoesNotThrow(
+ () -> eviction.get(10, TimeUnit.SECONDS),
+ "eviction and unrelated puts must finish while the segment
remains locked");
+ });
+ } finally {
+ writer.shutdownNow();
+ Assertions.assertTrue(writer.awaitTermination(5, TimeUnit.SECONDS));
+ }
+ }
+
+ @Test
+ void testIndexRemovalPreservesLegacyCaffeineSubclassOverride() {
+ List<EntityCacheKey> removedKeys = new ArrayList<>();
+ CaffeineEntityCache cache =
+ new CaffeineEntityCache(countBoundedConfig(1), Runnable::run) {
+ /** {@inheritDoc} */
+ @Deprecated
+ @Override
+ protected void invalidateExpiredItem(EntityCacheKey key) {
+ removedKeys.add(key);
+ super.invalidateExpiredItem(key);
+ }
+ };
+ BaseMetalake entity = TestUtil.getTestMetalake(1L, "metalake", "legacy
hook");
+ EntityCacheKey key = metalakeKey(entity);
+ cache.put(entity);
+ cache.getCacheData().policy().eviction().get().setMaximum(0);
+ cache.getCacheData().cleanUp();
+ Assertions.assertEquals(List.of(key), removedKeys, "eviction must reach
the legacy override");
+
+ removedKeys.clear();
+ cache.getCacheData().policy().eviction().get().setMaximum(1);
+ cache.put(entity);
+ cache.invalidate(entity.nameIdentifier(), entity.type());
+ Assertions.assertEquals(List.of(key), removedKeys, "explicit invalidation
must reach it too");
+ Assertions.assertEquals(0, cache.size());
+ }
+
+ @Test
+ void testOversizedEntryRemovedBeforeIndexingWithInlineMaintenance() {
+ Config config = new Config() {};
+ config.set(Configs.CACHE_STATS_ENABLED, false);
+ CaffeineEntityCache cache = new CaffeineEntityCache(config, Runnable::run);
+
cache.getCacheData().policy().eviction().get().setMaximum(EntityCacheWeigher.SCHEMA_WEIGHT
- 1);
+ SchemaEntity schema = TestUtil.getTestSchemaEntity();
+ EntityCacheKey key = EntityCacheKey.of(schema.nameIdentifier(),
schema.type());
+
+ cache.put(schema);
+
+
Assertions.assertNull(cache.getCacheData().policy().getIfPresentQuietly(key));
+ Assertions.assertEquals(0, cache.size(), "an entry already rejected must
not be indexed");
+ }
+
+ @Test
+ void testOversizedEntryRemainsIndexedUntilDeferredMaintenance() {
+ Config config = new Config() {};
+ config.set(Configs.CACHE_STATS_ENABLED, false);
+ Queue<Runnable> maintenance = new ArrayDeque<>();
+ CaffeineEntityCache cache = new CaffeineEntityCache(config,
maintenance::add);
+
cache.getCacheData().policy().eviction().get().setMaximum(EntityCacheWeigher.SCHEMA_WEIGHT
- 1);
+ runAll(maintenance);
+ SchemaEntity schema = TestUtil.getTestSchemaEntity();
+ EntityCacheKey key = EntityCacheKey.of(schema.nameIdentifier(),
schema.type());
+
+ cache.put(schema);
+
+
Assertions.assertNotNull(cache.getCacheData().policy().getIfPresentQuietly(key));
+ Assertions.assertEquals(1, cache.size(), "the presence check does not
enforce admission");
+ runAll(maintenance);
+
Assertions.assertNull(cache.getCacheData().policy().getIfPresentQuietly(key));
+ Assertions.assertEquals(0, cache.size(), "the SIZE notification must
remove the index entry");
+ }
+
+ @Test
+ void testClearBeforeStaleRemovalPreservesReinsertedEntry() {
+ Queue<Runnable> deferredNotifications = new ArrayDeque<>();
+ CaffeineEntityCache cache =
+ new CaffeineEntityCache(countBoundedConfig(1),
deferredNotifications::add);
+ BaseMetalake entity = TestUtil.getTestMetalake(1L, "metalake",
"reinserted");
+ EntityCacheKey key = metalakeKey(entity);
+ cache.put(entity);
+ runAll(deferredNotifications);
+ cache.getCacheData().policy().eviction().get().setMaximum(0);
+ cache.getCacheData().cleanUp();
+ cache.clear();
+ cache.getCacheData().policy().eviction().get().setMaximum(1);
+ cache.put(entity);
+ runAll(deferredNotifications);
+
+ Assertions.assertEquals(1, cache.size());
+
Assertions.assertNotNull(cache.getCacheData().policy().getIfPresentQuietly(key));
+ cache.invalidate(entity.nameIdentifier(), entity.type());
+ Assertions.assertEquals(0, cache.size());
+ }
+
+ @Test
+ void testConcurrentEvictionUnderCountBoundCompletesAndKeepsIndexConsistent()
throws Exception {
+ int writers = 8;
+ int putsPerWriter = 5_000;
+ // The production executors: Caffeine's bounded cleanup pool runs rejected
work inline. These
+ // are
+ // static, so if this deadlock regresses the shared cleanup thread stays
stuck for this JVM and
+ // later cache tests in the same fork can time out too; the timeout below
reports it first.
+ CaffeineEntityCache cache = new
CaffeineEntityCache(countBoundedConfig(64));
+ CountDownLatch start = new CountDownLatch(1);
+ ExecutorService pool = Executors.newFixedThreadPool(writers);
+ try {
+ List<Future<?>> results = new ArrayList<>();
+ for (int w = 0; w < writers; w++) {
+ int writer = w;
+ results.add(
+ pool.submit(
+ () -> {
+ start.await();
+ for (int i = 0; i < putsPerWriter; i++) {
+ long id = (long) writer * putsPerWriter + i;
+ cache.put(TestUtil.getTestMetalake(id, "metalake" + id,
"stress"));
+ }
+ return null;
+ }));
+ }
+ start.countDown();
+ for (Future<?> result : results) {
+ Assertions.assertDoesNotThrow(
+ () -> result.get(60, TimeUnit.SECONDS), "concurrent puts must not
deadlock");
+ }
+ } finally {
+ pool.shutdownNow();
+ }
+
+ cache.getCacheData().cleanUp();
+ Awaitility.await()
+ .atMost(10, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ cache.getCacheData().cleanUp();
+ Assertions.assertEquals(cache.getCacheData().asMap().size(),
cache.size());
+ });
+ Assertions.assertTrue(cache.size() <= 64);
+ }
+
+ private static void runAll(Queue<Runnable> tasks) {
+ Runnable task;
+ while ((task = tasks.poll()) != null) {
+ task.run();
+ }
+ }
+
+ private static EntityCacheKey metalakeKey(BaseMetalake metalake) {
+ return EntityCacheKey.of(metalake.nameIdentifier(),
Entity.EntityType.METALAKE);
+ }
+
+ private static Config countBoundedConfig(int maxEntries) {
+ Config config = new Config() {};
+ config.set(Configs.CACHE_WEIGHER_ENABLED, false);
+ config.set(Configs.CACHE_MAX_ENTRIES, maxEntries);
+ config.set(Configs.CACHE_STATS_ENABLED, false);
+ return config;
+ }
+}
