jerryshao commented on code in PR #13702:
URL: https://github.com/apache/gravitino/pull/13702#discussion_r4228005434
##########
core/src/main/java/org/apache/gravitino/cache/CaffeineEntityCache.java:
##########
@@ -216,9 +248,12 @@ protected <E extends Entity & HasIdentifier> void doPut(E
entity) {
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. This
policy lookup does not
+ // schedule maintenance or acquire Caffeine's eviction lock.
+ if (cacheData.policy().getIfPresentQuietly(entityCacheKey) !=
null) {
Review Comment:
[Question] The comment this replaced carried the reason the check exists at
all: "If the entry was rejected (e.g. it exceeds the maximum weight), skip
indexing it." That rationale is the more valuable half, because
`CACHE_WEIGHER_ENABLED` defaults to true (Configs.java:514-521), so weight
rejection is the production case rather than a corner case.
Two things worth confirming:
1. Is the check still intended to cover rejection? If so, that sentence is
worth keeping alongside the new atomicity note.
2. With the production executor it will usually not catch rejection anyway.
`BoundedLocalCache.afterWrite` only cleans up inline when the write-buffer
offer fails; otherwise it defers to the maintenance executor, so an oversized
entry is normally still present when this lookup runs and does get indexed - it
is the later SIZE notification that unindexes it. If that is understood and
accepted, saying so here would stop a future reader from relying on this check
to keep oversized keys out of the index.
Verified by: javap of `BoundedLocalCache.afterWrite` in caffeine-2.9.3
(`MpscGrowableArrayQueue.offer` -> `scheduleAfterWrite`/`scheduleDrainBuffers`,
with `performCleanUp` only on the offer-failure branch); read
Configs.java:514-521 for the weigher default; the removed comment is in the
diff for this hunk.
##########
core/src/main/java/org/apache/gravitino/cache/CaffeineEntityCache.java:
##########
@@ -292,7 +330,7 @@ protected void invalidateExpiredItem(EntityCacheKey key) {
*/
private void invalidateHierarchy(EntityCacheKey key) {
cacheData.invalidate(key);
- cacheIndex.remove(key.toString());
+ invalidateExpiredItem(key);
Review Comment:
[Nit] `invalidateExpiredItem` is now the index remover for explicit
invalidation as well (here, and line 352 for descendants), and its body is a
guarded index-only removal that has nothing to do with expiry. The name reads
wrong at both new call sites: nothing has expired when a caller invokes
`invalidate(...)`. The SPI javadoc was updated (BaseEntityCache.java:184-192)
but the name was not, so the next reader has to open the body to learn what the
call does. Consider renaming the protected method to match, e.g.
`unindexIfAbsent` or `removeIndexEntryIfAbsent`.
Verified by: read both call sites (CaffeineEntityCache.java:333 and 352) and
the body (299-305); grepped `invalidateExpiredItem` across the repo - the only
other implementations are NoOpsCache.java:48 and TestBaseEntityCache.java:135,
so the rename touches three files.
##########
core/src/test/java/org/apache/gravitino/cache/TestCaffeineEntityCacheEviction.java:
##########
@@ -0,0 +1,253 @@
+/*
+ * 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.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());
+ // 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();
+ }
+ }
+
+ @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);
+
+ cache.withCacheLock(
+ heldKey,
+ () -> {
+ // A slow backend read holds this segment after its entry is
evicted. Cleanup must not
+ // wait for it, otherwise evictions of unrelated keys accumulate
behind this one.
Review Comment:
[Nit] The stated rationale does not match what this test can detect.
Caffeine's executor here is `Runnable::run`, so every removal notification runs
inline on this same test thread - the thread that already holds `heldSegment`.
Segment locks are Guava `Striped` `ReentrantLock`s, so even the pre-PR body of
`invalidateExpiredItem` (`segmentedLock.withLock(key, ...)`) would reacquire
them reentrantly and never block; keys hashing to `heldSegment` are skipped on
top of that. So nothing can wait or accumulate behind the held segment here,
and reintroducing the segment lock in `invalidateExpiredItem` would leave this
test green - only `testEvictionCompletesWhileSegmentLocksAreHeld`, which evicts
from a second thread, catches that.
What this test does verify is valuable: the index-vs-data invariant and the
`<= 64` bound under 2,000 inline evictions. Suggest either rewording the
comment to claim that, or driving the puts from a separate thread so the
no-waiting property is what is actually exercised.
Verified by: read the test (138-165) against SegmentedLock.java:72-74 and
94-113 (Guava `Striped.lock`, reentrant read-lock reacquisition documented at
37-45), and the pre-PR `invalidateExpiredItem` body in origin/main.
--
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]
