apache · Caideyipi · Jul 31, 2026
diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
@@ -2099,11 +2099,11 @@ private DataNodePipeMessages() {}
   public static final String PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_PREFETCH_INITIALIZED_STARTSEARCHINDEX_69B53EE6 =
       "ConsensusPrefetchingQueue {}: prefetch initialized, startSearchIndex={}, progressSource={}, "
           + "recoveryWriterCount={}";
-  public static final String PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_PERIODIC_STATS_LAG_PENDINGDELTA_D75375D0 =
+  public static final String PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_PERIODIC_STATS_LAG_PENDINGDELTA_WALGAPSKIPPEDENTRIES_9A4E6608 =
       "ConsensusPrefetchingQueue {}: periodic stats, lag={}, pendingDelta={}, walDelta={}, "
-          + "pendingTotal={}, walTotal={}, pendingQueueSize={}, prefetchingQueueSize={}, "
-          + "inFlightEventsSize={}, realtimeWriterCount={}, walHasNext={}, isActive={}, "
-          + "subtaskScheduled={}";
+          + "pendingTotal={}, walTotal={}, walGapSkippedEntries={}, pendingQueueSize={}, "
+          + "prefetchingQueueSize={}, inFlightEventsSize={}, realtimeWriterCount={}, "
+          + "walHasNext={}, isActive={}, subtaskScheduled={}";
   public static final String PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_WAITING_MS_FOR_WAL_GAP_TO_BECOME_7D91C6C5 =
       "ConsensusPrefetchingQueue {}: waiting {}ms for WAL gap [{}, {}) to become visible, "
           + "currentNextExpected={}, currentWalIndex={}, seekGeneration={}";
@@ -2183,6 +2183,14 @@ private DataNodePipeMessages() {}
       "ProgressWALIterator: error reading WAL";
   public static final String PIPE_LOG_PROGRESSWALITERATOR_FAILED_TO_OPEN_WAL_FILE_SKIPPING_29CA1092 =
       "ProgressWALIterator: failed to open WAL file {}, skipping";
+  public static final String PIPE_LOG_PROGRESSWALITERATOR_SKIPPED_UNREADABLE_RETAINED_WAL_FILES_FFC8455E =
+      "ProgressWALIterator: skipped {} unreadable retained WAL files in directory {}, "
+          + "firstFile={}, lastFile={}, firstError={}; historical subscription data in these "
+          + "files cannot be replayed";
+  public static final String PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_WAL_REPLAY_SKIPPED_UNAVAILABLE_SEARCH_INDEXES_B8023B64 =
+      "ConsensusPrefetchingQueue {}: WAL replay skipped unavailable search indexes [{}, {}), "
+          + "skippedEntries={}, totalWalGapSkippedEntries={}; the missing WAL data may have been "
+          + "reclaimed before subscription consumption";
   public static final String PIPE_LOG_PIPE_TERMINATE_EVENT_COMMITTED_FOR_HISTORICAL_TRANSFER_CREATIONTIME_9B807B28 =
       "Pipe {}@{}: terminate event committed for historical transfer. creationTime: {}, "
           + "shouldMark: {}. {}";

diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
@@ -1948,10 +1948,11 @@ private DataNodePipeMessages() {}
   public static final String PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_PREFETCH_INITIALIZED_STARTSEARCHINDEX_69B53EE6 =
       "ConsensusPrefetchingQueue {}：prefetch 已初始化，startSearchIndex={}，progressSource={}，"
           + "recoveryWriterCount={}";
-  public static final String PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_PERIODIC_STATS_LAG_PENDINGDELTA_D75375D0 =
+  public static final String PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_PERIODIC_STATS_LAG_PENDINGDELTA_WALGAPSKIPPEDENTRIES_9A4E6608 =
       "ConsensusPrefetchingQueue {}：周期统计，lag={}，pendingDelta={}，walDelta={}，pendingTotal={}，"
