This is an automated email from the ASF dual-hosted git repository.
jerryshao 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 3b4778feb3 [#12376] fix(core): fence batchGet write-back against
in-flight cache invalidation (#13374)
3b4778feb3 is described below
commit 3b4778feb3023c97b880ee5fba65ecabc5c4c43e
Author: Qi Yu <[email protected]>
AuthorDate: Tue Sep 22 19:24:16 2026 +0800
[#12376] fix(core): fence batchGet write-back against in-flight cache
invalidation (#13374)
### What changes were proposed in this pull request?
- `RelationalEntityStore` keeps an invalidation epoch that every cache
invalidation and clear advances. All of the store's own invalidations go
through `invalidateCache()` / `clearCache()`, and the change-log
listener now takes a narrow `EntityCacheChangeLogListener.Target`
provided by the store instead of the raw `EntityCache`, so replayed
invalidations advance the epoch too. The existing
`EntityCacheChangeLogListener(EntityCache)` constructor is kept.
- `RelationalEntityStore.batchGet` samples the epoch before the backend
read and, under each entry's cache lock, writes the entity back only if
the epoch is unchanged.
- Tests: `TestRelationalEntityStoreBatchGetLateFill` — write-back
succeeds normally; is skipped when a change-log invalidation, a local
`delete`, or the listener's clear fallback lands during the backend
read.
### Why are the changes needed?
`get()` holds the entry's segment lock across the backend read and the
write-back, so it cannot race with the poller's `invalidate`.
`batchGet()` did not: it read the backend and called `cache.put()`
unlocked, so an invalidation landing in between was overwritten by the
stale copy, which then survived until the cache TTL (default 1h) instead
of one poll interval. This is on the authorization hot path
(`MetadataAuthzHelper`, `JcasbinAuthorizer`).
Holding one lock per key across the batch DB round trip was rejected:
with the default 16 segments it would stall unrelated cache reads on the
node. The epoch is cache-wide, so a batch fill that coincides with any
invalidation is skipped (the next `get` refills); it can never resurrect
a stale value.
The residual window where a *hierarchical* invalidation (e.g. schema
drop) removes a child key without taking the child's lock is shared with
`get()` and unchanged here.
Part of: #12376 (the fresh-read channel and the action-sensitive call
sites are handled under #12150).
Fixed: #12376
### Does this PR introduce _any_ user-facing change?
No. The `EntityCache` SPI is unchanged; everything lives in
`storage.relational`.
### How was this patch tested?
New unit tests above; `./gradlew :core:test --tests
'org.apache.gravitino.cache.*' --tests
'org.apache.gravitino.storage.relational.TestRelationalEntityStore*'
--tests 'org.apache.gravitino.storage.relational.TestEntityCache*'
--tests 'org.apache.gravitino.storage.relational.TestEntityChangeLog*'
-PskipITs` (126 tests green).
---
.../relational/EntityCacheChangeLogListener.java | 52 ++++-
.../storage/relational/RelationalEntityStore.java | 94 ++++++--
.../TestRelationalEntityStoreBatchGetLateFill.java | 260 +++++++++++++++++++++
3 files changed, 388 insertions(+), 18 deletions(-)
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/EntityCacheChangeLogListener.java
b/core/src/main/java/org/apache/gravitino/storage/relational/EntityCacheChangeLogListener.java
index ed7b942e00..ad9d568f10 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/EntityCacheChangeLogListener.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/EntityCacheChangeLogListener.java
@@ -61,16 +61,58 @@ public class EntityCacheChangeLogListener implements
EntityChangeLogListener {
private static final Logger LOG =
LoggerFactory.getLogger(EntityCacheChangeLogListener.class);
- private final EntityCache cache;
+ /**
+ * The two invalidation entry points this listener needs from the cache it
keeps coherent. The
+ * entity store hands in its own implementation so that it observes every
invalidation, including
+ * the ones replayed from other nodes (see {@code
RelationalEntityStore#batchGet}).
+ */
+ public interface Target {
+ /**
+ * Invalidates the cache entry of the given entity, see {@link
EntityCache#invalidate}.
+ *
+ * @param ident the identifier of the changed entity
+ * @param type the type of the changed entity
+ */
+ void invalidate(NameIdentifier ident, EntityType type);
+
+ /** Clears the whole cache, see {@link EntityCache#clear()}. */
+ void clear();
+ }
+
+ private final Target target;
/**
- * Creates a listener that invalidates the given entity store cache.
+ * Creates a listener that invalidates the given entity store cache directly.
*
* @param cache the per-node entity store cache to keep coherent
*/
public EntityCacheChangeLogListener(EntityCache cache) {
+ this(asTarget(cache));
+ }
+
+ /**
+ * Creates a listener that invalidates through the given target.
+ *
+ * @param target the invalidation entry points of the per-node cache to keep
coherent
+ */
+ public EntityCacheChangeLogListener(Target target) {
+ Preconditions.checkArgument(target != null, "target cannot be null");
+ this.target = target;
+ }
+
+ private static Target asTarget(EntityCache cache) {
Preconditions.checkArgument(cache != null, "cache cannot be null");
- this.cache = cache;
+ return new Target() {
+ @Override
+ public void invalidate(NameIdentifier ident, EntityType type) {
+ cache.invalidate(ident, type);
+ }
+
+ @Override
+ public void clear() {
+ cache.clear();
+ }
+ };
}
@Override
@@ -86,7 +128,7 @@ public class EntityCacheChangeLogListener implements
EntityChangeLogListener {
try {
LOG.debug("Invalidating entity cache due to entity change log: {}
({})", ident, type);
- cache.invalidate(ident, type);
+ target.invalidate(ident, type);
} catch (RuntimeException e) {
// Dropping a single invalidation would leave this node serving that
entity stale until it
// expires. Clearing the whole cache is the safe superset, and it also
covers the rest of
@@ -97,7 +139,7 @@ public class EntityCacheChangeLogListener implements
EntityChangeLogListener {
ident,
type,
e);
- cache.clear();
+ target.clear();
return;
}
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/RelationalEntityStore.java
b/core/src/main/java/org/apache/gravitino/storage/relational/RelationalEntityStore.java
index 50572c5b9d..186d0b8cc8 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/RelationalEntityStore.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/RelationalEntityStore.java
@@ -29,6 +29,7 @@ import java.util.Arrays;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;
import java.util.function.Function;
import javax.annotation.Nullable;
@@ -46,6 +47,7 @@ import org.apache.gravitino.RelationQuery;
import org.apache.gravitino.RelationUpdate;
import org.apache.gravitino.RelationalEntity;
import org.apache.gravitino.SupportsRelationOperations;
+import org.apache.gravitino.cache.BaseEntityCache;
import org.apache.gravitino.cache.CacheFactory;
import org.apache.gravitino.cache.CachedEntityIdResolver;
import org.apache.gravitino.cache.Coherence;
@@ -75,6 +77,11 @@ public class RelationalEntityStore
private EntityChangeLogCleaner entityChangeLogCleaner;
private EntityCache cache;
+ // Advanced before every invalidation observed by this store, whether local
or replayed from the
+ // change log. A shared cache without a local change-log listener needs its
own distributed
+ // version check: this counter cannot detect changes made on another node.
+ private final AtomicLong cacheInvalidationEpoch = new AtomicLong();
+
// Non-null only for a LOCAL_PER_NODE cache, which needs cross-node
invalidation. SHARED and NONE
// caches have no per-node copy to invalidate, so no listener is registered.
@Nullable private EntityCacheChangeLogListener entityCacheChangeLogListener;
@@ -127,10 +134,32 @@ public class RelationalEntityStore
return;
}
- this.entityCacheChangeLogListener = new
EntityCacheChangeLogListener(cache);
+ this.entityCacheChangeLogListener = newCacheChangeLogListener();
this.entityChangeLogPoller.registerListener(entityCacheChangeLogListener);
}
+ /**
+ * Creates the change-log listener that keeps this store's cache coherent
with changes made on
+ * other nodes.
+ *
+ * @return a listener whose invalidations go through this store, see {@link
#batchGet}
+ */
+ @VisibleForTesting
+ EntityCacheChangeLogListener newCacheChangeLogListener() {
+ return new EntityCacheChangeLogListener(
+ new EntityCacheChangeLogListener.Target() {
+ @Override
+ public void invalidate(NameIdentifier ident, Entity.EntityType type)
{
+ invalidateCache(ident, type);
+ }
+
+ @Override
+ public void clear() {
+ clearCache();
+ }
+ });
+ }
+
private RelationalBackend createRelationalEntityBackend(Config config) {
String backendName = config.get(ENTITY_RELATIONAL_STORE);
String className =
@@ -177,7 +206,7 @@ public class RelationalEntityStore
// An overwrite is resolved by the database, which may keep the identity
and version of the
// row it already had. Caching the copy handed in here would publish
values the stored row
// does not carry, so the next read is served from the backend instead.
- cache.invalidate(e.nameIdentifier(), e.type());
+ invalidateCache(e.nameIdentifier(), e.type());
} else {
cache.put(e);
}
@@ -188,7 +217,7 @@ public class RelationalEntityStore
NameIdentifier ident, Class<E> type, Entity.EntityType entityType,
Function<E, E> updater)
throws IOException, NoSuchEntityException, EntityAlreadyExistsException {
E updatedEntity = backend.update(ident, entityType, updater);
- cache.invalidate(ident, entityType);
+ invalidateCache(ident, entityType);
return updatedEntity;
}
@@ -223,9 +252,35 @@ public class RelationalEntityStore
return entity.isEmpty();
})
.toList();
+ // Unlike get(), the backend read is not done under the entries' cache
locks: holding one lock
+ // per key across a batch DB round trip would stall unrelated reads on the
same segments. So an
+ // invalidation can land between the read and the write-back. The epoch
sampled here detects
+ // that and skips the write-back, otherwise the stale copy would survive
until the TTL. The
+ // per-key lock makes the check and the put atomic against an invalidation
of the same key.
+ long epochBeforeRead = cacheInvalidationEpoch.get();
List<E> fetchEntities = backend.batchGet(noCacheIdents, entityType);
for (E entity : fetchEntities) {
- cache.put(entity);
+ if (cache instanceof BaseEntityCache &&
!BaseEntityCache.isCacheable(entity.type())) {
+ // BaseEntityCache.put may invalidate a related entry even when it
does not cache this
+ // entity. Keep that hook, but avoid taking a key lock for a value
that cannot be cached.
+ if (cacheInvalidationEpoch.get() == epochBeforeRead) {
+ cache.put(entity);
+ }
+ allEntities.add(entity);
+ continue;
+ }
+ cache.withCacheLock(
+ EntityCacheKey.of(entity.nameIdentifier(), entity.type()),
+ () -> {
+ if (cacheInvalidationEpoch.get() == epochBeforeRead) {
+ cache.put(entity);
+ // A whole-cache clear can run while this key lock is held. If
it happened during
+ // put, remove the value we may have written after the clear.
+ if (cacheInvalidationEpoch.get() != epochBeforeRead) {
+ cache.invalidate(entity.nameIdentifier(), entity.type());
+ }
+ }
+ });
allEntities.add(entity);
}
return allEntities;
@@ -240,7 +295,7 @@ public class RelationalEntityStore
} catch (NoSuchEntityException e) {
return false;
} finally {
- cache.invalidate(ident, entityType);
+ invalidateCache(ident, entityType);
}
}
@@ -254,7 +309,7 @@ public class RelationalEntityStore
try {
return backend.deleteAndGet(ident, entityType, clazz, postDeleteAction);
} finally {
- cache.invalidate(ident, entityType);
+ invalidateCache(ident, entityType);
}
}
@@ -358,8 +413,8 @@ public class RelationalEntityStore
// relation write can change data materialized into the endpoint entity.
Note this is not free —
// EntityCache#invalidate cascades over the identifier hierarchy, so
invalidating a catalog also
// drops every cached schema and table beneath it.
- cache.invalidate(srcIdentifier, srcType);
- cache.invalidate(dstIdentifier, dstType);
+ invalidateCache(srcIdentifier, srcType);
+ invalidateCache(dstIdentifier, dstType);
}
@Override
@@ -379,9 +434,9 @@ public class RelationalEntityStore
// Invalidate both endpoints for the same reason as insertRelation,
including the hierarchy
// cascade noted there.
for (NameIdentifier ident : srcIdentifiers) {
- cache.invalidate(ident, srcType);
+ invalidateCache(ident, srcType);
}
- cache.invalidate(dstIdentifier, dstType);
+ invalidateCache(dstIdentifier, dstType);
}
@Override
@@ -407,7 +462,7 @@ public class RelationalEntityStore
backend.updateEntityRelations(
relType, srcEntityIdent, srcEntityType, destEntitiesToAdd,
destEntitiesToRemove);
Entity.EntityType targetEntityType = relationUpdateTargetType(relType);
- cache.invalidate(srcEntityIdent, srcEntityType);
+ invalidateCache(srcEntityIdent, srcEntityType);
invalidateRelationTargetCache(targetEntityType, update.targetsToAdd());
invalidateRelationTargetCache(targetEntityType, update.targetsToRemove());
@@ -440,7 +495,7 @@ public class RelationalEntityStore
// Invalidate after the backend write, not before: invalidating first
opens a window where a
// concurrent read could repopulate the cache with stale pre-commit data.
Entity.EntityType targetEntityType =
relationUpdateTargetType(update.relationType());
- cache.invalidate(update.sourceIdentifier(), update.sourceEntityType());
+ invalidateCache(update.sourceIdentifier(), update.sourceEntityType());
invalidateRelationTargetCache(targetEntityType, targetsToAdd);
invalidateRelationTargetCache(targetEntityType, targetsToRemove);
@@ -463,7 +518,7 @@ public class RelationalEntityStore
private void invalidateRelationTargetCache(
Entity.EntityType targetEntityType, RelationEdgeTarget[]
relationTargets) {
for (RelationEdgeTarget relationTarget : relationTargets) {
- cache.invalidate(relationTarget.nameIdentifier(), targetEntityType);
+ invalidateCache(relationTarget.nameIdentifier(), targetEntityType);
}
}
@@ -508,4 +563,17 @@ public class RelationalEntityStore
String.format("Doesn't support the relation type %s", relType));
}
}
+
+ private void invalidateCache(NameIdentifier ident, Entity.EntityType type) {
+ // Advance before removing, so a batchGet() that samples the epoch after
this point reads the
+ // backend after the change that triggered the invalidation is visible.
+ cacheInvalidationEpoch.incrementAndGet();
+ cache.invalidate(ident, type);
+ }
+
+ @VisibleForTesting
+ void clearCache() {
+ cacheInvalidationEpoch.incrementAndGet();
+ cache.clear();
+ }
}
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/TestRelationalEntityStoreBatchGetLateFill.java
b/core/src/test/java/org/apache/gravitino/storage/relational/TestRelationalEntityStoreBatchGetLateFill.java
new file mode 100644
index 0000000000..8ac5da1689
--- /dev/null
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/TestRelationalEntityStoreBatchGetLateFill.java
@@ -0,0 +1,260 @@
+/*
+ * 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.storage.relational;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+
+import java.time.Instant;
+import java.util.List;
+import org.apache.commons.lang3.reflect.FieldUtils;
+import org.apache.gravitino.Config;
+import org.apache.gravitino.Entity;
+import org.apache.gravitino.HasIdentifier;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.Namespace;
+import org.apache.gravitino.cache.CaffeineEntityCache;
+import org.apache.gravitino.cache.EntityCache;
+import org.apache.gravitino.cache.EntityCacheKey;
+import org.apache.gravitino.meta.AuditInfo;
+import org.apache.gravitino.meta.RoleEntity;
+import org.apache.gravitino.meta.TableEntity;
+import org.apache.gravitino.storage.relational.po.cache.EntityChangeRecord;
+import org.apache.gravitino.storage.relational.po.cache.OperateType;
+import org.apache.gravitino.utils.TestUtil;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+/**
+ * Verifies that {@link RelationalEntityStore#batchGet} cannot write an entity
back into the cache
+ * after that entity was invalidated while the backend read was in flight (the
late-fill race).
+ *
+ * <p>The tests trigger invalidation during the backend read or during the
cache write, so both
+ * sides of the write-back check are covered without relying on thread timing.
+ */
+public class TestRelationalEntityStoreBatchGetLateFill {
+
+ private static final Namespace SCHEMA_NS = Namespace.of("metalake",
"catalog", "schema");
+
+ private RelationalEntityStore store;
+ private RelationalBackend backend;
+ private CaffeineEntityCache cache;
+
+ @BeforeEach
+ void setUp() throws IllegalAccessException {
+ store = new RelationalEntityStore();
+ backend = Mockito.mock(RelationalBackend.class);
+ cache = Mockito.spy(new CaffeineEntityCache(new Config() {}));
+ FieldUtils.writeField(store, "backend", backend, true);
+ FieldUtils.writeField(store, "cache", cache, true);
+ }
+
+ private static EntityChangeRecord dropRecord(NameIdentifier ident,
Entity.EntityType type) {
+ return new EntityChangeRecord(
+ 1L,
+ ident.namespace().level(0),
+ type.name(),
+ EntityChangeLogNameIdentifierCodec.encode(ident),
+ OperateType.DROP,
+ 0L);
+ }
+
+ @Test
+ void testBatchGetWritesBackWhenNoInvalidationHappens() {
+ TableEntity table = TestUtil.getTestTableEntity(1L, "t1", SCHEMA_NS);
+ Mockito.when(backend.batchGet(any(),
eq(Entity.EntityType.TABLE))).thenReturn(List.of(table));
+
+ List<TableEntity> result =
+ store.batchGet(List.of(table.nameIdentifier()),
Entity.EntityType.TABLE, TableEntity.class);
+
+ Assertions.assertEquals(List.of(table), result);
+ Assertions.assertTrue(cache.contains(table.nameIdentifier(),
Entity.EntityType.TABLE));
+ }
+
+ @Test
+ void testBatchGetSkipsWriteBackWhenChangeLogInvalidatesDuringBackendRead() {
+ TableEntity table = TestUtil.getTestTableEntity(1L, "t1", SCHEMA_NS);
+ NameIdentifier ident = table.nameIdentifier();
+ EntityChangeLogListener poller = store.newCacheChangeLogListener();
+ Mockito.when(backend.batchGet(any(), eq(Entity.EntityType.TABLE)))
+ .thenAnswer(
+ invocation -> {
+ poller.onEntityChange(List.of(dropRecord(ident,
Entity.EntityType.TABLE)));
+ return List.of(table);
+ });
+
+ List<TableEntity> result =
+ store.batchGet(List.of(ident), Entity.EntityType.TABLE,
TableEntity.class);
+
+ Assertions.assertEquals(List.of(table), result);
+ Assertions.assertFalse(
+ cache.contains(ident, Entity.EntityType.TABLE),
+ "a value invalidated during the backend read must not be written
back");
+ }
+
+ @Test
+ void testBatchGetSkipsWriteBackWhenLocalDeleteInvalidatesDuringBackendRead()
{
+ TableEntity table = TestUtil.getTestTableEntity(1L, "t1", SCHEMA_NS);
+ NameIdentifier ident = table.nameIdentifier();
+ Mockito.when(backend.batchGet(any(), eq(Entity.EntityType.TABLE)))
+ .thenAnswer(
+ invocation -> {
+ store.delete(ident, Entity.EntityType.TABLE, false);
+ return List.of(table);
+ });
+
+ store.batchGet(List.of(ident), Entity.EntityType.TABLE, TableEntity.class);
+
+ Assertions.assertFalse(cache.contains(ident, Entity.EntityType.TABLE));
+ }
+
+ @Test
+ void
testBatchGetSkipsWriteBackWhenUnrelatedKeyInvalidatesDuringBackendRead() {
+ TableEntity table = TestUtil.getTestTableEntity(1L, "t1", SCHEMA_NS);
+ NameIdentifier unrelatedIdent = NameIdentifier.of(SCHEMA_NS, "t2");
+ Mockito.when(backend.batchGet(any(), eq(Entity.EntityType.TABLE)))
+ .thenAnswer(
+ invocation -> {
+ store.delete(unrelatedIdent, Entity.EntityType.TABLE, false);
+ return List.of(table);
+ });
+
+ store.batchGet(List.of(table.nameIdentifier()), Entity.EntityType.TABLE,
TableEntity.class);
+
+ Assertions.assertFalse(cache.contains(table.nameIdentifier(),
Entity.EntityType.TABLE));
+ }
+
+ @Test
+ void testBatchGetDirectClearDuringBackendRead() {
+ TableEntity table = TestUtil.getTestTableEntity(1L, "t1", SCHEMA_NS);
+ NameIdentifier ident = table.nameIdentifier();
+ Mockito.when(backend.batchGet(any(), eq(Entity.EntityType.TABLE)))
+ .thenAnswer(
+ invocation -> {
+ store.clearCache();
+ return List.of(table);
+ });
+
+ store.batchGet(List.of(ident), Entity.EntityType.TABLE, TableEntity.class);
+
+ Assertions.assertFalse(cache.contains(ident, Entity.EntityType.TABLE));
+ }
+
+ @Test
+ void testBatchGetRemovesValueWrittenAfterClear() throws
IllegalAccessException {
+ TableEntity table = TestUtil.getTestTableEntity(1L, "t1", SCHEMA_NS);
+ RecordingCache recordingCache = new RecordingCache();
+ recordingCache.beforePut = store::clearCache;
+ FieldUtils.writeField(store, "cache", recordingCache, true);
+ Mockito.when(backend.batchGet(any(),
eq(Entity.EntityType.TABLE))).thenReturn(List.of(table));
+
+ List<TableEntity> result =
+ store.batchGet(List.of(table.nameIdentifier()),
Entity.EntityType.TABLE, TableEntity.class);
+
+ Assertions.assertEquals(List.of(table), result);
+ Assertions.assertTrue(recordingCache.beforePutRan);
+ Assertions.assertFalse(
+ recordingCache.contains(table.nameIdentifier(),
Entity.EntityType.TABLE));
+ }
+
+ @Test
+ void testBatchGetDoesNotLockNonCacheableRole() throws IllegalAccessException
{
+ RoleEntity role =
+ RoleEntity.builder()
+ .withId(1L)
+ .withName("role")
+ .withNamespace(Namespace.of("metalake"))
+ .withAuditInfo(
+
AuditInfo.builder().withCreator("creator").withCreateTime(Instant.EPOCH).build())
+ .build();
+ RecordingCache recordingCache = new RecordingCache();
+ FieldUtils.writeField(store, "cache", recordingCache, true);
+ Mockito.when(backend.batchGet(any(),
eq(Entity.EntityType.ROLE))).thenReturn(List.of(role));
+
+ List<RoleEntity> result =
+ store.batchGet(List.of(role.nameIdentifier()), Entity.EntityType.ROLE,
RoleEntity.class);
+
+ Assertions.assertEquals(List.of(role), result);
+ Assertions.assertFalse(recordingCache.cacheLockUsed);
+ Assertions.assertTrue(recordingCache.keyChangeHookCalled);
+ }
+
+ @Test
+ void
testBatchGetSkipsWriteBackWhenListenerFallsBackToClearDuringBackendRead() {
+ TableEntity table = TestUtil.getTestTableEntity(1L, "t1", SCHEMA_NS);
+ NameIdentifier ident = table.nameIdentifier();
+ EntityChangeLogListener poller = store.newCacheChangeLogListener();
+ // A failed targeted invalidation makes the listener clear the whole
cache; that must fence
+ // in-flight fills too.
+ Mockito.doThrow(new RuntimeException("boom"))
+ .when(cache)
+ .invalidate(ident, Entity.EntityType.TABLE);
+ Mockito.when(backend.batchGet(any(), eq(Entity.EntityType.TABLE)))
+ .thenAnswer(
+ invocation -> {
+ poller.onEntityChange(List.of(dropRecord(ident,
Entity.EntityType.TABLE)));
+ return List.of(table);
+ });
+
+ store.batchGet(List.of(ident), Entity.EntityType.TABLE, TableEntity.class);
+
+ Mockito.verify(cache).clear();
+ Assertions.assertFalse(cache.contains(ident, Entity.EntityType.TABLE));
+ }
+
+ private static class RecordingCache extends CaffeineEntityCache {
+ private boolean cacheLockUsed;
+ private boolean keyChangeHookCalled;
+ private boolean beforePutRan;
+ private Runnable beforePut;
+
+ RecordingCache() {
+ super(new Config() {});
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public <E extends Exception> void withCacheLock(
+ EntityCacheKey key, EntityCache.ThrowingRunnable<E> action) throws E {
+ cacheLockUsed = true;
+ super.withCacheLock(key, action);
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public <E extends Entity & HasIdentifier> void invalidateOnKeyChange(E
entity) {
+ keyChangeHookCalled = true;
+ super.invalidateOnKeyChange(entity);
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ protected <E extends Entity & HasIdentifier> void doPut(E entity) {
+ if (beforePut != null) {
+ Runnable action = beforePut;
+ beforePut = null;
+ action.run();
+ beforePutRan = true;
+ }
+ super.doPut(entity);
+ }
+ }
+}
