From 4b93570a4a139532f57644b01593213df9aadb1b Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 14 Aug 2026 10:54:08 +0800 Subject: [PATCH] [Subscription] Fix WAL replay after leader failover --- .../consensus/ConsensusPrefetchingQueue.java | 20 ++--- .../ConsensusPrefetchingQueueTest.java | 86 +++++++++++++++++++ .../consensus/ProgressWALIteratorTest.java | 11 ++- 3 files changed, 100 insertions(+), 17 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java index 2d2060e15f75..271ff0ebb1bd 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java @@ -1003,8 +1003,7 @@ protected ReplayLocateDecision scanReplayStartForRequests( final Map effectiveRecoveryWriterProgress = new LinkedHashMap<>(requestedWriterProgress); final Set exactVisibleWriterIds = new LinkedHashSet<>(); - Long firstUncoveredReplayableSearchIndex = null; - boolean sawBlockingNonReplayableUncovered = false; + Long firstUncoveredLocalSearchIndex = null; while (requests.hasNext()) { final IndexedConsensusRequest request = requests.next(); @@ -1026,11 +1025,9 @@ && compareWriterProgress(requestProgress, storedWriterProgress) == 0) { } if (request.getSearchIndex() >= 0) { - if (Objects.isNull(firstUncoveredReplayableSearchIndex)) { - firstUncoveredReplayableSearchIndex = request.getSearchIndex(); + if (Objects.isNull(firstUncoveredLocalSearchIndex)) { + firstUncoveredLocalSearchIndex = request.getSearchIndex(); } - } else if (Objects.isNull(firstUncoveredReplayableSearchIndex)) { - sawBlockingNonReplayableUncovered = true; } } @@ -1045,14 +1042,11 @@ && compareWriterProgress(requestProgress, storedWriterProgress) == 0) { final RegionProgress effectiveRecoveryRegionProgress = new RegionProgress(effectiveRecoveryWriterProgress); - if (sawBlockingNonReplayableUncovered) { - return ReplayLocateDecision.locateMiss( - effectiveRecoveryRegionProgress, - "uncovered non-replayable WAL records appear before the first local replayable record"); - } - if (Objects.nonNull(firstUncoveredReplayableSearchIndex)) { + // The iterator's lower bound filters only locally indexed requests. Replicated requests stay + // visible and are deduplicated by writer progress, so they do not block local cursor lookup. + if (Objects.nonNull(firstUncoveredLocalSearchIndex)) { return ReplayLocateDecision.found( - firstUncoveredReplayableSearchIndex, + firstUncoveredLocalSearchIndex, effectiveRecoveryRegionProgress, "resolved first uncovered replayable WAL record"); } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueTest.java index b1fd02cc4c51..056a5f2b6c48 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueTest.java @@ -106,6 +106,78 @@ public void testInitializationAndActivationUseIndependentMonitors() throws Excep .getModifiers())); } + @Test + public void testReplayStartPreservesUncoveredFollowerEntries() throws Exception { + final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); + final File systemDir = temporaryFolder.newFolder("replay-start-with-follower-entry"); + ConsensusPrefetchingQueue queue = null; + try { + final DataRegionId regionId = new DataRegionId(1); + final FakeConsensusReqReader reader = new FakeConsensusReqReader(); + reader.currentSearchIndex = 1L; + final IoTConsensusServerImpl serverImpl = mock(IoTConsensusServerImpl.class); + when(serverImpl.getConsensusReqReader()).thenReturn(reader); + when(serverImpl.getWriterSafeFrontierTracker()).thenReturn(new WriterSafeFrontierTracker()); + + queue = + new ConsensusPrefetchingQueue( + "consumerGroup", + "topic", + TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE, + regionId, + serverImpl, + new SubscriptionWalRetentionPolicy( + "topic", + SubscriptionWalRetentionPolicy.UNBOUNDED, + SubscriptionWalRetentionPolicy.UNBOUNDED), + mock(ConsensusLogToTabletConverter.class), + newCommitManager(systemDir), + new RegionProgress(Collections.emptyMap()), + 1L, + 1L, + true); + + final WriterId formerLeader = new WriterId(regionId.toString(), 8); + final WriterProgress committedProgress = new WriterProgress(100L, 10L); + final RegionProgress regionProgress = + new RegionProgress(Collections.singletonMap(formerLeader, committedProgress)); + final List requests = + Arrays.asList( + createRequest(-1L, 10L, 100L, 8), + createRequest(-1L, 11L, 101L, 8), + createRequest(1L, 1L, 200L, 7)); + + final ConsensusPrefetchingQueue.ReplayLocateDecision decision = + queue.scanReplayStartForRequests(requests.iterator(), regionProgress, true); + + assertEquals(ConsensusPrefetchingQueue.ReplayLocateStatus.FOUND, decision.getStatus()); + assertEquals(1L, decision.getStartSearchIndex()); + assertEquals( + committedProgress, + decision.getRecoveryRegionProgress().getWriterPositions().get(formerLeader)); + + // With no uncovered local request, keep the local cursor at the tail without advancing the + // recovery progress past the still-uncovered follower request. + reader.currentSearchIndex = 5L; + final ConsensusPrefetchingQueue.ReplayLocateDecision tailDecision = + queue.scanReplayStartForRequests( + Collections.singletonList(createRequest(-1L, 11L, 101L, 8)).iterator(), + regionProgress, + true); + + assertEquals(ConsensusPrefetchingQueue.ReplayLocateStatus.AT_END, tailDecision.getStatus()); + assertEquals(5L, tailDecision.getStartSearchIndex()); + assertEquals( + committedProgress, + tailDecision.getRecoveryRegionProgress().getWriterPositions().get(formerLeader)); + } finally { + if (queue != null) { + queue.close(); + } + IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir); + } + } + @Test @SuppressWarnings("unchecked") public void testAdmissionClearCannotLeaveEntryEnqueuedAfterFence() throws Exception { @@ -1858,6 +1930,20 @@ private static IndexedConsensusRequest createRequest(final long searchIndex) { .setNodeId(7); } + private static IndexedConsensusRequest createRequest( + final long searchIndex, + final long localSeq, + final long physicalTime, + final int writerNodeId) { + return new IndexedConsensusRequest( + searchIndex, + localSeq, + Collections.singletonList( + StatementTestUtils.genInsertRowNode(Math.toIntExact(localSeq)))) + .setPhysicalTime(physicalTime) + .setNodeId(writerNodeId); + } + private static IndexedConsensusRequest createSizedRequest( final long searchIndex, final long rawMemorySize, final int serializedMemorySize) { final IndexedConsensusRequest request = diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIteratorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIteratorTest.java index 92928663126c..fed8febea521 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIteratorTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIteratorTest.java @@ -242,7 +242,8 @@ public void testIteratorDoesNotSkipNextWalFileAfterExhaustingCurrentOne() throws } @Test - public void testFollowerEntryDoesNotSynthesizeSearchIndexFromProgressLocalSeq() throws Exception { + public void testLocalLowerBoundKeepsFollowerEntryWithoutSynthesizingSearchIndex() + throws Exception { final Path dir = Files.createTempDirectory("progress-wal-iterator-follower"); final File firstWal = dir.resolve(WALFileUtils.getLogFileName(0, 0, WALFileStatus.CONTAINS_SEARCH_INDEX)) @@ -255,17 +256,19 @@ public void testFollowerEntryDoesNotSynthesizeSearchIndexFromProgressLocalSeq() try (WALWriter writer = new WALWriter(firstWal, WALFileVersion.V3)) { writer.write(searchableEntry(-1L), singleEntryMeta(19, -1L, 1L, 900L, 5, 1009L)); } - try (WALWriter ignored = new WALWriter(lastWal, WALFileVersion.V3)) { - // Create a readable successor for the first WAL file. + try (WALWriter writer = new WALWriter(lastWal, WALFileVersion.V3)) { + writer.write(searchableEntry(1L), singleEntryMeta(19, 1L, 1L, 1000L, 6, 1L)); } - try (ProgressWALIterator iterator = new ProgressWALIterator(dir.toFile(), Long.MIN_VALUE)) { + try (ProgressWALIterator iterator = new ProgressWALIterator(dir.toFile(), 1L)) { assertTrue(iterator.hasNext()); final IndexedConsensusRequest request = iterator.next(); assertEquals(-1L, request.getSearchIndex()); assertEquals(1009L, request.getProgressLocalSeq()); assertEquals(900L, request.getPhysicalTime()); assertEquals(5, request.getNodeId()); + assertTrue(iterator.hasNext()); + assertEquals(1L, iterator.next().getSearchIndex()); assertFalse(iterator.hasNext()); } } finally {