Copilot commented on code in PR #13702:
URL: https://github.com/apache/gravitino/pull/13702#discussion_r4218973929
##########
core/src/main/java/org/apache/gravitino/cache/CaffeineEntityCache.java:
##########
@@ -80,6 +93,24 @@ public class CaffeineEntityCache extends BaseEntityCache {
},
new ThreadPoolExecutor.CallerRunsPolicy());
+ /**
+ * Removes evicted and expired keys from the prefix index. The removal
listener only hands keys to
+ * this executor; the cleanup itself takes the key's segment lock, which is
safe here because this
+ * thread never holds a Caffeine lock.
+ *
+ * <p>The queue is unbounded and the executor is never shut down, so handing
off a key never fails
+ * or runs the cleanup on Caffeine's thread. Each task holds only a key, so
the backlog is bounded
+ * by the number of pending evictions. A segment held across a slow store
read delays the cleanup
+ * behind it, which only lets {@link #size()} over-count and the stale keys
use memory for longer.
+ */
+ private static final ExecutorService INDEX_CLEANUP_EXECUTOR =
+ Executors.newSingleThreadExecutor(
+ r -> {
+ Thread t = new Thread(r, "CaffeineEntityCache-IndexCleanup");
+ t.setDaemon(true);
+ return t;
+ });
Review Comment:
The index-cleanup executor uses an unbounded queue and is explicitly never
shut down. Under sustained eviction (or if segment locks are held for long
periods), the backlog can grow without bound and retain many `EntityCacheKey`
instances, increasing memory pressure. Consider using a bounded queue with a
rejection policy that *does not* run in the caller (e.g., log-and-drop), and/or
de-duplicating scheduled cleanups per key (e.g., a concurrent set of ‘pending’
keys) so repeated evictions don’t enqueue unlimited duplicate tasks. If
lifecycle management is possible in this project, also consider wiring shutdown
into cache/service shutdown to avoid long-lived background threads in
environments that reload components.
##########
core/src/test/java/org/apache/gravitino/cache/TestCaffeineEntityCacheEviction.java:
##########
@@ -0,0 +1,202 @@
+/*
+ * 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 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.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 indexCleanup = Executors.newSingleThreadExecutor();
+ 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,
indexCleanup);
+ 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());
+ // The evicted entry leaves the prefix index once its segment is free
again.
+ Awaitility.await()
+ .atMost(5, TimeUnit.SECONDS)
+ .untilAsserted(
+ () ->
+ Assertions.assertEquals(
+
cache.getCacheData().policy().getIfPresentQuietly(firstKey) == null ? 0 : 1,
+ cache.size()));
+ } finally {
+ evictor.shutdownNow();
+ indexCleanup.shutdownNow();
+ }
+ }
+
+ @Test
+ void testStaleEvictionDoesNotUnindexReinsertedEntry() {
+ // Defer Caffeine's removal notifications so one can arrive after the key
was put back. They run
+ // later on this thread, which then holds no Caffeine lock, so a direct
index cleanup is safe.
+ Queue<Runnable> deferredNotifications = new ArrayDeque<>();
+ CaffeineEntityCache cache =
+ new CaffeineEntityCache(countBoundedConfig(10),
deferredNotifications::add, Runnable::run);
+ 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);
Review Comment:
This test assumes `setMaximum(0)` makes the entry immediately absent, but
eviction can be asynchronous depending on Caffeine’s internal scheduling and
maintenance execution. This can make the assertion flaky. To make it
deterministic, trigger/await maintenance (e.g., call
`cache.getCacheData().cleanUp()` and/or use Awaitility to wait until
`getIfPresentQuietly(catalogKey)` becomes null) before asserting.
##########
core/src/main/java/org/apache/gravitino/cache/CaffeineEntityCache.java:
##########
@@ -252,20 +296,52 @@ public <E, T extends Exception> E withCacheLock(
}
/**
- * 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 that Caffeine evicted or expired from the prefix index.
Runs on the index cleanup
+ * executor, never on a thread that holds a Caffeine lock.
+ *
+ * <p>A put can add the key back between Caffeine removing it and this
cleanup running. That put
+ * indexes the key under the same segment lock, so the index entry is
removed only when the key is
+ * still absent from the cache; otherwise it belongs to the live entry.
*
- * @param key The key of the expired entity
+ * @param key The key of the evicted or expired entity
*/
@Override
protected void invalidateExpiredItem(EntityCacheKey key) {
segmentedLock.withLock(
key,
() -> {
- cacheIndex.remove(key.toString());
+ if (cacheData.policy().getIfPresentQuietly(key) == null) {
+ cacheIndex.remove(key.toString());
+ }
});
}
+ /**
+ * Hands an evicted or expired key to the index cleanup executor without
blocking.
+ *
+ * @param key The key Caffeine removed
+ * @param cause Why Caffeine removed it
+ */
+ private void scheduleIndexCleanup(EntityCacheKey key, RemovalCause cause) {
+ try {
+ indexCleanupExecutor.execute(
+ () -> {
+ try {
+ invalidateExpiredItem(key);
+ } catch (Throwable t) {
+ LOG.error(
+ "Failed to remove entity key={} from the cache index,
cause={}", key, cause, t);
+ }
+ });
+ } catch (RejectedExecutionException e) {
+ // Unreachable with the production executor, which is unbounded and
never shut down. Never
+ // fall
+ // back to running the cleanup here: this thread may hold Caffeine's
eviction lock. A key left
+ // in the index only costs memory; a later invalidation of it or an
ancestor removes it.
+ LOG.error("Failed to schedule cache index cleanup for key={}, cause={}",
key, cause, e);
+ }
+ }
Review Comment:
`Executor.execute(...)` can throw runtime exceptions beyond
`RejectedExecutionException` (e.g., from custom executors or wrappers). Since
this method is invoked from Caffeine’s removal listener, letting an unexpected
runtime exception escape risks disrupting eviction/notification paths. Consider
broadening the catch to `RuntimeException` (or even `Throwable` if you want to
be maximally defensive here) to ensure the listener never throws, while still
avoiding any fallback that runs cleanup inline on the caller thread.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
