Repository navigation
[#13893] fix(cache): Recover late change logs and fence cache publication - #13831
Conversation
…ache publication Signed-off-by: yuqi <yuqi@datastrato.com>
…ility Signed-off-by: yuqi <yuqi@datastrato.com>
…verview Signed-off-by: yuqi <yuqi@datastrato.com>
There was a problem hiding this comment.
Warning
Copilot couldn't run its full agentic review because it didn't start before the timeout. Make sure your repository has a runner available, or add a copilot-code-review.yml file specifying one with the runs-on attribute. See the docs for more details.
Copilot review overview
4 open findings
rememberedIds.putIfAbsent(change.getId(), now)never refreshes the timestamp for IDs that are… · NewshutdownNow()but don’tawaitTermination(). If a… · New
What changed in this PR
Implements bounded recent-history replay for the entity change-log poller and adds epoch-fenced cache publication to prevent stale cache entries from surviving invalidations due to commit ordering and publication races.
Changes:
- Add recent-history replay (window + bounded dedup) with explicit listener opt-in via
supportsReplay(), plus diagnostics/metrics/doc updates. - Fence cache publication in
RelationalEntityStoreusing an invalidation epoch to prevent raced point-read/insert publications. - Update built-in listeners (catalog/entity/jcasbin) and add regression tests covering late commit ordering, replay behavior, and cache publication races.
| File | Description |
|---|---|
| server-common/src/test/java/org/apache/gravitino/server/authorization/jcasbin/TestJcasbinChangePoller.java | Adds test ensuring replay opt-in listeners tolerate repeated invalidations. |
| server-common/src/main/java/org/apache/gravitino/server/authorization/jcasbin/JcasbinChangeListener.java | Opts Jcasbin listener into replay; updates failure-recovery docstring. |
| docs/metrics.md | Documents replay effects on polling cycles and metrics semantics. |
| docs/gravitino-server-config.md | Documents new replay configs and operational behavior/limits. |
| core/src/test/java/org/apache/gravitino/tag/TestTagManager.java | Mocks new replay config values in tests. |
| core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/base/TestEntityChangeLogMapper.java | Adds tests for recent replay SQL pagination/clock sources. |
| core/src/test/java/org/apache/gravitino/storage/relational/TestRelationalEntityStoreHierarchicalCache.java | Mocks new replay config values in cache tests. |
| core/src/test/java/org/apache/gravitino/storage/relational/TestRelationalEntityStore.java | Adds regression test for epoch-fenced invalidation behavior with uncached entities. |
| core/src/test/java/org/apache/gravitino/storage/relational/TestEntityChangeLogPoller.java | Adds extensive unit tests for replay, dedup, rate-limiting, and diagnostics logging. |
| core/src/test/java/org/apache/gravitino/storage/relational/TestCacheCommitOrdering.java | Adds JDBC regression test for low-ID late commit reaching peer cache. |
| core/src/test/java/org/apache/gravitino/storage/AbstractEntityStorageTest.java | Mocks new replay config values for store tests. |
| core/src/test/java/org/apache/gravitino/stats/TestStatisticManager.java | Mocks new replay config values for stats tests. |
| core/src/test/java/org/apache/gravitino/policy/TestPolicyManager.java | Mocks new replay config values for policy tests. |
| core/src/test/java/org/apache/gravitino/hook/TestFilesetHookDispatcher.java | Mocks new replay config values for hook tests. |
| core/src/test/java/org/apache/gravitino/catalog/TestCatalogWrapperLease.java | Validates repeated catalog invalidation safety under active leases. |
| core/src/test/java/org/apache/gravitino/catalog/TestCatalogMutationMarkerRace.java | Adds regression probe removing reliance on local mutation markers. |
| core/src/test/java/org/apache/gravitino/catalog/TestCatalogManager.java | Updates tests to expect conservative invalidation (including local/replayed rows). |
| core/src/test/java/org/apache/gravitino/catalog/TestCatalogChangeLogListener.java | Updates listener tests for “always invalidate catalog” semantics. |
| core/src/test/java/org/apache/gravitino/cache/TestEntityCachePublicationRace.java | Adds deterministic tests for cache publication fencing and non-reentrant SPI locks. |
| core/src/test/java/org/apache/gravitino/authorization/TestOwnerManager.java | Mocks new replay config values for authorization tests. |
| core/src/test/java/org/apache/gravitino/authorization/TestAccessControlManager.java | Mocks new replay config values for access control tests. |
| core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/EntityChangeLogPostgreSQLProvider.java | Exposes DB clock expression override for replay queries. |
| core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/EntityChangeLogBaseSQLProvider.java | Adds selectRecentEntityChanges query and DB clock abstraction. |
| core/src/main/java/org/apache/gravitino/storage/relational/mapper/EntityChangeLogSQLProviderFactory.java | Adds factory method for recent replay query. |
| core/src/main/java/org/apache/gravitino/storage/relational/mapper/EntityChangeLogMapper.java | Adds mapper method for recent replay query. |
| core/src/main/java/org/apache/gravitino/storage/relational/RelationalEntityStore.java | Adds epoch-fenced publication for put/get/batchGet and avoids re-entrant locking. |
| core/src/main/java/org/apache/gravitino/storage/relational/EntityChangeLogPoller.java | Implements recent-history replay, bounded dedup, capability gating, and diagnostics. |
| core/src/main/java/org/apache/gravitino/storage/relational/EntityChangeLogListener.java | Adds supportsReplay() opt-in API with idempotency contract. |
| core/src/main/java/org/apache/gravitino/storage/relational/EntityCacheChangeLogListener.java | Opts entity cache listener into replay. |
| core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java | Removes local mutation markers; relies on conservative invalidation + leases. |
| core/src/main/java/org/apache/gravitino/catalog/CatalogChangeLogListener.java | Opts into replay and invalidates for all valid catalog events (local/remote/replayed). |
| core/src/main/java/org/apache/gravitino/Configs.java | Introduces replay window and max-entry config keys with defaults/validation. |
🧠 Review effort: Lite
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
Catalog operation tests and the entity-store benchmark mock Config, so the new replay window and dedup capacity entries returned null and RelationalEntityStore.initialize failed with an NPE.
…lay metrics Each replay sweep rereads the whole window, so starting one every poll read every recent row about window/interval (20 by default) times per server. Start a new sweep at most once per max(pollIntervalSecs, replayWindowSecs / 4); pages of a sweep in progress and failure retries are unchanged. Add replay-records-total, replay-failures-total and dedup-evictions-total so replay health is visible in metrics, not only in logs.
Code Coverage Report
Files
|
Keep the publication epoch sample in put() alongside the lock-wait timeout mapping from apache#13792, and stub the replay settings in the new TestRelationalEntityStoreLockWaitTimeout mocked Config.
yuqi1129
changed the title
jerryshao
left a comment
jerryshao
left a comment
There was a problem hiding this comment.
Verdict: ship it
Reviewed the full change on e791a8f (39 files, +1873/-279) in a local checkout of pull/13831/head, reading every changed main-source file end to end plus BaseEntityCache, CaffeineEntityCache, SegmentedLock, the three listener implementations, the change-log schema scripts, and the new tests. No correctness defect survived verification: the replay keyset query is sound (moving the window lower bound forward during a multi-page sweep can only drop rows already paged past, and the (created_at, id) tiebreak plus the fixed upperId ceiling keep pages stable), the dedup lifetime argument in discussion_r4291682643 holds for any fixed clock offset (an entry is inserted at or after the row's created_at and expires one window later, so it always outlives the row's presence in the query), and all three registered listeners are idempotent, so the replay opt-in is safe. The markLocalMutation removal leaves no stale references anywhere in the tree, and the CatalogWrapper lease/retire protocol does cover the NPE risk the old marker comment described, since cleanup is claimed only when retired && activeOps == 0.
The two Important items below are efficiency/effectiveness regressions, not correctness blockers, so they are the author's call to fix here or in a follow-up.
Tests: adequate for the behaviour being fixed. TestCacheCommitOrdering reproduces the real low-id-late-commit race against H2/MySQL/PostgreSQL, TestEntityChangeLogPoller covers the throttle floor, dedup expiry/eviction, replay failure isolation and rate limiting, the opt-in gate and the capability-check failure path, and TestEntityCachePublicationRace covers the fencing. Two gaps worth a line each: nothing asserts the query volume of a multi-page replay-only drain (the drain tests at TestEntityChangeLogPoller.java:658 and :698 only exercise incremental drains), which is the first finding below; and nothing exercises a mid-sweep replay page failure followed by resumption from the saved page cursor, although the saved-cursor code path is simple enough to read.
No findings had to be dropped for size.
Review drafted with Claude, requested by @jerryshao.
| e); | ||
| } | ||
| } | ||
| return incrementalBacklog || replayBacklog; |
There was a problem hiding this comment.
[Important] A full replay page makes this return true, so runPollCycle sets nextDelayMillis = 0 (:322) and the next cycle starts at once. But fetchNextDelivery samples the database tail whenever changes.size() < batchSize (:581-583), and during a replay-only drain the incremental query is empty, so that first disjunct is always true. The once-per-interval guard whose comment at :584-585 says "Sample at most once per interval during a drain" therefore never applies to a replay-driven drain: every zero-delay cycle issues three queries back to back (incremental, tail, replay). A sweep that needs more than one page -- window rows above pollBatchSize, 2000 by default, or any cold-start sweep over a busy window -- runs that triple at full speed with no delay between cycles, which is exactly the query amplification the guard was added to bound.
The comment at :320-322 and metrics.md both note the three queries, so this is clearly known; noting the cost is not the same as wanting it. Suggest extending the existing rate limit to cover immediately-rescheduled cycles, e.g. skip the tail sample when the previous cycle returned a backlog and nowNanos - lastTailSampleNanos < pollIntervalSecs, regardless of which of the two queries was full. The tail is observability-only (:577-578), so a 15s-stale gauge during a drain costs nothing.
Verified by: reading doPollChanges (:372-408), runPollCycle (:313-325) and fetchNextDelivery (:574-600) at this head, and confirming no test asserts tail-query count for a replay drain.
There was a problem hiding this comment.
Thanks. You are right that the interval guard does not cover a replay-only drain. I'm keeping it as-is for this PR, for two reasons:
- Cost: the extra query is
SELECT COALESCE(MAX(id), 0), which is a single primary-key index lookup. That is negligible next to the replay page query it runs beside. - Count: it is bounded by the pages in one sweep, at most
windowRows / pollBatchSize - 1extra calls per sweep. For example, at 100 writes/s the window holds about 6,000 rows, which means about 2 extraMAX(id)calls every 15 s.
Fixing it would add one more piece of "previous cycle had a backlog" state to a poller that already carries a lot of state, and this PR aims to stay a short-term fix. If the tail query ever shows up in profiles, the rate limit you suggest is the right follow-up. The extra query is already documented (metrics.md and the comment at :320-322).
| @Override | ||
| public <E extends Entity & HasIdentifier> void put(E e, boolean overwritten) | ||
| throws IOException, EntityAlreadyExistsException { | ||
| long epochBeforeWrite = cacheInvalidationEpoch.get(); |
There was a problem hiding this comment.
[Important] This samples the fence before backend.insert, so the publication of e is discarded if the epoch moved at any point during the whole write round trip. cacheInvalidationEpoch is a single process-wide counter (:100) incremented by every invalidateCache (:693) -- once per local put/update/delete, and once per change-log row delivered from another node, because the poller's target routes through this same method (newCacheChangeLogListener, :173-188). So an invalidation of a completely unrelated entity suppresses the caching of a value this node just wrote and knows to be current, and a poller batch of N rows bumps the counter N times, so any write (or point read, :273-276) overlapping a delivery burst loses its cache fill. Before this PR both put and get published unconditionally, so this is a new cost on cache fill rate, paid under exactly the write load where the cache matters most.
The same-key race the fence targets is already excluded here by the key lock taken in cacheIfEpochUnchanged plus the post-put recheck in publishIfEpochUnchanged (:680-686): a same-key delete that commits before this insert has already bumped the epoch before the sample, and one that commits after will invalidate the entry once it gets the key lock. So the pre-write sample buys nothing for put that a sample taken after backend.insert would not.
Two options worth considering: sample the epoch after the insert for put, or make the fence per key (a per-key invalidation version, or a per-segment counter) so unrelated invalidations stop suppressing publication on all three paths. I have not measured the hit-rate impact, so the magnitude is open -- but the mechanism is unconditional.
Verified by: reading put (:228-245), the epoch field (:100), invalidateCache (:690-695) and newCacheChangeLogListener (:173-188) at this head.
There was a problem hiding this comment.
Thanks for the careful analysis. The cost you describe is real. A fill is skipped on any epoch change, and this was an accepted trade-off: a skipped fill only means the next read goes to the database. The pre-insert sample is still deliberate, though. A post-insert sample reopens the ancestor race. CaffeineEntityCache.invalidate takes only the ancestor key's segment lock and scans an index snapshot; it never takes the child's lock:
| step | T1: put(table t) |
T2: drop schema s (cascade) |
|---|---|---|
| 1 | insert t holding FOR SHARE on s, commit |
waits for FOR UPDATE on s |
| 2 | cascade-deletes t, commits |
|
| 3 | epoch++, invalidates s and its descendants; t is not cached yet |
|
| 4 | post-insert sample reads the new epoch | |
| 5 | takes t's key lock, sees an unchanged epoch, publishes t |
The dropped t then stays cached until TTL. That is the stale-publication case #13828 P0-2 targets. With the sample taken before the insert, step 5 sees a changed epoch and skips. The key lock and the post-put recheck only exclude a same-key invalidation, not an ancestor invalidation that has already finished.
A per-key fence would remove the false skips. One option is striped counters checked along the ancestor chain, with clears still bumping the global epoch. That is a larger change than this P0 needs, so I'd rather measure the hit-rate impact first, as part of the #13828 performance work (item 6). Hit rate is the next step there.
| @Override | ||
| protected String currentTimeMillisSql() { | ||
| return CURRENT_TIME_MILLIS_SQL; |
There was a problem hiding this comment.
[Question] This override points the replay window filter at CAST(EXTRACT(EPOCH FROM CURRENT_TIMESTAMP) * 1000 AS BIGINT) (DatabaseTimeSQL.java:43-44), and insertEntityChange stamps created_at with the same expression (:35, :45-50). On PostgreSQL CURRENT_TIMESTAMP is the transaction start time, so a transaction that begins, writes its change-log row, and commits 51s later records created_at at its start: by the time the row becomes visible its recorded age already exceeds replayWindowSecs - sweepInterval (45s with defaults) and replay never sees it. That is precisely the long-transaction late-commit case this PR exists to fix, and PostgreSQL is the only backend with this semantics -- MySQL/H2 CURRENT_TIMESTAMP is statement time, so there the window behaves as documented.
The PR description and docs/gravitino-server-config.md both call this out, so it is a known limitation rather than an oversight. Was clock_timestamp() (statement time) considered for the PostgreSQL insert path? It would restore the full window for long transactions. If it was rejected because pruneOldEntityChanges (:51-62) and retentionSecs would then be measured from statement time instead, saying so in the doc paragraph would help operators who are sizing the window against their transaction durations.
Verified by: reading the PostgreSQL provider in full, DatabaseTimeSQL (:42-45), and the new selectRecentEntityChanges filter (EntityChangeLogBaseSQLProvider.java:72-88).
There was a problem hiding this comment.
Yes, it was considered. clock_timestamp() on the PostgreSQL insert would make created_at statement time, matching MySQL/H2. Since EntityChangeLogWriter.append runs right before commit, that would almost remove the long-transaction gap.
I'm keeping the existing writer in this PR on purpose:
- Scope: the gap only matters for PostgreSQL transactions that take longer than the window minus one sweep (45 s by default) from start to commit. Normal metadata transactions take milliseconds to seconds, so the ordering problem this PR fixes is covered at the default window.
- Mitigation: the limitation is documented (PR description and
gravitino-server-config.md), andreplayWindowSecscan be raised, up to 86400. - Risk: changing how
created_atis stamped on PostgreSQL is a separate behavior change that needs its own PostgreSQL validation.
If long transactions turn out to lose invalidations in practice, switching the PostgreSQL insert to clock_timestamp() is a small follow-up, and retention would not be affected.
…n and startup replay logging - Document that invalidateOnKeyChange may run under the entity's own cache lock and must not take another entity's lock. - Drop the duplicated non-cacheable branch in batchGet; the publication helper already handles it. - Log the first replay sweep after startup at INFO as startup history and keep the late-commit WARN for later sweeps.
… cursor after a failure
|
@jerryshao thanks for the thorough review. I replied to each inline comment. Summary:
On the two test gaps:
|
What changes were proposed in this pull request?
max(pollIntervalSecs, replayWindowSecs / 4)(15s by default), and exposereplay-records-total,replay-failures-total, anddedup-evictions-totalmetrics.invalidateOnKeyChangemay run under the entity's own cache lock.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(default60;0disables replay) andgravitino.entityChangeLog.replayMaxEntries(default100000), plus threeentity-change-logcounters:replay-records-total,replay-failures-total, anddedup-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?
TestEntityChangeLogPollertests pass, including new throttle, poll-interval floor, and eviction-counter tests; the two throttle tests fail with a zero sweep interval.TestCacheCommitOrderingpasses on H2, MySQL, and PostgreSQL.Config(generic lakehouse, fileset, Kafka, model) now stub the replay settings: 29/232/15/37 tests passed; the JMH benchmark compiles.TestRelationalEntityStoreLockWaitTimeoutmocksConfigand now stubs the replay settings (all 7 cases NPE without them). 14 affected core classes on H2: 204 tests passed.TestEntityChangeLogPoller(39) plus related store/cache classes: 115 tests passed on H2.Unit-suite command: