Skip to content

Commit 28fd283

Browse files
yuqi1129yuqi
yuqi1129
and
yuqi
authored
[#13893] fix(cache): Recover late change logs and fence cache publication (#13831)
### What changes were proposed in this pull request? - Replay recent change-log rows behind the cursor with bounded ID deduplication, diagnostics, and explicit listener opt-in. - Start a replay sweep at most once per `max(pollIntervalSecs, replayWindowSecs / 4)` (15s by default), and expose `replay-records-total`, `replay-failures-total`, and `dedup-evictions-total` metrics. - Fence point-read/insert cache publication without requiring reentrant SPI locks, and document that `invalidateOnKeyChange` may run under the entity's own cache lock. - Invalidate local and replayed catalog events while preserving operation leases; update configuration and metrics documentation. ### Why are the changes needed? ID allocation order differs from commit order. Concurrent cache publication and name-based local catalog markers can also leave stale entries. Fix: #13893 Part of #13828; this PR implements its three P0 fixes and leaves the remaining EPIC work open. ### Does this PR introduce _any_ user-facing change? Adds `gravitino.entityChangeLog.replayWindowSecs` (default `60`; `0` disables replay) and `gravitino.entityChangeLog.replayMaxEntries` (default `100000`), plus three `entity-change-log` counters: `replay-records-total`, `replay-failures-total`, and `dedup-evictions-total`. Existing third-party listeners remain incremental-only. No schema or cache-TTL change. Replay adds database reads and possible catalog reloads. Its window uses recorded `created_at`, including PostgreSQL transaction-start timestamps; long transactions, pruned history, or overload can still lose invalidations. Each sweep rereads the whole window. Throttling sweep starts to a quarter of the window bounds this to about four reads per row per server, instead of window/poll interval (20 by default). The cost: late-commit recovery can take up to one extra sweep interval, and a row is guaranteed to be replayed only if it becomes visible within the window minus one sweep interval (45s by default). A finer gap/low-water-mark scheme was considered and deferred for maintainability. Replay-only full-page cycles still perform incremental, tail, and replay queries. A late row behind an already scanned page waits for the next sweep; the replay window is a history range, not a delivery deadline. ### How was this patch tested? - Full core suite: 2,761 tests, zero failures/errors, two existing skips. Final focused suite: 39 tests, including the added timing regression, all passed. Previously validated unchanged server-common: 485 tests, all passed. - Local H2/MySQL/PostgreSQL late-commit probes and shared-PostgreSQL three-JVM checks for ten cacheable entity types passed. - Three P0 regressions fail when old behavior is restored; the non-reentrant SPI regression fails on the initial implementation and passes after the compatibility fix. - Default 3s polling: real PostgreSQL/three-JVM probes converged in 2.702s (incremental) and 3.101s (late commit) with small history; the late-commit figure predates sweep throttling, which can add up to one sweep interval. A controlled ordinary-incremental budget test delivers in 5.7s with 0.9s incremental work and 0.9s replay work; moving delivery after replay fails at 6.6s. A separate multi-page late-replay probe takes 7.5s, so these checks do not establish a universal 6s replay deadline. - Sweep throttling and metrics: 37 `TestEntityChangeLogPoller` tests pass, including new throttle, poll-interval floor, and eviction-counter tests; the two throttle tests fail with a zero sweep interval. `TestCacheCommitOrdering` passes on H2, MySQL, and PostgreSQL. - Catalog modules that mock `Config` (generic lakehouse, fileset, Kafka, model) now stub the replay settings: 29/232/15/37 tests passed; the JMH benchmark compiles. - After merging main (#13792): the new `TestRelationalEntityStoreLockWaitTimeout` mocks `Config` and now stubs the replay settings (all 7 cases NPE without them). 14 affected core classes on H2: 204 tests passed. - Review round (5a1312f, 6a242d6): the first replay sweep after startup logs at INFO and later sweeps keep the late-commit WARN (pinned in tests; the INFO assertion fails without the change); a failed mid-sweep page resumes from the saved page cursor (fails if the cursor is reset). `TestEntityChangeLogPoller` (39) plus related store/cache classes: 115 tests passed on H2. - Spotless, compilation, and diff checks passed. Unit-suite command: ```bash LC_ALL=en_US.UTF-8 LANG=en_US.UTF-8 ./gradlew :core:test :server-common:test -PskipITs -PskipDockerTests=true ``` --------- Signed-off-by: yuqi <yuqi@datastrato.com> Co-authored-by: yuqi <yuqi4733@protonmail.com>
1 parent 654106b commit 28fd283

41 files changed

Lines changed: 1974 additions & 292 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎catalogs/catalog-fileset/src/test/java/org/apache/gravitino/catalog/fileset/TestFilesetCatalogOperations.java‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,10 +19,14 @@
1919
package org.apache.gravitino.catalog.fileset;
2020

2121
import static org.apache.gravitino.Configs.DEFAULT_ENTITY_CHANGE_LOG_POLL_BATCH_SIZE;
22+
import static org.apache.gravitino.Configs.DEFAULT_ENTITY_CHANGE_LOG_REPLAY_MAX_ENTRIES;
23+
import static org.apache.gravitino.Configs.DEFAULT_ENTITY_CHANGE_LOG_REPLAY_WINDOW_SECS;
2224
import static org.apache.gravitino.Configs.DEFAULT_ENTITY_RELATIONAL_STORE;
2325
import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS;
2426
import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_POLL_BATCH_SIZE;
2527
import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS;
28+
import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_REPLAY_MAX_ENTRIES;
29+
import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_REPLAY_WINDOW_SECS;
2630
import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_RETENTION_SECS;
2731
import static org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_DRIVER;
2832
import static org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_MAX_CONNECTIONS;
@@ -290,6 +294,10 @@ public static void setUp() throws IOException, IllegalAccessException {
290294
when(config.get(ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS)).thenReturn(3L);
291295
when(config.get(ENTITY_CHANGE_LOG_POLL_BATCH_SIZE))
292296
.thenReturn(DEFAULT_ENTITY_CHANGE_LOG_POLL_BATCH_SIZE);
297+
when(config.get(ENTITY_CHANGE_LOG_REPLAY_WINDOW_SECS))
298+
.thenReturn(DEFAULT_ENTITY_CHANGE_LOG_REPLAY_WINDOW_SECS);
299+
when(config.get(ENTITY_CHANGE_LOG_REPLAY_MAX_ENTRIES))
300+
.thenReturn(DEFAULT_ENTITY_CHANGE_LOG_REPLAY_MAX_ENTRIES);
293301
when(config.get(ENTITY_CHANGE_LOG_RETENTION_SECS)).thenReturn(24 * 60 * 60L);
294302
when(config.get(ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS)).thenReturn(60 * 60L);
295303
// Fix cache config for test

‎catalogs/catalog-kafka/src/test/java/org/apache/gravitino/catalog/kafka/TestKafkaCatalogOperations.java‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,10 +20,14 @@
2020

2121
import static org.apache.gravitino.Catalog.Type.MESSAGING;
2222
import static org.apache.gravitino.Configs.DEFAULT_ENTITY_CHANGE_LOG_POLL_BATCH_SIZE;
23+
import static org.apache.gravitino.Configs.DEFAULT_ENTITY_CHANGE_LOG_REPLAY_MAX_ENTRIES;
24+
import static org.apache.gravitino.Configs.DEFAULT_ENTITY_CHANGE_LOG_REPLAY_WINDOW_SECS;
2325
import static org.apache.gravitino.Configs.DEFAULT_ENTITY_RELATIONAL_STORE;
2426
import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS;
2527
import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_POLL_BATCH_SIZE;
2628
import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS;
29+
import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_REPLAY_MAX_ENTRIES;
30+
import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_REPLAY_WINDOW_SECS;
2731
import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_RETENTION_SECS;
2832
import static org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_DRIVER;
2933
import static org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_MAX_CONNECTIONS;
@@ -171,6 +175,10 @@ public static void setUp() throws IOException, IllegalAccessException {
171175
when(config.get(ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS)).thenReturn(3L);
172176
when(config.get(ENTITY_CHANGE_LOG_POLL_BATCH_SIZE))
173177
.thenReturn(DEFAULT_ENTITY_CHANGE_LOG_POLL_BATCH_SIZE);
178+
when(config.get(ENTITY_CHANGE_LOG_REPLAY_WINDOW_SECS))
179+
.thenReturn(DEFAULT_ENTITY_CHANGE_LOG_REPLAY_WINDOW_SECS);
180+
when(config.get(ENTITY_CHANGE_LOG_REPLAY_MAX_ENTRIES))
181+
.thenReturn(DEFAULT_ENTITY_CHANGE_LOG_REPLAY_MAX_ENTRIES);
174182
when(config.get(ENTITY_CHANGE_LOG_RETENTION_SECS)).thenReturn(24 * 60 * 60L);
175183
when(config.get(ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS)).thenReturn(60 * 60L);
176184
// Fix cache config for test

‎catalogs/catalog-lakehouse-generic/src/test/java/org/apache/gravitino/catalog/lakehouse/generic/TestGenericCatalogOperations.java‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,10 +19,14 @@
1919
package org.apache.gravitino.catalog.lakehouse.generic;
2020

2121
import static org.apache.gravitino.Configs.DEFAULT_ENTITY_CHANGE_LOG_POLL_BATCH_SIZE;
22+
import static org.apache.gravitino.Configs.DEFAULT_ENTITY_CHANGE_LOG_REPLAY_MAX_ENTRIES;
23+
import static org.apache.gravitino.Configs.DEFAULT_ENTITY_CHANGE_LOG_REPLAY_WINDOW_SECS;
2224
import static org.apache.gravitino.Configs.DEFAULT_ENTITY_RELATIONAL_STORE;
2325
import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS;
2426
import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_POLL_BATCH_SIZE;
2527
import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS;
28+
import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_REPLAY_MAX_ENTRIES;
29+
import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_REPLAY_WINDOW_SECS;
2630
import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_RETENTION_SECS;
2731
import static org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_DRIVER;
2832
import static org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_MAX_CONNECTIONS;
@@ -130,6 +134,10 @@ public static void setUp() throws IOException, IllegalAccessException {
130134
when(config.get(ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS)).thenReturn(3L);
131135
when(config.get(ENTITY_CHANGE_LOG_POLL_BATCH_SIZE))
132136
.thenReturn(DEFAULT_ENTITY_CHANGE_LOG_POLL_BATCH_SIZE);
137+
when(config.get(ENTITY_CHANGE_LOG_REPLAY_WINDOW_SECS))
138+
.thenReturn(DEFAULT_ENTITY_CHANGE_LOG_REPLAY_WINDOW_SECS);
139+
when(config.get(ENTITY_CHANGE_LOG_REPLAY_MAX_ENTRIES))
140+
.thenReturn(DEFAULT_ENTITY_CHANGE_LOG_REPLAY_MAX_ENTRIES);
133141
when(config.get(ENTITY_CHANGE_LOG_RETENTION_SECS)).thenReturn(24 * 60 * 60L);
134142
when(config.get(ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS)).thenReturn(60 * 60L);
135143
Mockito.when(config.get(Configs.CACHE_ENABLED)).thenReturn(false);

‎catalogs/catalog-model/src/test/java/org/apache/gravtitino/catalog/model/TestModelCatalogOperations.java‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,10 +19,14 @@
1919
package org.apache.gravtitino.catalog.model;
2020

2121
import static org.apache.gravitino.Configs.DEFAULT_ENTITY_CHANGE_LOG_POLL_BATCH_SIZE;
22+
import static org.apache.gravitino.Configs.DEFAULT_ENTITY_CHANGE_LOG_REPLAY_MAX_ENTRIES;
23+
import static org.apache.gravitino.Configs.DEFAULT_ENTITY_CHANGE_LOG_REPLAY_WINDOW_SECS;
2224
import static org.apache.gravitino.Configs.DEFAULT_ENTITY_RELATIONAL_STORE;
2325
import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS;
2426
import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_POLL_BATCH_SIZE;
2527
import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS;
28+
import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_REPLAY_MAX_ENTRIES;
29+
import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_REPLAY_WINDOW_SECS;
2630
import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_RETENTION_SECS;
2731
import static org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_DRIVER;
2832
import static org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_MAX_CONNECTIONS;
@@ -138,6 +142,10 @@ public static void setUp() throws IOException, IllegalAccessException {
138142
when(config.get(ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS)).thenReturn(3L);
139143
when(config.get(ENTITY_CHANGE_LOG_POLL_BATCH_SIZE))
140144
.thenReturn(DEFAULT_ENTITY_CHANGE_LOG_POLL_BATCH_SIZE);
145+
when(config.get(ENTITY_CHANGE_LOG_REPLAY_WINDOW_SECS))
146+
.thenReturn(DEFAULT_ENTITY_CHANGE_LOG_REPLAY_WINDOW_SECS);
147+
when(config.get(ENTITY_CHANGE_LOG_REPLAY_MAX_ENTRIES))
148+
.thenReturn(DEFAULT_ENTITY_CHANGE_LOG_REPLAY_MAX_ENTRIES);
141149
when(config.get(ENTITY_CHANGE_LOG_RETENTION_SECS)).thenReturn(24 * 60 * 60L);
142150
when(config.get(ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS)).thenReturn(60 * 60L);
143151
// Fix cache config for test

‎core/src/jmh/java/org/apache/gravitino/cache/it/AbstractEntityStorageBenchmark.java‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,11 +20,15 @@
2020
package org.apache.gravitino.cache.it;
2121

2222
import static org.apache.gravitino.Configs.DEFAULT_ENTITY_CHANGE_LOG_POLL_BATCH_SIZE;
23+
import static org.apache.gravitino.Configs.DEFAULT_ENTITY_CHANGE_LOG_REPLAY_MAX_ENTRIES;
24+
import static org.apache.gravitino.Configs.DEFAULT_ENTITY_CHANGE_LOG_REPLAY_WINDOW_SECS;
2325
import static org.apache.gravitino.Configs.DEFAULT_ENTITY_RELATIONAL_STORE;
2426
import static org.apache.gravitino.Configs.DEFAULT_RELATIONAL_JDBC_BACKEND_MAX_IDLE_CONNECTIONS;
2527
import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS;
2628
import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_POLL_BATCH_SIZE;
2729
import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS;
30+
import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_REPLAY_MAX_ENTRIES;
31+
import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_REPLAY_WINDOW_SECS;
2832
import static org.apache.gravitino.Configs.ENTITY_CHANGE_LOG_RETENTION_SECS;
2933
import static org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_DRIVER;
3034
import static org.apache.gravitino.Configs.ENTITY_RELATIONAL_JDBC_BACKEND_MAX_CONNECTIONS;
@@ -302,6 +306,10 @@ protected void initStore() throws IOException {
302306
Mockito.when(config.get(ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS)).thenReturn(3L);
303307
Mockito.when(config.get(ENTITY_CHANGE_LOG_POLL_BATCH_SIZE))
304308
.thenReturn(DEFAULT_ENTITY_CHANGE_LOG_POLL_BATCH_SIZE);
309+
Mockito.when(config.get(ENTITY_CHANGE_LOG_REPLAY_WINDOW_SECS))
310+
.thenReturn(DEFAULT_ENTITY_CHANGE_LOG_REPLAY_WINDOW_SECS);
311+
Mockito.when(config.get(ENTITY_CHANGE_LOG_REPLAY_MAX_ENTRIES))
312+
.thenReturn(DEFAULT_ENTITY_CHANGE_LOG_REPLAY_MAX_ENTRIES);
305313
Mockito.when(config.get(ENTITY_CHANGE_LOG_RETENTION_SECS)).thenReturn(24 * 60 * 60L);
306314
Mockito.when(config.get(ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS)).thenReturn(60 * 60L);
307315
Mockito.when(config.get(VERSION_RETENTION_COUNT)).thenReturn(1L);

‎core/src/main/java/org/apache/gravitino/Configs.java‎

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -233,6 +233,32 @@ private Configs() {}
233233
.checkValue(value -> value > 0, ConfigConstants.POSITIVE_NUMBER_ERROR_MSG)
234234
.createWithDefault(DEFAULT_ENTITY_CHANGE_LOG_POLL_BATCH_SIZE);
235235

236+
/** Default recent-history replay window in seconds. */
237+
public static final long DEFAULT_ENTITY_CHANGE_LOG_REPLAY_WINDOW_SECS = 60L;
238+
239+
/** Default maximum locally remembered event IDs. */
240+
public static final int DEFAULT_ENTITY_CHANGE_LOG_REPLAY_MAX_ENTRIES = 100000;
241+
242+
/** Window for replaying recent change records to catch out-of-order commits. */
243+
public static final ConfigEntry<Long> ENTITY_CHANGE_LOG_REPLAY_WINDOW_SECS =
244+
new ConfigBuilder("gravitino.entityChangeLog.replayWindowSecs")
245+
.doc(
246+
"Recent change log replay window in seconds; 0 disables replay. This bounds recorded created_at age, not commit age. "
247+
+ "Replay sweeps start at most once per max(pollIntervalSecs, replayWindowSecs / 4)")
248+
.version(ConfigConstants.VERSION_2_0_0)
249+
.longConf()
250+
.checkValue(value -> value >= 0 && value <= 86400, "Must be between 0 and 86400 seconds")
251+
.createWithDefault(DEFAULT_ENTITY_CHANGE_LOG_REPLAY_WINDOW_SECS);
252+
253+
/** Maximum event IDs retained locally to suppress repeated replay invalidations. */
254+
public static final ConfigEntry<Integer> ENTITY_CHANGE_LOG_REPLAY_MAX_ENTRIES =
255+
new ConfigBuilder("gravitino.entityChangeLog.replayMaxEntries")
256+
.doc("Maximum locally remembered event IDs for recent change log replay deduplication")
257+
.version(ConfigConstants.VERSION_2_0_0)
258+
.intConf()
259+
.checkValue(value -> value > 0, ConfigConstants.POSITIVE_NUMBER_ERROR_MSG)
260+
.createWithDefault(DEFAULT_ENTITY_CHANGE_LOG_REPLAY_MAX_ENTRIES);
261+
236262
public static final ConfigEntry<Long> ENTITY_CHANGE_LOG_RETENTION_SECS =
237263
new ConfigBuilder("gravitino.entityChangeLog.retentionSecs")
238264
.doc("The retention time in seconds for entity change logs. Set 0 to disable cleanup")

‎core/src/main/java/org/apache/gravitino/cache/BaseEntityCache.java‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -107,10 +107,10 @@ public static boolean isCacheable(Entity.EntityType type) {
107107
public final <E extends Entity & HasIdentifier> void put(E entity) {
108108
Preconditions.checkArgument(entity != null, "Entity cannot be null");
109109

110-
// Called before the cacheability check and before any subclass takes this entity's lock: it
111-
// may take another key's lock (e.g. the model key when inserting a model version), and nesting
112-
// two entity locks could deadlock. A non-cacheable entity can still invalidate a cacheable one,
113-
// so this runs for every entity.
110+
// Called before the cacheability check: a non-cacheable entity can still invalidate a cacheable
111+
// one, so this runs for every entity. The caller may already hold this entity's lock, so the
112+
// hook must not take another entity's lock (see
113+
// SupportsEntityStoreCache#invalidateOnKeyChange).
114114
invalidateOnKeyChange(entity);
115115

116116
if (!isCacheable(entity.type())) {

‎core/src/main/java/org/apache/gravitino/cache/SupportsEntityStoreCache.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -89,6 +89,12 @@ <E extends Entity & HasIdentifier> Optional<E> getIfPresent(
8989
* <p>For example, inserting a {@code ModelVersion} may require invalidating the corresponding
9090
* {@code Model} entry in the cache to maintain consistency.
9191
*
92+
* <p>Callers may run {@link #put(Entity)}, and therefore this method, while holding the inserted
93+
* entity's own cache lock (see {@link EntityCache#withCacheLock(EntityCacheKey,
94+
* EntityCache.ThrowingRunnable)}); the relational entity store publishes reads and inserts that
95+
* way. Implementations must therefore not acquire another entity's cache lock here, since nesting
96+
* two entity locks can deadlock.
97+
*
9298
* @param entity The entity being inserted
9399
* @param <E> The type of the entity
94100
*/

‎core/src/main/java/org/apache/gravitino/catalog/CatalogChangeLogListener.java‎

Lines changed: 18 additions & 37 deletions
Original file line numberDiff line numberDiff line change
@@ -36,18 +36,17 @@
3636
* <p>This listener is called <em>synchronously</em> in the poller thread. Implementations must not
3737
* block or perform expensive I/O; only fast, in-memory cache invalidations are permitted.
3838
*
39-
* <p>The poller hands each batch to a listener only once, so a listener has to clean up after
40-
* itself when it fails. This one does what {@code EntityCacheChangeLogListener} and {@code
39+
* <p>The poller does not retry a failed callback, so a listener has to clean up after itself when
40+
* it fails. This one does what {@code EntityCacheChangeLogListener} and {@code
4141
* JcasbinChangeListener} do: if removing one catalog from the cache fails, it clears the whole
4242
* catalog cache, which also covers the entry it failed to remove and the rest of the batch. A row
4343
* that cannot be parsed is simply skipped, because it does not point at any catalog and so cannot
4444
* leave anything stale.
4545
*
46-
* <p>Before removing anything, the listener first goes through the whole batch and marks off the
47-
* changes this node made itself. Doing it in that order means a later failure cannot leave one of
48-
* those marks behind, which would otherwise make a future change from another node look like a
49-
* local one. If the clear itself fails, the exception goes up to the poller, which logs it at
50-
* {@code ERROR} and moves on, and the catalog stays stale until it expires.
46+
* <p>Every valid catalog event invalidates its entry, including local events and replayed rows. A
47+
* name-based local marker cannot reliably identify an originating transaction. Extra removals are
48+
* preferable to suppressing a remote mutation; operation leases protect wrappers in use. If the
49+
* clear itself fails, the exception goes to the poller for error logging.
5150
*
5251
* <p><b>What clearing costs:</b> dropping a catalog from the cache retires its {@code
5352
* CatalogWrapper}. Idle wrappers release their connection pools and {@code IsolatedClassLoader}s
@@ -72,9 +71,15 @@ public CatalogChangeLogListener(CatalogManager catalogManager) {
7271
this.catalogManager = catalogManager;
7372
}
7473

74+
/** {@inheritDoc} */
75+
@Override
76+
public boolean supportsReplay() {
77+
return true;
78+
}
79+
7580
@Override
7681
public void onEntityChange(List<EntityChangeRecord> changes) {
77-
List<CatalogInvalidation> remoteInvalidations = new ArrayList<>();
82+
List<CatalogInvalidation> invalidations = new ArrayList<>();
7883
for (EntityChangeRecord change : changes) {
7984
if (!isCatalogChange(change)) {
8085
continue;
@@ -88,50 +93,26 @@ public void onEntityChange(List<EntityChangeRecord> changes) {
8893
}
8994
NameIdentifier ident = identOpt.get();
9095

91-
boolean localMutation;
92-
try {
93-
localMutation = catalogManager.consumeLocalMutation(ident);
94-
} catch (RuntimeException e) {
95-
// We could not tell whether this change came from this node or another one. The name is
96-
// valid, so assume it came from another node: the worst case is one extra cache removal,
97-
// while skipping it could leave an old catalog cached forever.
98-
LOG.error(
99-
"Failed to check local mutation state for catalog {}, treating change log record id {} "
100-
+ "as remote to avoid serving stale metadata",
101-
ident,
102-
change.getId(),
103-
e);
104-
localMutation = false;
105-
}
106-
107-
if (localMutation) {
108-
LOG.debug(
109-
"Skipping catalog cache invalidation for local mutation: {}, change log id {}",
110-
ident,
111-
change.getId());
112-
continue;
113-
}
114-
115-
remoteInvalidations.add(new CatalogInvalidation(change, ident));
96+
invalidations.add(new CatalogInvalidation(change, ident));
11697
}
11798

118-
for (CatalogInvalidation invalidation : remoteInvalidations) {
99+
for (CatalogInvalidation invalidation : invalidations) {
119100
EntityChangeRecord change = invalidation.change;
120101
NameIdentifier ident = invalidation.ident;
121102
// INFO on purpose: dropping the catalog from the cache retires its wrapper and eventually
122103
// closes the connection pool and isolated classloader, either immediately or after active
123104
// leases finish. This is the main thing the change log does across nodes.
124105
LOG.info(
125-
"Invalidating catalog cache for {} due to a remote {} recorded in change log id {}",
106+
"Invalidating catalog cache for {} due to a {} recorded in change log id {}",
126107
ident,
127108
change.getOperateType(),
128109
change.getId());
129110

130111
try {
131112
catalogManager.getCatalogCache().invalidate(ident);
132113
} catch (RuntimeException e) {
133-
// This batch will never be sent again, so giving up here would keep serving the old
134-
// catalog until it expires on its own. Clear the whole cache instead; active operation
114+
// Failed callbacks have no guaranteed retry; bounded replay may not redeliver this row.
115+
// Clear the whole cache instead of serving an old catalog until expiry; active operation
135116
// leases defer resource cleanup until those operations finish.
136117
LOG.error(
137118
"Failed to evict catalog {} for change log id {}, clearing the whole catalog cache to "

0 commit comments

Comments
 (0)