This is an automated email from the ASF dual-hosted git repository. yuqi1129 pushed a commit to branch me/fix-entity-change-log-poller-11736 in repository https://gitbox.apache.org/repos/asf/gravitino.git
commit 8abb18ffe723be9f2f855ba6b36e30964d3ecce2 Author: yuqi <[email protected]> AuthorDate: Fri Jun 26 22:02:19 2026 +0800 [#11739] improvement(core): keep id-cursor poller, add listener-failure pause and missed-row logging Revert the created_at watermark + overlap-window cursor (#11736/#11739). Its overlap window re-delivered the same change row across poll cycles, which defeated CatalogManager's single-shot consumeLocalMutation dedup and made the poller invalidate the node's own in-use catalog. That invalidation synchronously closed the catalog's IsolatedClassLoader (and connection pool) while a request was still using it, producing a permanently cached NoClassDefFoundError that failed the whole Ranger authz IT suite. The overlap cure was worse than the rare commit-ordering miss it targeted, which for the catalog cache is also bounded by TTL eviction. Keep the listener-failure handling: the cursor advances only after every listener applies the batch; on failure it pauses and re-dispatches until all succeed (ERROR-logged), so a transient listener failure cannot silently drop invalidations. Instead of preventing the commit-ordering gap, make it observable: ids are assigned at INSERT but visible at COMMIT, so a row can commit below the id>cursor cursor after it advanced past it and be missed. The poller now records skipped ids (bounded: narrow gaps only, capped count, stale entries dropped) and WARN-logs any that later become visible, so the real-world frequency of the race is measurable. Tests: listener-failure pause/retry, cursor-advance-on-recovery, and late-commit missed-row detection. --- .../storage/relational/EntityChangeLogPoller.java | 159 ++++++++++++++++++--- .../relational/TestEntityChangeLogPoller.java | 84 ++++++++++- 2 files changed, 223 insertions(+), 20 deletions(-) diff --git a/core/src/main/java/org/apache/gravitino/storage/relational/EntityChangeLogPoller.java b/core/src/main/java/org/apache/gravitino/storage/relational/EntityChangeLogPoller.java index 5719cea9f6..50192c593d 100644 --- a/core/src/main/java/org/apache/gravitino/storage/relational/EntityChangeLogPoller.java +++ b/core/src/main/java/org/apache/gravitino/storage/relational/EntityChangeLogPoller.java @@ -21,7 +21,10 @@ package org.apache.gravitino.storage.relational; import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Preconditions; import java.util.Collections; +import java.util.HashSet; import java.util.List; +import java.util.Set; +import java.util.TreeSet; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; @@ -38,8 +41,11 @@ import org.slf4j.LoggerFactory; * * <p>The poller owns the single high-water mark for a Gravitino server process and dispatches each * consumed batch to registered listeners. Listeners should only perform idempotent local cache - * invalidation. The cursor always advances after dispatch regardless of individual listener - * failures, so a faulty listener cannot block other listeners or prevent pruning. + * invalidation. The cursor is advanced only after every listener applies the batch; if any listener + * fails, forward progress is paused and the same batch is re-dispatched on subsequent cycles until + * all listeners succeed, so a transient listener failure cannot silently drop a batch's + * invalidations. Because the process owns one shared cursor, a persistently failing listener blocks + * progress for all listeners until the stuck rows age past the retention window (logged at ERROR). */ public class EntityChangeLogPoller implements AutoCloseable { @@ -48,12 +54,40 @@ public class EntityChangeLogPoller implements AutoCloseable { /** Max entity-change rows to fetch per poller cycle. */ private static final int ENTITY_CHANGE_POLLER_MAX_ROWS = 500; + /** + * Upper bound on the number of candidate "missed id" gaps tracked at once, so the detection state + * can never grow without bound regardless of write/rollback patterns. + */ + private static final int MAX_TRACKED_GAP_IDS = 10_000; + + /** + * Gaps wider than this are not tracked as missed-row candidates. A real commit-ordering gap (a + * few concurrent in-flight transactions whose ids are interleaved with their commit order) is + * narrow; a wide gap is almost always rolled-back/abandoned auto-increment ids that will never + * commit, and tracking them would only add noise. + */ + private static final long MAX_GAP_WIDTH = 256; + + /** + * Candidate gap ids further than this below the cursor are dropped as stale (never committed). + */ + private static final long GAP_STALE_LOOKBACK = 1_000_000; + private final List<EntityChangeLogListener> listeners = new CopyOnWriteArrayList<>(); private final long pollIntervalSecs; private final long retentionMs; private final long cleanupIntervalMs; private final LongSupplier clockMs; + /** + * Auto-increment ids below the cursor that were absent when the cursor advanced past them. If + * such an id later becomes visible it was committed after the {@code id > cursor} query had + * already skipped it — i.e. a permanently missed change-log row (see {@link + * #detectFilledGaps()}). This is observability-only state; it never affects dispatch. Guarded by + * {@code doPollChanges}' monitor. + */ + private final TreeSet<Long> pendingGapIds = new TreeSet<>(); + private ScheduledExecutorService scheduler; private volatile long entityPollHighWaterId = 0; private volatile long lastCleanupMs = Long.MIN_VALUE; @@ -170,31 +204,115 @@ public class EntityChangeLogPoller implements AutoCloseable { private synchronized void doPollChanges() { List<EntityChangeRecord> changes = fetchEntityChanges(); - if (changes.isEmpty()) { - pruneExpiredChangesIfNeeded(); - return; - } - long maxSeenId = entityPollHighWaterId; - for (EntityChangeRecord change : changes) { - if (change.getId() > maxSeenId) { - maxSeenId = change.getId(); + if (!changes.isEmpty()) { + long previousCursor = entityPollHighWaterId; + Set<Long> receivedIds = new HashSet<>(); + long maxSeenId = entityPollHighWaterId; + for (EntityChangeRecord change : changes) { + receivedIds.add(change.getId()); + if (change.getId() > maxSeenId) { + maxSeenId = change.getId(); + } } - } - List<EntityChangeRecord> dispatchedChanges = Collections.unmodifiableList(changes); - for (EntityChangeLogListener listener : listeners) { - try { - listener.onEntityChange(dispatchedChanges); - } catch (Exception e) { - LOG.warn("Entity change listener {} failed", listener.getClass().getName(), e); + List<EntityChangeRecord> dispatchedChanges = Collections.unmodifiableList(changes); + boolean allListenersSucceeded = true; + for (EntityChangeLogListener listener : listeners) { + try { + listener.onEntityChange(dispatchedChanges); + } catch (Exception e) { + allListenersSucceeded = false; + LOG.warn("Entity change listener {} failed", listener.getClass().getName(), e); + } + } + + // Only advance the cursor when every listener applied the batch. A listener failure must not + // drop the batch's invalidations: keeping the cursor in place re-dispatches the same batch on + // the next cycle until all listeners succeed. Listeners are idempotent, so re-dispatching to + // an already-applied listener is harmless. + if (allListenersSucceeded) { + entityPollHighWaterId = maxSeenId; + recordNewGaps(previousCursor, maxSeenId, receivedIds); + } else { + // Forward progress is paused until every listener applies the batch; the same batch is + // re-dispatched every cycle. If this persists, the stuck rows will eventually be pruned by + // retention cleanup and their invalidations lost permanently, leaving caches to serve stale + // data. Surface at ERROR so operators can act. + LOG.error( + "Entity change cursor is paused at id {} because at least one listener failed to apply " + + "the current batch; invalidations will be lost if this is not resolved before the " + + "stuck rows age past the retention window", + entityPollHighWaterId); } } - entityPollHighWaterId = maxSeenId; + // A missed row's id is below the cursor, so the id>cursor fetch above never returns it; the + // fill check must therefore run on every cycle, including cycles where the fetch was empty. + detectFilledGaps(); pruneExpiredChangesIfNeeded(); } + /** + * Records ids in {@code (previousCursor, maxSeenId]} that were absent from this batch as + * candidate missed rows. Narrow gaps only (see {@link #MAX_GAP_WIDTH}) and bounded in total. + * Observability only; does not affect dispatch or the cursor. + */ + private void recordNewGaps(long previousCursor, long maxSeenId, Set<Long> receivedIds) { + for (long id = previousCursor + 1; id <= maxSeenId; id++) { + if (id > previousCursor + MAX_GAP_WIDTH && pendingGapIds.isEmpty()) { + // The batch's id span is far wider than a plausible concurrent-commit gap and we have no + // gaps to confirm; treat the bulk as rolled-back ids rather than scanning the whole span. + break; + } + if (!receivedIds.contains(id)) { + pendingGapIds.add(id); + } + } + while (pendingGapIds.size() > MAX_TRACKED_GAP_IDS) { + pendingGapIds.pollFirst(); + } + } + + /** + * Logs any previously recorded gap id that has since become visible in the DB: such a row was + * committed after the {@code id > cursor} cursor had already advanced past it, so it was + * permanently skipped — a missed cache invalidation. Purely diagnostic; it re-reads from below + * the lowest pending gap and never dispatches the rows. Stale gaps (far below the cursor, never + * committed) are pruned so the lookback stays cheap. + */ + private void detectFilledGaps() { + if (pendingGapIds.isEmpty()) { + return; + } + + long staleFloor = entityPollHighWaterId - GAP_STALE_LOOKBACK; + pendingGapIds.headSet(staleFloor).clear(); + if (pendingGapIds.isEmpty()) { + return; + } + + long lookbackFrom = pendingGapIds.first() - 1; + List<EntityChangeRecord> filled = + SessionUtils.getWithoutCommit( + EntityChangeLogMapper.class, + m -> m.selectEntityChanges(lookbackFrom, ENTITY_CHANGE_POLLER_MAX_ROWS)); + for (EntityChangeRecord record : filled) { + if (pendingGapIds.remove(record.getId())) { + LOG.warn( + "entity_change_log MISSED a change row (commit-ordering gap): id={} fullName={} " + + "entityType={} operateType={} became visible below the poll cursor (current " + + "cursor id={}) and was permanently skipped by the id>cursor query, so its cache " + + "invalidation was dropped on this node", + record.getId(), + record.getFullName(), + record.getEntityType(), + record.getOperateType(), + entityPollHighWaterId); + } + } + } + private List<EntityChangeRecord> fetchEntityChanges() { return SessionUtils.getWithoutCommit( EntityChangeLogMapper.class, @@ -245,4 +363,9 @@ public class EntityChangeLogPoller implements AutoCloseable { private static long getOrDefault(Long value) { return value == null ? 0L : value; } + + @VisibleForTesting + Set<Long> pendingGapIds() { + return pendingGapIds; + } } diff --git a/core/src/test/java/org/apache/gravitino/storage/relational/TestEntityChangeLogPoller.java b/core/src/test/java/org/apache/gravitino/storage/relational/TestEntityChangeLogPoller.java index 4eea24e24f..68ad7c13d6 100644 --- a/core/src/test/java/org/apache/gravitino/storage/relational/TestEntityChangeLogPoller.java +++ b/core/src/test/java/org/apache/gravitino/storage/relational/TestEntityChangeLogPoller.java @@ -29,6 +29,7 @@ import static org.mockito.Mockito.when; import java.util.ArrayList; import java.util.List; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Consumer; import java.util.function.Function; import org.apache.gravitino.storage.relational.mapper.EntityChangeLogMapper; @@ -80,11 +81,12 @@ public class TestEntityChangeLogPoller { } @Test - void testListenerFailureDoesNotBlockOtherListenersAndCursorStillAdvances() { + void testListenerFailureDoesNotBlockOthersAndPausesCursorForRetry() { EntityChangeLogMapper mapper = mock(EntityChangeLogMapper.class); EntityChangeRecord change = change(1L, "CATALOG", "ml1.cat1"); + // The cursor must NOT advance past a batch that any listener failed to apply, so the same batch + // is re-fetched from id 0 on every cycle until the failing listener recovers. when(mapper.selectEntityChanges(0L, 500)).thenReturn(List.of(change)); - when(mapper.selectEntityChanges(1L, 500)).thenReturn(List.of()); List<EntityChangeRecord> received = new ArrayList<>(); @@ -108,7 +110,85 @@ public class TestEntityChangeLogPoller { poller.pollChanges(); } + // Healthy listener is never blocked by the failing one; because the cursor stays put, the batch + // is re-dispatched on both cycles. + Assertions.assertEquals(List.of(change, change), received); + } + + @Test + void testCursorAdvancesOnceFailingListenerRecovers() { + EntityChangeLogMapper mapper = mock(EntityChangeLogMapper.class); + EntityChangeRecord change = change(1L, "CATALOG", "ml1.cat1"); + when(mapper.selectEntityChanges(0L, 500)).thenReturn(List.of(change)); + when(mapper.selectEntityChanges(1L, 500)).thenReturn(List.of()); + + List<EntityChangeRecord> received = new ArrayList<>(); + AtomicBoolean firstCall = new AtomicBoolean(true); + + try (MockedStatic<SessionUtils> sessionUtils = mockStatic(SessionUtils.class)) { + sessionUtils + .when(() -> SessionUtils.getWithoutCommit(any(), any())) + .thenAnswer( + invocation -> { + Function<Object, Object> func = invocation.getArgument(1); + return func.apply(mapper); + }); + + EntityChangeLogPoller poller = new EntityChangeLogPoller(1); + poller.registerListener( + changes -> { + if (firstCall.getAndSet(false)) { + throw new RuntimeException("transient listener failure"); + } + received.addAll(changes); + }); + + // Cycle 1 fails -> cursor stays at 0. Cycle 2 succeeds -> cursor advances to 1. Cycle 3 then + // fetches from the advanced cursor and finds nothing. + poller.pollChanges(); + poller.pollChanges(); + poller.pollChanges(); + } + Assertions.assertEquals(List.of(change), received); + // Proves the cursor advanced to 1 after recovery (cycle 3 queried from id 1). + verify(mapper).selectEntityChanges(1L, 500); + } + + @Test + void testDetectsLateCommittedRowAsMissedGap() { + EntityChangeLogMapper mapper = mock(EntityChangeLogMapper.class); + EntityChangeRecord seen = change(2L, "CATALOG", "ml1.cat2"); + EntityChangeRecord late = change(1L, "CATALOG", "ml1.cat1"); + + // Cycle 1: only id=2 is visible; id=1 was assigned but not yet committed, leaving a gap at 1. + when(mapper.selectEntityChanges(0L, 500)).thenReturn(List.of(seen)); + when(mapper.selectEntityChanges(2L, 500)).thenReturn(List.of()); + + try (MockedStatic<SessionUtils> sessionUtils = mockStatic(SessionUtils.class)) { + sessionUtils + .when(() -> SessionUtils.getWithoutCommit(any(), any())) + .thenAnswer( + invocation -> { + Function<Object, Object> func = invocation.getArgument(1); + return func.apply(mapper); + }); + + EntityChangeLogPoller poller = new EntityChangeLogPoller(1); + poller.registerListener(records -> {}); + + poller.pollChanges(); + // The skipped id is tracked as a candidate missed row. + Assertions.assertTrue(poller.pendingGapIds().contains(1L)); + + // id=1 now commits late, below the already-advanced cursor (id=2). A non-overlapping + // id>cursor cursor can never re-read it, so the poller surfaces it as a missed row and clears + // the candidate. + when(mapper.selectEntityChanges(0L, 500)).thenReturn(List.of(late, seen)); + poller.pollChanges(); + + Assertions.assertFalse(poller.pendingGapIds().contains(1L)); + } } @Test
