yuqi1129 commented on code in PR #13702:
URL: https://github.com/apache/gravitino/pull/13702#discussion_r4228425000
##########
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:
Addressed in 693a09440. Restored the rationale for skipping an entry already
removed by maintenance, and clarified that the quiet lookup is not a
synchronous admission check: with deferred maintenance an oversized entry can
be indexed temporarily, and the later SIZE notification removes it. Added
weighted-cache tests for both inline rejection before indexing and deferred
eviction after indexing, using the production weigher. Removing the presence
guard makes the inline-rejection test fail.
##########
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:
Addressed in 693a09440. The eviction callback and both explicit-invalidation
paths now call removeIndexEntryIfAbsent. I retained invalidateExpiredItem as a
deprecated compatibility hook: the default new entry point delegates to it, so
existing BaseEntityCache and CaffeineEntityCache subclass overrides still
receive calls. Added regression tests for that dispatch, including eviction and
explicit invalidation. This avoids changing the existing protected extension
contract just to improve the call-site name.
##########
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:
Addressed in 693a09440. The main thread now keeps the segment locked while a
separate worker evicts the held key and completes exactly 2,000 insertions on
other segments. The worker must finish before the main thread releases the
lock; the test also checks the index/data invariant and the 64-entry bound.
Reintroducing segment-lock acquisition makes this test fail with
TimeoutException, confirming it can no longer pass through reentrant
acquisition. Updated the comments and PR testing description accordingly.
--
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]