-          + "walTotal={}，pendingQueueSize={}，prefetchingQueueSize={}，inFlightEventsSize={}，"
-          + "realtimeWriterCount={}，walHasNext={}，isActive={}，subtaskScheduled={}";
+          + "walTotal={}，walGapSkippedEntries={}，pendingQueueSize={}，prefetchingQueueSize={}，"
+          + "inFlightEventsSize={}，realtimeWriterCount={}，walHasNext={}，isActive={}，"
+          + "subtaskScheduled={}";
   public static final String PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_WAITING_MS_FOR_WAL_GAP_TO_BECOME_7D91C6C5 =
       "ConsensusPrefetchingQueue {}：等待 {}ms，使 WAL 缺口 [{}, {}) 可见，currentNextExpected={}，"
           + "currentWalIndex={}，seekGeneration={}";
@@ -2027,6 +2028,12 @@ private DataNodePipeMessages() {}
       "ProgressWALIterator：读取 WAL 出错";
   public static final String PIPE_LOG_PROGRESSWALITERATOR_FAILED_TO_OPEN_WAL_FILE_SKIPPING_29CA1092 =
       "ProgressWALIterator：打开 WAL 文件 {} 失败，跳过该文件";
+  public static final String PIPE_LOG_PROGRESSWALITERATOR_SKIPPED_UNREADABLE_RETAINED_WAL_FILES_FFC8455E =
+      "ProgressWALIterator：跳过了 {} 个无法读取的保留 WAL 文件，directory={}，firstFile={}，"
+          + "lastFile={}，firstError={}；这些文件中的历史订阅数据无法重放";
+  public static final String PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_WAL_REPLAY_SKIPPED_UNAVAILABLE_SEARCH_INDEXES_B8023B64 =
+      "ConsensusPrefetchingQueue {}：WAL 重放跳过了不可用的 searchIndex 区间 [{}, {})，"
+          + "skippedEntries={}，totalWalGapSkippedEntries={}；缺失的 WAL 数据可能已在订阅消费前被回收";
   public static final String PIPE_LOG_PIPE_TERMINATE_EVENT_COMMITTED_FOR_HISTORICAL_TRANSFER_CREATIONTIME_9B807B28 =
       "Pipe {}@{}：历史传输的终止事件已提交。creationTime：{}，shouldMark：{}。{}";
   public static final String PIPE_LOG_PIPE_HISTORICAL_SOURCE_HAS_SUPPLIED_ALL_EVENTS_EMITTING_8B58DE19 =

