[Subscription] Fix WAL replay after leader failover

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java

@@ -1003,8 +1003,7 @@ protected ReplayLocateDecision scanReplayStartForRequests(
     final Map<WriterId, WriterProgress> effectiveRecoveryWriterProgress =
         new LinkedHashMap<>(requestedWriterProgress);
     final Set<WriterId> 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");
     }

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<IndexedConsensusRequest> 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 =

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 {