[Subscription] Keep WAL iterator across bounded gap rounds

Author: Caideyipi

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

@@ -1861,10 +1861,15 @@ private MaterializationResult fillGapFromWAL(
       final int maxTablets,
       final long maxBatchBytes) {
     pendingWalGapRetryRequested = false;
-    resetSubscriptionWALPosition(fromIndex);
     if (seekGeneration.get() != expectedSeekGeneration || isClosed) {
       return MaterializationResult.STALE;
     }
+    // Rebuilding the iterator for every bounded gap round rereads the retained WAL prefix to
+    // reach fromIndex. Keep its current position; advanceTo skips whole files only when their
+    // writer progress is already covered, and otherwise replays them sequentially.
+    if (Objects.nonNull(subscriptionWALIterator)) {
+      subscriptionWALIterator.advanceTo(fromIndex, this::isWriterProgressCoveredForWalFastForward);
+    }
     // A pending-queue jump can span millions of WAL entries when real-time delivery has overflowed.
     // Keep gap recovery within the normal per-round budget so sparse topic filtering cannot hold
     // the

iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueTest.java

@@ -1410,17 +1410,8 @@ public void testPendingGapReplayHonorsPerRoundWalEntryLimit() throws Exception {
               });
       when(converter.getDatabaseName()).thenReturn("db");
 
-      final Iterator<IndexedConsensusRequest> walEntries =
-          Arrays.asList(
-                  createRequest(1L),
-                  createRequest(2L),
-                  createRequest(3L),
-                  createRequest(4L),
-                  createRequest(5L))
-              .iterator();
-      final ProgressWALIterator iterator = mock(ProgressWALIterator.class);
-      when(iterator.hasNext()).thenAnswer(ignored -> walEntries.hasNext());
-      when(iterator.next()).thenAnswer(ignored -> walEntries.next());
+      final AtomicInteger iteratorCreationCount = new AtomicInteger();
+      final AtomicInteger walEntryReadCount = new AtomicInteger();
 
       queue =
           new ConsensusPrefetchingQueue(
@@ -1442,6 +1433,23 @@ public void testPendingGapReplayHonorsPerRoundWalEntryLimit() throws Exception {
             @Override
             protected ProgressWALIterator createSubscriptionWALIterator(
                 final long startSearchIndex) {
+              iteratorCreationCount.incrementAndGet();
+              final Iterator<IndexedConsensusRequest> walEntries =
+                  Arrays.asList(
+                          createRequest(1L),
+                          createRequest(2L),
+                          createRequest(3L),
+                          createRequest(4L),
+                          createRequest(5L))
+                      .iterator();
+              final ProgressWALIterator iterator = mock(ProgressWALIterator.class);
+              when(iterator.hasNext()).thenAnswer(ignored -> walEntries.hasNext());
+              when(iterator.next())
+                  .thenAnswer(
+                      ignored -> {
+                        walEntryReadCount.incrementAndGet();
+                        return walEntries.next();
+                      });
               return iterator;
             }
           };
@@ -1462,6 +1470,8 @@ protected ProgressWALIterator createSubscriptionWALIterator(
       queue.drivePrefetchOnce();
       assertEquals(5, conversionCount.get());
       assertEquals(6L, queue.getCurrentReadSearchIndex());
+      assertEquals(1, iteratorCreationCount.get());
+      assertEquals(5, walEntryReadCount.get());
     } finally {
       if (queue != null) {
         queue.close();