yuqi1129 commented on code in PR #13831:
URL: https://github.com/apache/gravitino/pull/13831#discussion_r4291682643
##########
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:
Kept as-is on purpose. Expiry removes a prefix of the `LinkedHashMap` in
insertion order, which only stays correct if timestamps are monotonic in that
order; refreshing on re-sight would break that (or force a full scan /
access-order map). It is also unnecessary: the replay query drops a row once
its DB-recorded age exceeds the window, and first delivery always happens after
`created_at`, so the row leaves the query no later than its dedup entry
expires. The rationale is in the code comment added in 80a42d67fd.
##########
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:
Fixed in df4b5f107c: both executors now `shutdownNow()` and assert
`awaitTermination(10, SECONDS)`.
##########
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:
Fixed in df4b5f107c: the test no longer adds/removes a logger config; it
attaches a dedicated appender to the existing `EntityChangeLogPoller` logger
and restores the previous level in `finally`.
--
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]
