Copilot commented on code in PR #13831: URL: https://github.com/apache/gravitino/pull/13831#discussion_r4259165565
########## core/src/test/java/org/apache/gravitino/cache/TestEntityCachePublicationRace.java: ########## @@ -0,0 +1,264 @@ +/* + * 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.io.IOException; +import java.io.UncheckedIOException; +import java.lang.reflect.Field; +import java.time.Instant; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; +import java.util.stream.IntStream; +import org.apache.gravitino.Config; +import org.apache.gravitino.Entity; +import org.apache.gravitino.Entity.EntityType; +import org.apache.gravitino.HasIdentifier; +import org.apache.gravitino.NameIdentifier; +import org.apache.gravitino.Namespace; +import org.apache.gravitino.meta.AuditInfo; +import org.apache.gravitino.meta.TableEntity; +import org.apache.gravitino.storage.relational.RelationalBackend; +import org.apache.gravitino.storage.relational.RelationalEntityStore; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; + +/** Deterministic regression probes for cache invalidation races. */ +public class TestEntityCachePublicationRace { + @Test + void testPointReadDoesNotReacquireSpiLockForPublication() throws Exception { + NonReentrantCache cache = new NonReentrantCache(); + RelationalEntityStore store = new RelationalEntityStore(); + RelationalBackend backend = Mockito.mock(RelationalBackend.class); + setField(store, RelationalEntityStore.class, "cache", cache); + setField(store, RelationalEntityStore.class, "backend", backend); + TableEntity entity = table("non_reentrant"); + Mockito.when(backend.get(entity.nameIdentifier(), EntityType.TABLE)).thenReturn(entity); + Assertions.assertSame( + entity, store.get(entity.nameIdentifier(), EntityType.TABLE, TableEntity.class)); + Assertions.assertEquals(1, cache.acquisitions); + Mockito.verify(backend).get(entity.nameIdentifier(), EntityType.TABLE); + } + + @Test + void testParentDropMustFenceAnInflightChildPointRead() throws Exception { + CaffeineEntityCache cache = new CaffeineEntityCache(new Config() {}); + RelationalEntityStore store = new RelationalEntityStore(); + RelationalBackend backend = Mockito.mock(RelationalBackend.class); + setField(store, RelationalEntityStore.class, "cache", cache); + setField(store, RelationalEntityStore.class, "backend", backend); + NameIdentifier parent = NameIdentifier.of("lake", "catalog", "schema"); + Field lockField = CaffeineEntityCache.class.getDeclaredField("segmentedLock"); + lockField.setAccessible(true); + SegmentedLock locks = (SegmentedLock) lockField.get(cache); + // Enum identity hash codes differ across JVMs. Select keys on distinct real segments instead + // of assuming fixed names always exercise the ancestor/child race. + TableEntity old = + IntStream.range(0, 1000) + .mapToObj(i -> table("table" + i)) + .filter( + t -> + locks.getSegmentLock(EntityCacheKey.of(parent, EntityType.SCHEMA)) + != locks.getSegmentLock( + EntityCacheKey.of(t.nameIdentifier(), EntityType.TABLE))) + .findFirst() + .orElseThrow(() -> new AssertionError("No distinct child cache segment")); + Assertions.assertNotSame( + locks.getSegmentLock(EntityCacheKey.of(parent, EntityType.SCHEMA)), + locks.getSegmentLock(EntityCacheKey.of(old.nameIdentifier(), EntityType.TABLE))); + CountDownLatch readOld = new CountDownLatch(1); + CountDownLatch finishRead = new CountDownLatch(1); + Mockito.when(backend.get(old.nameIdentifier(), EntityType.TABLE)) + .thenAnswer( + invocation -> { + readOld.countDown(); + Assertions.assertTrue(finishRead.await(10, TimeUnit.SECONDS)); + return old; + }); + ExecutorService readers = Executors.newSingleThreadExecutor(); + try { + Future<TableEntity> read = + readers.submit( + () -> store.get(old.nameIdentifier(), EntityType.TABLE, TableEntity.class)); + Assertions.assertTrue(readOld.await(10, TimeUnit.SECONDS)); + Mockito.when(backend.delete(parent, EntityType.SCHEMA, true)).thenReturn(true); + Assertions.assertTrue(store.delete(parent, EntityType.SCHEMA, true)); + finishRead.countDown(); + read.get(10, TimeUnit.SECONDS); + Assertions.assertFalse( + cache.contains(old.nameIdentifier(), EntityType.TABLE), + "A child loaded before the committed parent drop must not survive its invalidation"); + } finally { + finishRead.countDown(); + readers.shutdownNow(); Review Comment: These tests create executor threads and call `shutdownNow()` but don’t `awaitTermination()`. If a task ever blocks unexpectedly (or ignores interrupts), this can leak non-daemon threads and cause suite hangs. Consider awaiting termination with a short timeout (and failing the test if termination doesn’t occur) to keep the suite deterministic. ########## docs/metrics.md: ########## @@ -78,28 +78,37 @@ The `entity-change-log` source exposes each server's change-log processing state `/prometheus/metrics`. For example, `entity-change-log.record-lag` in the metrics registry becomes `entity_change_log_record_lag` in Prometheus. Gauges read only in-memory values; the poller samples the database tail on partial and empty polls, and at most once per `pollIntervalSecs` during -a full-batch drain (including failed sample attempts). If only the tail sample fails, delivery -continues and the tail value remains at its last successful sample. +a full incremental-batch drain (including failed sample attempts). A full replay page also +schedules the next cycle immediately; when its incremental query is partial or empty, that +cycle still samples the tail. Each such replay-only drain cycle therefore performs an incremental +query, a tail query, and a replay query. If only the tail sample fails, delivery continues and +the tail value remains at its last successful sample. | Metric suffix | Type and unit | Meaning | | ------------------------------------------------------------ | ------------------------ | ----------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `db-tail-id`, `cursor-id` | gauge, change ID | Latest sampled database ID and last delivered ID on this server. | +| `db-tail-id`, `cursor-id` | gauge, change ID | Latest sampled database ID and last incrementally delivered ID on this server; replay does not move the cursor. | | `record-lag` | gauge, records | Sampled tail minus cursor, clamped at zero. Interpret only while tail sampling succeeds. | | `seconds-since-last-successful-poll` | gauge, seconds | Time since a successful database poll, including an empty result; `-1` before the first poll. | | `seconds-since-last-successful-tail-sample` | gauge, seconds | Time since `db-tail-id` was last refreshed; `-1` before the first sample. Trust `db-tail-id` and `record-lag` only while this stays near the poll interval. | | `poll-failures-total` | counter, failures | Failed poll cycles. | | `tail-sample-failures-total` | counter, failures | Failed database-tail samples; fetched batches can still be delivered. | -| `listener-failures-total`, `listener-failures.<class>-total` | counter, failures | Total failures and failures by registered listener class. | -| `records-fetched-total`, `records-delivered-total` | counter, records | Rows fetched and rows delivered successfully to listeners; one row delivered to two listeners counts twice as delivered. | +| `listener-failures-total`, `listener-failures.<class>-total` | counter, failures | Callback failures and callback failures by registered listener class. | Review Comment: The description has a duplicated phrase (“Callback failures and callback failures…”). Consider rewording to something like “Callback failures total, and by registered listener class” to avoid confusion. ########## core/src/main/java/org/apache/gravitino/storage/relational/EntityChangeLogPoller.java: ########## @@ -296,14 +356,184 @@ boolean pollChanges() { private synchronized boolean doPollChanges() { BatchDelivery delivery = fetchNextDelivery(); - if (delivery == null) { + if (replayWindowMs > 0) { + expireRememberedIds(); + } + boolean incrementalBacklog = delivery != null && delivery.changes.size() >= batchSize; + if (delivery != null) { + deliver(delivery); + remember(delivery.changes); + } + // Replay failure must not discard or delay delivery of a successfully fetched new batch. + boolean replayBacklog = false; + if (replayWindowMs > 0 + && entityPollHighWaterId > 0 + && (!replayFailed + || nanoTime() - replayFailureNanos >= TimeUnit.SECONDS.toNanos(pollIntervalSecs))) { + try { + replayBacklog = replayRecentChanges(); + replayFailed = false; + } catch (RuntimeException e) { + if (handleInterruptIfAny(e, "Entity change log replay")) { + return false; + } + replayFailed = true; + replayFailureNanos = nanoTime(); + LOG.warn( + "Entity change log replay failed cursor={} replayAfterCreatedAt={} replayAfterId={} " + + "replayUpperId={} windowMs={}; incremental delivery continues; replay retries after the poll interval", + entityPollHighWaterId, + replayAfterCreatedAt, + replayAfterId, + replayUpperId, + replayWindowMs, + e); + } + } + return incrementalBacklog || replayBacklog; + } + + private boolean replayRecentChanges() { + if (replayUpperId == 0) { + replayUpperId = entityPollHighWaterId; + } + List<EntityChangeRecord> page = + SessionUtils.getWithoutCommit( + EntityChangeLogMapper.class, + m -> + m.selectRecentEntityChanges( + replayAfterCreatedAt, replayAfterId, replayUpperId, replayWindowMs, batchSize)); + List<EntityChangeRecord> unseen = new ArrayList<>(); + for (EntityChangeRecord change : page) { + if (!rememberedIds.containsKey(change.getId())) { + unseen.add(change); + // Absence from this bounded set is evidence of a possible late commit, not proof: + // startup history and capacity eviction can also produce previously unseen IDs. + LOG.debug( + "Replaying previously unseen entity change behind cursor: changeId={} cursor={} " + + "startupCursor={} entityType={} fullName={} operation={} createdAt={} approximateAgeMs={} " + + "windowMs={}; possible late commit, startup history, or dedup eviction; " + + "age uses JVM time and is not transaction duration", + change.getId(), + entityPollHighWaterId, + startupHighWaterId, + change.getEntityType(), + change.getFullName(), + change.getOperateType(), + change.getCreatedAt(), + System.currentTimeMillis() - change.getCreatedAt(), + replayWindowMs); + } + } + if (!unseen.isEmpty()) { + unseen.sort(Comparator.comparingLong(EntityChangeRecord::getId)); + LOG.warn( + "Entity change log replay found {} previously unseen record(s) behind cursor={} " + + "startupCursor={} windowMs={} approximateOldestAgeMs={} records={}; " + + "possible late commit, startup history, or dedup eviction; record details at DEBUG; " + + "age uses JVM time, not transaction duration", + unseen.size(), + entityPollHighWaterId, + startupHighWaterId, + replayWindowMs, + System.currentTimeMillis() + - unseen.stream().mapToLong(EntityChangeRecord::getCreatedAt).min().orElse(0), + summarize(unseen)); + List<EntityChangeRecord> immutable = List.copyOf(unseen); + notifyListeners( + new BatchDelivery( + immutable, immutable.get(immutable.size() - 1).getId(), replayListeners())); + remember(immutable); + } + if (page.size() < batchSize) { + LOG.debug( + "Entity change log replay sweep complete cursor={} upperId={} fetched={} replayed={} " + + "deduplicated={} windowMs={}", + entityPollHighWaterId, + replayUpperId, + page.size(), + unseen.size(), + page.size() - unseen.size(), + replayWindowMs); + replayAfterCreatedAt = -1; + replayAfterId = 0; + replayUpperId = 0; return false; } - deliver(delivery); - // A full batch is the backlog signal. The sampled tail id would be more precise, but that - // sample is best-effort and may be missing; a full batch with nothing behind it only costs one - // extra empty poll before the normal interval resumes. - return delivery.changes.size() >= batchSize; + EntityChangeRecord last = page.get(page.size() - 1); + replayAfterCreatedAt = last.getCreatedAt(); + replayAfterId = last.getId(); + LOG.debug( + "Entity change log replay page cursor={} upperId={} afterCreatedAt={} afterId={} " + + "fetched={} replayed={} deduplicated={}", + entityPollHighWaterId, + replayUpperId, + replayAfterCreatedAt, + replayAfterId, + page.size(), + unseen.size(), + page.size() - unseen.size()); + return true; + } + + private List<EntityChangeLogListener> replayListeners() { + List<EntityChangeLogListener> eligible = new ArrayList<>(); + for (EntityChangeLogListener listener : listeners) { + try { + if (listener.supportsReplay()) { + eligible.add(listener); + } + } catch (Throwable e) { + LOG.warn( + "Entity change log replay capability check failed listener={}; skipping its replay", + listener.getClass().getName(), + e); + } + } + return List.copyOf(eligible); + } + + private void expireRememberedIds() { + long now = nanoTime(); + Iterator<Map.Entry<Long, Long>> entries = rememberedIds.entrySet().iterator(); + while (entries.hasNext()) { + if (now - entries.next().getValue() < TimeUnit.MILLISECONDS.toNanos(replayWindowMs)) { + break; + } + entries.remove(); + } + } + + private void remember(List<EntityChangeRecord> changes) { + if (replayWindowMs == 0) { + return; + } + long now = nanoTime(); + int evicted = 0; + for (EntityChangeRecord change : changes) { + rememberedIds.putIfAbsent(change.getId(), now); + if (rememberedIds.size() > replayMaxEntries) { + Iterator<Long> ids = rememberedIds.keySet().iterator(); + ids.next(); + ids.remove(); + evicted++; + } + } Review Comment: `rememberedIds.putIfAbsent(change.getId(), now)` never refreshes the timestamp for IDs that are seen again, so a frequently-delivered ID can still “expire” based on its first-seen time and then be replayed again sooner than intended. To make the dedup window behave as “last seen” (and reduce unnecessary duplicate invalidations), update the stored timestamp on every delivery and ensure eviction order matches that policy (e.g., use access-order `LinkedHashMap` or remove+put to move the entry to the end).” ########## core/src/test/java/org/apache/gravitino/storage/relational/TestEntityChangeLogPoller.java: ########## @@ -64,13 +74,354 @@ public class TestEntityChangeLogPoller { private static final int MAX_ROWS = 2000; + @Test + void testReplayRecoversLateLowerIdWithoutMovingCursorBackOrRedelivering() { + EntityChangeLogMapper mapper = mock(EntityChangeLogMapper.class); + EntityChangeRecord late = + new EntityChangeRecord(1, "ml1", "TABLE", "ml1.cat.schema.late", OperateType.DROP, 100); + EntityChangeRecord fast = + new EntityChangeRecord(2, "ml1", "TABLE", "ml1.cat.schema.fast", OperateType.ALTER, 200); + when(mapper.selectEntityChanges(0, 2)).thenReturn(List.of(fast)); + when(mapper.selectEntityChanges(2, 2)).thenReturn(List.of()); + when(mapper.selectRecentEntityChanges(-1, 0, 2, 60000, 2)) + .thenReturn(List.of(fast), List.of(late, fast), List.of(late, fast)); + when(mapper.selectRecentEntityChanges(200, 2, 2, 60000, 2)).thenReturn(List.of()); + List<EntityChangeRecord> received = new ArrayList<>(); + EntityChangeLogMetricsSource metrics = new EntityChangeLogMetricsSource(); + try (MockedStatic<SessionUtils> sessions = mockStatic(SessionUtils.class)) { + mockSessionUtils(sessions, mapper); + EntityChangeLogPoller poller = new EntityChangeLogPoller(1, 2, metrics); + poller.registerListener(replayListener(received)); + Assertions.assertFalse(poller.pollChanges()); + Assertions.assertTrue(poller.pollChanges()); + Assertions.assertFalse(poller.pollChanges()); + Assertions.assertTrue(poller.pollChanges()); + Assertions.assertEquals(List.of(2L, 1L), ids(received)); + Assertions.assertEquals( + 2L, metrics.getMetricRegistry().getGauges().get("cursor-id").getValue()); + } + } + + @Test + void testReplayPagesEqualTimestampsAndRestartsForLaterVisibleRows() { + EntityChangeLogMapper mapper = mock(EntityChangeLogMapper.class); + when(mapper.selectEntityChanges(0, 2)).thenReturn(changes(4, 4)); + when(mapper.selectEntityChanges(4, 2)).thenReturn(List.of()); + when(mapper.selectRecentEntityChanges(-1, 0, 4, 60000, 2)) + .thenReturn(changes(2, 3), changes(1, 2)); + when(mapper.selectRecentEntityChanges(0, 3, 4, 60000, 2)).thenReturn(changes(4, 4)); + List<EntityChangeRecord> received = new ArrayList<>(); + try (MockedStatic<SessionUtils> sessions = mockStatic(SessionUtils.class)) { + mockSessionUtils(sessions, mapper); + EntityChangeLogPoller poller = + new EntityChangeLogPoller(1, 2, new EntityChangeLogMetricsSource()); + poller.registerListener(replayListener(received)); + Assertions.assertTrue(poller.pollChanges()); + Assertions.assertFalse(poller.pollChanges()); + Assertions.assertTrue(poller.pollChanges()); + Assertions.assertEquals(List.of(4L, 2L, 3L, 1L), ids(received)); + verify(mapper).selectRecentEntityChanges(0, 3, 4, 60000, 2); + } + } + + @Test + void testReplayFailureDoesNotBlockIncrementalDeliveryOrRetryNewRows() { + EntityChangeLogMapper mapper = mock(EntityChangeLogMapper.class); + when(mapper.selectEntityChanges(0, 2)).thenReturn(changes(1, 1)); + when(mapper.selectEntityChanges(1, 2)).thenReturn(changes(2, 2)); + when(mapper.selectRecentEntityChanges(-1, 0, 1, 60000, 2)) + .thenThrow(new IllegalStateException("replay unavailable")) + .thenReturn(changes(1, 1)); + List<EntityChangeRecord> received = new ArrayList<>(); + try (MockedStatic<SessionUtils> sessions = mockStatic(SessionUtils.class)) { + mockSessionUtils(sessions, mapper); + AtomicLong clock = new AtomicLong(); + EntityChangeLogPoller poller = + spy(new EntityChangeLogPoller(1, 2, new EntityChangeLogMetricsSource())); + doAnswer(ignored -> clock.get()).when(poller).nanoTime(); + poller.registerListener(replayListener(received)); + Assertions.assertFalse(poller.pollChanges()); + clock.set(TimeUnit.SECONDS.toNanos(1)); + Assertions.assertFalse(poller.pollChanges()); + Assertions.assertEquals(List.of(1L, 2L), ids(received)); + verify(mapper, times(2)).selectRecentEntityChanges(-1, 0, 1, 60000, 2); + } + } + + @Test + void testReplayFailureIsRateLimitedWhileIncrementalBacklogDrains() { + EntityChangeLogMapper mapper = mock(EntityChangeLogMapper.class); + when(mapper.selectEntityChanges(0, 2)).thenReturn(changes(1, 2)); + when(mapper.selectEntityChanges(2, 2)).thenReturn(changes(3, 4)); + when(mapper.selectEntityChanges(4, 2)).thenReturn(List.of()); + when(mapper.selectRecentEntityChanges(-1, 0, 2, 60000, 2)) + .thenThrow(new IllegalStateException("replay unavailable")); + AtomicLong clock = new AtomicLong(); + List<EntityChangeRecord> received = new ArrayList<>(); + try (MockedStatic<SessionUtils> sessions = mockStatic(SessionUtils.class)) { + mockSessionUtils(sessions, mapper); + EntityChangeLogPoller poller = + spy(new EntityChangeLogPoller(1, 2, new EntityChangeLogMetricsSource())); + doAnswer(ignored -> clock.get()).when(poller).nanoTime(); + poller.registerListener(replayListener(received)); + Assertions.assertTrue(poller.pollChanges()); + Assertions.assertTrue(poller.pollChanges()); + verify(mapper).selectRecentEntityChanges(-1, 0, 2, 60000, 2); + Assertions.assertEquals(List.of(1L, 2L, 3L, 4L), ids(received)); + clock.set(TimeUnit.SECONDS.toNanos(1)); + Assertions.assertFalse(poller.pollChanges()); + verify(mapper, times(2)).selectRecentEntityChanges(-1, 0, 2, 60000, 2); + } + } + + @Test + void testCapacityWarningsAreRateLimitedAndCountSuppressedEvictions() { + EntityChangeLogMapper mapper = mock(EntityChangeLogMapper.class); + when(mapper.selectEntityChanges(0, 2)).thenReturn(changes(1, 2)); + when(mapper.selectEntityChanges(2, 2)).thenReturn(changes(3, 4)); + when(mapper.selectEntityChanges(4, 2)).thenReturn(changes(5, 6)); + AtomicLong clock = new AtomicLong(); + List<LogEvent> logs = + captureReplayLogs( + () -> { + try (MockedStatic<SessionUtils> sessions = mockStatic(SessionUtils.class)) { + mockSessionUtils(sessions, mapper); + EntityChangeLogPoller poller = + spy(new EntityChangeLogPoller(1, 2, 60, 1, new EntityChangeLogMetricsSource())); + doAnswer(ignored -> clock.get()).when(poller).nanoTime(); + poller.pollChanges(); + poller.pollChanges(); + clock.set(TimeUnit.SECONDS.toNanos(1)); + poller.pollChanges(); + } + }); + List<String> warnings = + logs.stream() + .map(e -> e.getMessage().getFormattedMessage()) + .filter(m -> m.contains("dedup capacity reached")) + .toList(); + Assertions.assertEquals(2, warnings.size()); + Assertions.assertTrue(warnings.get(0).contains("evictedIds=1")); + Assertions.assertTrue(warnings.get(1).contains("evictedIds=4")); + } + + @Test + void testDisabledReplayDoesNotQueryRecentHistory() { + EntityChangeLogMapper mapper = mock(EntityChangeLogMapper.class); + when(mapper.selectEntityChanges(0, 2)).thenReturn(changes(1, 1)); + try (MockedStatic<SessionUtils> sessions = mockStatic(SessionUtils.class)) { + mockSessionUtils(sessions, mapper); + EntityChangeLogPoller poller = + new EntityChangeLogPoller(1, 2, 0, 1, new EntityChangeLogMetricsSource()); + Assertions.assertFalse(poller.pollChanges()); + verify(mapper, never()) + .selectRecentEntityChanges(anyLong(), anyLong(), anyLong(), anyLong(), anyInt()); + } + } + + @Test + void testCapacityEvictionAllowsRedeliveryWithoutSkippingUnknownEvents() { + EntityChangeLogMapper mapper = mock(EntityChangeLogMapper.class); + when(mapper.selectEntityChanges(0, 2)).thenReturn(changes(1, 2)); + when(mapper.selectRecentEntityChanges(-1, 0, 2, 60000, 2)).thenReturn(changes(1, 1)); + List<EntityChangeRecord> received = new ArrayList<>(); + try (MockedStatic<SessionUtils> sessions = mockStatic(SessionUtils.class)) { + mockSessionUtils(sessions, mapper); + EntityChangeLogPoller poller = + new EntityChangeLogPoller(1, 2, 60, 1, new EntityChangeLogMetricsSource()); + poller.registerListener(replayListener(received)); + Assertions.assertTrue(poller.pollChanges()); + Assertions.assertEquals(List.of(1L, 2L, 1L), ids(received)); + } + } + + @Test + void testDedupExpiresUsingMonotonicClock() { + EntityChangeLogMapper mapper = mock(EntityChangeLogMapper.class); + when(mapper.selectEntityChanges(0, 2)).thenReturn(changes(1, 1)); + when(mapper.selectRecentEntityChanges(-1, 0, 1, 1000, 2)).thenReturn(changes(1, 1)); + AtomicLong clock = new AtomicLong(); + List<EntityChangeRecord> received = new ArrayList<>(); + try (MockedStatic<SessionUtils> sessions = mockStatic(SessionUtils.class)) { + mockSessionUtils(sessions, mapper); + EntityChangeLogPoller poller = + spy(new EntityChangeLogPoller(1, 2, 1, 2, new EntityChangeLogMetricsSource())); + doAnswer(ignored -> clock.get()).when(poller).nanoTime(); + poller.registerListener(replayListener(received)); + poller.pollChanges(); + clock.set(TimeUnit.SECONDS.toNanos(2)); + poller.pollChanges(); + Assertions.assertEquals(List.of(1L, 1L), ids(received)); + } + } + + @Test + void testReplayDiagnosticsExposeEventAndFailureContext() { + EntityChangeLogMapper mapper = mock(EntityChangeLogMapper.class); + when(mapper.selectEntityChanges(0, 2)).thenReturn(changes(2, 2)); + when(mapper.selectRecentEntityChanges(-1, 0, 2, 60000, 2)) + .thenReturn(changes(1, 1)) + .thenThrow(new IllegalStateException("replay database unavailable")); + List<LogEvent> logs = + captureReplayLogs( + () -> { + try (MockedStatic<SessionUtils> sessions = mockStatic(SessionUtils.class)) { + mockSessionUtils(sessions, mapper); + EntityChangeLogPoller poller = + new EntityChangeLogPoller(1, 2, new EntityChangeLogMetricsSource()); + poller.pollChanges(); + poller.pollChanges(); + } + }); + String diagnostics = + logs.stream() + .map(e -> e.getMessage().getFormattedMessage()) + .collect(Collectors.joining("\n")); + Assertions.assertTrue(diagnostics.contains("cursor=2"), diagnostics); + Assertions.assertTrue(diagnostics.contains("changeId=1"), diagnostics); + Assertions.assertTrue(diagnostics.contains("entityType=TABLE"), diagnostics); + Assertions.assertTrue(diagnostics.contains("fullName=ml1.cat1.schema1.table1"), diagnostics); + Assertions.assertTrue(diagnostics.contains("operation=ALTER"), diagnostics); + Assertions.assertTrue(diagnostics.contains("createdAt=0"), diagnostics); + Assertions.assertTrue(diagnostics.contains("approximateAgeMs="), diagnostics); + Assertions.assertTrue(diagnostics.contains("windowMs=60000"), diagnostics); + Assertions.assertTrue( + diagnostics.contains("possible late commit, startup history, or dedup eviction"), + diagnostics); + Assertions.assertTrue(diagnostics.contains("replay failed cursor=2"), diagnostics); + Assertions.assertTrue( + logs.stream().anyMatch(e -> e.getThrown() instanceof IllegalStateException)); + } + + @Test + void testReplayStartupIncludesLowerIdsAndCapacityWarning() { + EntityChangeLogMapper mapper = mock(EntityChangeLogMapper.class); + when(mapper.selectMaxChangeId()).thenReturn(3L); + when(mapper.selectRecentEntityChanges(-1, 0, 3, 60000, 2)).thenReturn(changes(1, 2)); + ScheduledExecutorService scheduler = mock(ScheduledExecutorService.class); + List<EntityChangeRecord> received = new ArrayList<>(); + List<LogEvent> logs = + captureReplayLogs( + () -> { + try (MockedStatic<SessionUtils> sessions = mockStatic(SessionUtils.class)) { + mockSessionUtils(sessions, mapper); + EntityChangeLogPoller poller = + spy(new EntityChangeLogPoller(1, 2, 60, 1, new EntityChangeLogMetricsSource())); + doReturn(scheduler).when(poller).createScheduler(); + poller.registerListener(replayListener(received)); + try { + poller.start(); + Assertions.assertTrue(poller.pollChanges()); + Assertions.assertEquals(List.of(1L, 2L), ids(received)); + } finally { + poller.close(); + } + } + }); + String diagnostics = + logs.stream() + .map(e -> e.getMessage().getFormattedMessage()) + .collect(Collectors.joining("\n")); + Assertions.assertTrue(diagnostics.contains("startupCursor=3"), diagnostics); + Assertions.assertTrue( + diagnostics.contains("dedup capacity reached cursor=3 evictedIds=1 maxEntries=1"), + diagnostics); + } + + @Test + void testReplayOptInPreservesExistingListenersAndIsolatesCapabilityFailure() { + EntityChangeLogMapper mapper = mock(EntityChangeLogMapper.class); + List<EntityChangeRecord> incremental = changes(2, 2); + when(mapper.selectEntityChanges(0, 2)).thenReturn(incremental); + when(mapper.selectRecentEntityChanges(-1, 0, 2, 60000, 2)).thenReturn(changes(1, 1)); + List<EntityChangeRecord> legacyReceived = new ArrayList<>(); + List<EntityChangeRecord> replayReceived = new ArrayList<>(); + EntityChangeLogListener broken = mock(EntityChangeLogListener.class); + when(broken.supportsReplay()).thenThrow(new AssertionError("capability unavailable")); + try (MockedStatic<SessionUtils> sessions = mockStatic(SessionUtils.class)) { + mockSessionUtils(sessions, mapper); + EntityChangeLogPoller poller = + new EntityChangeLogPoller(1, 2, new EntityChangeLogMetricsSource()); + poller.registerListener(legacyReceived::addAll); + poller.registerListener(broken); + poller.registerListener(replayListener(replayReceived)); + Assertions.assertFalse(poller.pollChanges()); + Assertions.assertEquals(List.of(2L), ids(legacyReceived)); + Assertions.assertEquals(List.of(2L, 1L), ids(replayReceived)); + verify(broken).onEntityChange(incremental); + } + } + + private static EntityChangeLogListener replayListener(List<EntityChangeRecord> received) { + return new EntityChangeLogListener() { + /** {@inheritDoc} */ + @Override + public void onEntityChange(List<EntityChangeRecord> changes) { + received.addAll(changes); + } + + /** {@inheritDoc} */ + @Override + public boolean supportsReplay() { + return true; + } + }; + } + + private static List<LogEvent> captureReplayLogs(Runnable action) { + LoggerContext context = + (LoggerContext) LogManager.getContext(EntityChangeLogPoller.class.getClassLoader(), false); + AbstractConfiguration configuration = (AbstractConfiguration) context.getConfiguration(); + ReplayLogCollector collector = new ReplayLogCollector(); + collector.start(); + configuration.addAppender(collector); + String name = EntityChangeLogPoller.class.getName(); + LoggerConfig previous = configuration.getLoggers().get(name); + LoggerConfig logger = new LoggerConfig(name, Level.DEBUG, false); + logger.addAppender(collector, Level.DEBUG, null); + configuration.addLogger(name, logger); + context.updateLoggers(); + try { + action.run(); + } finally { + configuration.removeLogger(name); + if (previous != null) { + configuration.addLogger(name, previous); + } + collector.stop(); + configuration.removeAppender(collector.getName()); + context.updateLoggers(); + } Review Comment: This test mutates the global Log4j2 configuration (adds/removes a logger) which can lead to flaky behavior if tests ever run in parallel or if other tests rely on logger configuration for the same class. Prefer attaching a dedicated test appender to the existing logger (e.g., Log4j2 `ListAppender` pattern) or using a scoped mechanism that avoids replacing the `LoggerConfig` entry entirely. -- 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]