diff --git a/...ain/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java b/...ain/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java
@@ -1480,13 +1480,14 @@ private void logPeriodicStatsIfNecessary() {
     final long currentWalAcceptedEntries = walPathAcceptedEntries.get();
     LOGGER.info(
         DataNodePipeMessages
-            .PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_PERIODIC_STATS_LAG_PENDINGDELTA_D75375D0,
+            .PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_PERIODIC_STATS_LAG_PENDINGDELTA_WALGAPSKIPPEDENTRIES_9A4E6608,
         this,
         getLag(),
         currentPendingAcceptedEntries - lastPendingAcceptedEntries,
         currentWalAcceptedEntries - lastWalAcceptedEntries,
         currentPendingAcceptedEntries,
         currentWalAcceptedEntries,
+        walGapSkippedEntries.get(),
         pendingEntries.size(),
         prefetchingQueue.size(),
         inFlightEvents.size(),
@@ -1835,11 +1836,11 @@ private MaterializationResult pumpFromSubscriptionWAL(
           continue;
         }
         if (shouldSkipForRecoveryProgress(walEntry)) {
-          advanceLocalCursorIfPresent(walEntry);
+          advanceWalReplayCursorIfPresent(walEntry);
           continue;
         }
         if (shouldSkipForMaterializedProgress(walEntry)) {
-          advanceLocalCursorIfPresent(walEntry);
+          advanceWalReplayCursorIfPresent(walEntry);
           continue;
         }
 
@@ -1853,7 +1854,7 @@ private MaterializationResult pumpFromSubscriptionWAL(
           return appendResult;
         }
         markMaterializedProgress(walEntry);
-        advanceLocalCursorIfPresent(walEntry);
+        advanceWalReplayCursorIfPresent(walEntry);
       } catch (final Exception e) {
         LOGGER.warn(
             DataNodePipeMessages
@@ -1875,6 +1876,28 @@ private MaterializationResult pumpFromSubscriptionWAL(
     return MaterializationResult.SUCCESS;
   }
 
+  private void advanceWalReplayCursorIfPresent(final IndexedConsensusRequest request) {
+    if (!hasLocalSearchIndex(request)) {
+      return;
+    }
+
+    final long actualSearchIndex = request.getSearchIndex();
+    final long expectedSearchIndex = nextExpectedSearchIndex.get();
+    if (actualSearchIndex > expectedSearchIndex) {
+      final long skippedEntries = actualSearchIndex - expectedSearchIndex;
+      final long totalSkippedEntries = walGapSkippedEntries.addAndGet(skippedEntries);
+      LOGGER.warn(
+          DataNodePipeMessages
+              .PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_WAL_REPLAY_SKIPPED_UNAVAILABLE_SEARCH_INDEXES_B8023B64,
+          this,
+          expectedSearchIndex,
+          actualSearchIndex,
+          skippedEntries,
+          totalSkippedEntries);
+    }
+    nextExpectedSearchIndex.set(actualSearchIndex + 1);
+  }
+
   private void ensureSubscriptionWalReadable() {
     if (Objects.isNull(subscriptionWALIterator)
         || subscriptionWALIterator.hasNext()

diff --git a/.../src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIterator.java b/.../src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIterator.java
@@ -74,6 +74,10 @@ public class ProgressWALIterator implements Closeable, Iterator<IndexedConsensus
   private boolean currentReaderUsesLiveSnapshot = false;
   private int consumedEntryCountInCurrentFile = 0;
   private final Set<Long> skippedBrokenWalVersionIds = new HashSet<>();
+  private int unreportedSkippedBrokenWalFileCount = 0;
+  private String firstUnreportedSkippedBrokenWalFile;
+  private String lastUnreportedSkippedBrokenWalFile;
+  private String firstUnreportedSkippedBrokenWalError;
   private IOException lastError;
   private boolean incompleteScan = false;
   private String incompleteScanDetail;
@@ -172,7 +176,9 @@ public boolean hasNext() {
     } catch (IOException e) {
       lastError = e;
       LOGGER.warn(DataNodePipeMessages.PIPE_LOG_PROGRESSWALITERATOR_ERROR_READING_WAL_2DB46D41, e);
-      return false;
+    }
+    if (nextReady == null) {
+      logSkippedBrokenWalFilesIfNecessary();
     }
     return nextReady != null;
   }
@@ -198,6 +204,10 @@ public boolean hasSkippedBrokenWalFiles() {
     return !skippedBrokenWalVersionIds.isEmpty();
   }
 
+  int getSkippedBrokenWalFileCount() {
+    return skippedBrokenWalVersionIds.size();
+  }
+
   public boolean hasIncompleteScan() {
     return incompleteScan || hasReadError() || hasSkippedBrokenWalFiles();
   }
@@ -406,16 +416,54 @@ private boolean openReaderAtIndex(
             e);
         return false;
       }
-      skippedBrokenWalVersionIds.add(versionId);
-      LOGGER.warn(
-          DataNodePipeMessages
-              .PIPE_LOG_PROGRESSWALITERATOR_FAILED_TO_OPEN_WAL_FILE_SKIPPING_29CA1092,
-          walFile.getName(),
-          e);
+      recordSkippedBrokenWalFile(versionId, walFile, e);
       return false;
     }
   }
 
+  private void recordSkippedBrokenWalFile(
+      final long versionId, final File walFile, final IOException error) {
+    if (!skippedBrokenWalVersionIds.add(versionId)) {
+      return;
+    }
+
+    if (unreportedSkippedBrokenWalFileCount == 0) {
+      firstUnreportedSkippedBrokenWalFile = walFile.getName();
+      firstUnreportedSkippedBrokenWalError = summarizeException(error);
+    }
+    lastUnreportedSkippedBrokenWalFile = walFile.getName();
+    unreportedSkippedBrokenWalFileCount++;
+    LOGGER.debug(
+        DataNodePipeMessages.PIPE_LOG_PROGRESSWALITERATOR_FAILED_TO_OPEN_WAL_FILE_SKIPPING_29CA1092,
+        walFile.getName(),
+        error);
+  }
+
+  private void logSkippedBrokenWalFilesIfNecessary() {
+    if (unreportedSkippedBrokenWalFileCount == 0) {
+      return;
+    }
+
+    LOGGER.warn(
+        DataNodePipeMessages
+            .PIPE_LOG_PROGRESSWALITERATOR_SKIPPED_UNREADABLE_RETAINED_WAL_FILES_FFC8455E,
+        unreportedSkippedBrokenWalFileCount,
+        logDirectory,
+        firstUnreportedSkippedBrokenWalFile,
+        lastUnreportedSkippedBrokenWalFile,
+        firstUnreportedSkippedBrokenWalError);
+    unreportedSkippedBrokenWalFileCount = 0;
+    firstUnreportedSkippedBrokenWalFile = null;
+    lastUnreportedSkippedBrokenWalFile = null;
+    firstUnreportedSkippedBrokenWalError = null;
+  }
+
+  private static String summarizeException(final IOException error) {
+    return error.getMessage() == null
+        ? error.getClass().getSimpleName()
+        : error.getClass().getSimpleName() + ": " + error.getMessage();
+  }
+
   private boolean skipEntries(final ProgressWALReader reader, final int skipEntries)
       throws IOException {
     return reader.skipToEntryIndex(skipEntries);

diff --git a/...java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueTest.java b/...java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueTest.java
@@ -58,6 +58,7 @@
 import java.lang.reflect.Modifier;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.Iterator;
 import java.util.List;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CountDownLatch;
@@ -742,6 +743,67 @@ ProgressWALReader openProgressWALReader(final File walFile) throws IOException {
     }
   }
 
+  @Test
+  public void testWalReplayCountsOnlyUnavailableSearchIndexes() throws Exception {
+    final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir();
+    final File systemDir = temporaryFolder.newFolder("wal-replay-gap-counter");
+    ConsensusPrefetchingQueue queue = null;
+    try {
+      final DataRegionId regionId = new DataRegionId(9);
+      final FakeConsensusReqReader reader = new FakeConsensusReqReader();
+      reader.currentSearchIndex = 4L;
+      final IoTConsensusServerImpl serverImpl = mock(IoTConsensusServerImpl.class);
+      when(serverImpl.getConsensusReqReader()).thenReturn(reader);
+      when(serverImpl.getWriterSafeFrontierTracker()).thenReturn(new WriterSafeFrontierTracker());
+
+      final ConsensusLogToTabletConverter converter = mock(ConsensusLogToTabletConverter.class);
+      when(converter.convert(any())).thenReturn(Collections.singletonList(createTablet()));
+      when(converter.getDatabaseName()).thenReturn("db");
+
+      final Iterator<IndexedConsensusRequest> retainedWalEntries =
+          Arrays.asList(createRequest(1L), createRequest(4L)).iterator();
+      final ProgressWALIterator walIterator = mock(ProgressWALIterator.class);
+      when(walIterator.hasNext()).thenAnswer(ignored -> retainedWalEntries.hasNext());
+      when(walIterator.next()).thenAnswer(ignored -> retainedWalEntries.next());
+
+      queue =
+          new ConsensusPrefetchingQueue(
+              "consumerGroup",
+              "topic",
+              TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE,
+              regionId,
+              serverImpl,
+              new SubscriptionWalRetentionPolicy(
+                  "topic",
+                  SubscriptionWalRetentionPolicy.UNBOUNDED,
+                  SubscriptionWalRetentionPolicy.UNBOUNDED),
+              converter,
+              newCommitManager(systemDir),
+              new RegionProgress(Collections.emptyMap()),
+              1L,
+              1L,
+              true) {
+            @Override
+            protected ProgressWALIterator createSubscriptionWALIterator(
+                final long startSearchIndex) {
+              return walIterator;
+            }
+          };
+
+      assertNull(queue.poll("consumer"));
+      queue.drivePrefetchOnce();
+
+      assertEquals(2L, queue.getWalPathAcceptedEntries());
+      assertEquals(2L, queue.getWalGapSkippedEntries());
+      assertEquals(5L, queue.getCurrentReadSearchIndex());
+    } finally {
+      if (queue != null) {
+        queue.close();
+      }
+      IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir);
+    }
+  }
+
   @Test
   public void testActivationRetriesUntilConfigNodeProgressIsExplicitlyAvailable() throws Exception {
     final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir();

diff --git a/.../test/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIteratorTest.java b/.../test/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIteratorTest.java
@@ -272,6 +272,44 @@ public void testFollowerEntryDoesNotSynthesizeSearchIndexFromProgressLocalSeq()
     }
   }
 
+  @Test
+  public void testIteratorAggregatesUnreadableRetainedWalFiles() throws Exception {
+    final Path dir = Files.createTempDirectory("progress-wal-iterator-unreadable-files");
+    final File firstBrokenWal =
+        dir.resolve(WALFileUtils.getLogFileName(0, 0, WALFileStatus.CONTAINS_SEARCH_INDEX))
+            .toFile();
+    final File secondBrokenWal =
+        dir.resolve(WALFileUtils.getLogFileName(1, 1, WALFileStatus.CONTAINS_SEARCH_INDEX))
+            .toFile();
+    final File lastWal =
+        dir.resolve(WALFileUtils.getLogFileName(2, 2, WALFileStatus.CONTAINS_SEARCH_INDEX))
+            .toFile();
+
+    try {
+      Files.write(firstBrokenWal.toPath(), new byte[128]);
+      Files.write(secondBrokenWal.toPath(), new byte[128]);
+      try (WALWriter writer = new WALWriter(lastWal, WALFileVersion.V3)) {
+        // Create a readable successor so both malformed WAL files are treated as retained history.
+        writer.write(searchableEntry(2L), singleEntryMeta(19, 2L, 1L, 200L, 7, 2L));
+      }
+
+      try (ProgressWALIterator iterator = new ProgressWALIterator(dir.toFile(), Long.MIN_VALUE)) {
+        assertTrue(iterator.hasNext());
+        assertEquals(2L, iterator.next().getSearchIndex());
+        assertFalse(iterator.hasNext());
+        assertTrue(iterator.hasSkippedBrokenWalFiles());
+        assertEquals(2, iterator.getSkippedBrokenWalFileCount());
+        assertTrue(iterator.hasIncompleteScan());
+        assertFalse(iterator.hasReadError());
+      }
+    } finally {
+      Files.deleteIfExists(firstBrokenWal.toPath());
+      Files.deleteIfExists(secondBrokenWal.toPath());
+      Files.deleteIfExists(lastWal.toPath());
+      Files.deleteIfExists(dir);
+    }
+  }
+
   @Test
   public void testIteratorMarksIncompleteScanWhenNearLiveWalCannotBeOpened() throws Exception {
     final Path dir = Files.createTempDirectory("progress-wal-iterator-incomplete-scan");