[Subscription] Fix WAL replay progress after realtime gaps

iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java

@@ -20,6 +20,8 @@
 package org.apache.iotdb.db.i18n;
 
 public final class DataNodePipeMessages {
+  public static final String LOG_SUBSCRIPTION_REPLAY_HAS_MADE_NO_CURSOR_PROGRESS_FOR_ARG_MS_ARG_INSPECT_PREFETCH_WORKER_STACKS_AND_WAL_I_O_FAE1F64B =
+      "Subscription replay has made no cursor progress for {} ms: {}. Inspect prefetch worker stacks and WAL I/O.";
 
   public static final String
       LOG_SUBSCRIPTION_PREFETCH_ROUND_STARTING_FOR_ARG_ACTIVE_ARG_INITIALIZED_ARG_CLOSED_ARG_PENDINGENTRIES_ARG_PREFETCHEDEVENTS_ARG_INFLIGHTEVENTS_ARG_NEXTREADSEARCHINDEX_ARG_F75FF2CD =

iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java

@@ -20,6 +20,8 @@
 package org.apache.iotdb.db.i18n;
 
 public final class DataNodePipeMessages {
+  public static final String LOG_SUBSCRIPTION_REPLAY_HAS_MADE_NO_CURSOR_PROGRESS_FOR_ARG_MS_ARG_INSPECT_PREFETCH_WORKER_STACKS_AND_WAL_I_O_FAE1F64B =
+      "订阅回放游标已连续 {} ms 未推进：{}。请检查预取工作线程栈和 WAL I/O。";
 
   public static final String
       LOG_SUBSCRIPTION_PREFETCH_ROUND_STARTING_FOR_ARG_ACTIVE_ARG_INITIALIZED_ARG_CLOSED_ARG_PENDINGENTRIES_ARG_PREFETCHEDEVENTS_ARG_INFLIGHTEVENTS_ARG_NEXTREADSEARCHINDEX_ARG_F75FF2CD =

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

@@ -321,6 +321,17 @@ public class ConsensusPrefetchingQueue {
 
   private final SubscriptionQueueTimeTracker timeTracker = new SubscriptionQueueTimeTracker();
 
+  private volatile long lastPrefetchDurationMs;
+  private volatile PrefetchRoundResult.Type lastPrefetchResult;
+  private volatile String walReplayState = "NOT_SCANNED";
+
+  // Accessed only by the serialized prefetch worker while holding the queue read lock.
+  private boolean replayProgressObserved;
+  private long lastObservedReplayCursor;
+  private long lastReplayProgressNanos;
+  private boolean replayStallWarningLogged;
+  private long lastReplayStallWarningNanos;
+
   private volatile String lastConsumerId = "";
 
   private volatile long lastPendingAcceptedEntries = 0L;
@@ -1031,8 +1042,7 @@ private boolean initPrefetchUnderInitializationLock(final RegionProgress regionP
     this.nextExpectedSearchIndex.set(resolvedStart.getStartSearchIndex());
     // Use the same factory as cursor resets; the default implementation still returns null for
     // readers without WAL support.
-    this.subscriptionWALIterator =
-        createSubscriptionWALIterator(resolvedStart.getStartSearchIndex());
+    resetSubscriptionWALPosition(resolvedStart.getStartSearchIndex());
     this.prefetchInitialized = true;
     timeTracker.start();
     this.observedSeekGeneration = seekGeneration.get();
@@ -1617,6 +1627,8 @@ public SubscriptionEvent pollTablets(
   private static final long WAL_GAP_WAIT_LOG_INTERVAL_MS = 5_000L;
 
   private static final long PREFETCH_STATS_LOG_INTERVAL_MS = 5_000L;
+  private static final long REPLAY_STALL_WARNING_THRESHOLD_NANOS = TimeUnit.SECONDS.toNanos(60L);
+  private static final long REPLAY_STALL_WARNING_INTERVAL_NANOS = TimeUnit.SECONDS.toNanos(30L);
 
   private static final long SLOW_PREFETCH_ROUND_THRESHOLD_MS = 10_000L;
 
@@ -1649,6 +1661,8 @@ public PrefetchRoundResult drivePrefetchOnce() {
       return result;
     } finally {
       final long durationMs = timeTracker.endPrefetch();
+      lastPrefetchDurationMs = durationMs;
+      lastPrefetchResult = Objects.nonNull(result) ? result.getType() : null;
       if (debugEnabled) {
         LOGGER.debug(
             DataNodePipeMessages
@@ -1859,6 +1873,8 @@ private PrefetchRoundResult drivePrefetchOnceInternal() {
       }
       return PrefetchRoundResult.rescheduleAfter(100L);
     } finally {
+      final ProgressWALIterator iterator = subscriptionWALIterator;
+      walReplayState = Objects.nonNull(iterator) ? iterator.toString() : "UNAVAILABLE";
       releaseReadLock();
     }
   }
@@ -1869,6 +1885,8 @@ private void logPeriodicStatsIfNecessary() {
       return;
     }
 
+    reportReplayStallIfNecessary(System.nanoTime());
+
     final long currentPendingAcceptedEntries = pendingPathAcceptedEntries.get();
     final long currentWalAcceptedEntries = walPathAcceptedEntries.get();
     LOGGER.info(
@@ -1893,7 +1911,36 @@ private void logPeriodicStatsIfNecessary() {
     lastWalAcceptedEntries = currentWalAcceptedEntries;
   }
 
+  private void reportReplayStallIfNecessary(final long nowNanos) {
+    final long cursor = nextExpectedSearchIndex.get();
+    if (!prefetchInitialized || !isActive || isClosed || closeRequested || getLag() <= 0L) {
+      replayProgressObserved = false;
+      replayStallWarningLogged = false;
+      return;
+    }
+    if (!replayProgressObserved || cursor != lastObservedReplayCursor) {
+      replayProgressObserved = true;
+      lastObservedReplayCursor = cursor;
+      lastReplayProgressNanos = nowNanos;
+      replayStallWarningLogged = false;
+      return;
+    }
+    final long stalledNanos = nowNanos - lastReplayProgressNanos;
+    if (stalledNanos >= REPLAY_STALL_WARNING_THRESHOLD_NANOS
+        && (!replayStallWarningLogged
+            || nowNanos - lastReplayStallWarningNanos >= REPLAY_STALL_WARNING_INTERVAL_NANOS)) {
+      LOGGER.warn(
+          DataNodePipeMessages
+              .LOG_SUBSCRIPTION_REPLAY_HAS_MADE_NO_CURSOR_PROGRESS_FOR_ARG_MS_ARG_INSPECT_PREFETCH_WORKER_STACKS_AND_WAL_I_O_FAE1F64B,
+          TimeUnit.NANOSECONDS.toMillis(stalledNanos),
+          this);
+      lastReplayStallWarningNanos = nowNanos;
+      replayStallWarningLogged = true;
+    }
+  }
+
   private void resetRoundStateForSeek(final long newSeekGeneration) {
+    replayProgressObserved = false;
     restorePendingSubscriptionWalCursor(newSeekGeneration);
     discardRealtimeEntries();
     discardBatch(lingerBatch);
@@ -2473,7 +2520,12 @@ private void resetSubscriptionWALPosition(final long startSearchIndex) {
 
   protected ProgressWALIterator createSubscriptionWALIterator(final long startSearchIndex) {
     if (consensusReqReader instanceof WALNode) {
-      return new ProgressWALIterator((WALNode) consensusReqReader, startSearchIndex);
+      final ProgressWALIterator iterator =
+          new ProgressWALIterator((WALNode) consensusReqReader, startSearchIndex);
+      // The constructor sets the local filtering lower bound. Complete initialization here by
+      // positioning past historical files only when every writer's progress is already covered.
+      iterator.advanceTo(startSearchIndex, this::isWriterProgressCoveredForWalFastForward);
+      return iterator;
     }
     return null;
   }
@@ -5093,6 +5145,9 @@ public Map<String, String> coreReportMessage() {
     result.put("maxDeliveryIntervalMs", String.valueOf(getMaxDeliveryIntervalMs()));
     result.put("deliveryIdleTimeMs", String.valueOf(getDeliveryIdleTimeMs()));
     result.put("prefetchDurationMs", String.valueOf(getPrefetchDurationMs()));
+    result.put("lastPrefetchDurationMs", String.valueOf(lastPrefetchDurationMs));
+    result.put("lastPrefetchResult", String.valueOf(lastPrefetchResult));
+    result.put("walReplayState", walReplayState);
     result.put("maxPrefetchDurationMs", String.valueOf(getMaxPrefetchDurationMs()));
     result.put("prefetchIdleTimeMs", String.valueOf(getPrefetchIdleTimeMs()));
     result.put("realtimeAdmissionBlocked", String.valueOf(realtimeAdmissionBlocked.get()));

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

@@ -69,12 +69,14 @@ public interface WriterProgressCoverage {
 
   /**
    * One shared budget for all WAL probes and reads in a prefetch round. The time limit is
-   * cooperative: it is checked between operations and cannot interrupt a single file read.
+   * cooperative: it starts with the first WAL probe, is checked between operations, and cannot
+   * interrupt a single file read. Queue housekeeping must not consume the scan's time slice.
    */
   static final class ScanBudget {
 
     private int remainingEntries;
-    private final long startNanos;
+    private long startNanos;
+    private boolean started;
     private final long maxDurationNanos;
     private final LongSupplier nanoTime;
     private boolean yielded;
@@ -87,11 +89,15 @@ static final class ScanBudget {
       this.remainingEntries = maxEntries;
       this.maxDurationNanos = maxDurationNanos;
       this.nanoTime = nanoTime;
-      this.startNanos = nanoTime.getAsLong();
     }
 
     private boolean shouldYield() {
-      if (remainingEntries <= 0 || nanoTime.getAsLong() - startNanos >= maxDurationNanos) {
+      final long nowNanos = nanoTime.getAsLong();
+      if (!started) {
+        startNanos = nowNanos;
+        started = true;
+      }
+      if (remainingEntries <= 0 || nowNanos - startNanos >= maxDurationNanos) {
         yielded = true;
       }
       return yielded;
@@ -141,6 +147,7 @@ boolean hasYielded() {
 
   private IndexedConsensusRequest nextReady;
   private boolean scanYielded;
+  private long bodyReadAttempts;
 
   public ProgressWALIterator(final File logDirectory) {
     this(logDirectory, Long.MIN_VALUE);
@@ -279,9 +286,13 @@ void retry(final IndexedConsensusRequest request) {
    */
   public void advanceTo(
       final long targetSearchIndex, final WriterProgressCoverage writerProgressCoverage) {
-    if (targetSearchIndex <= minimumSearchIndex) {
+    if (targetSearchIndex < minimumSearchIndex
+        || (targetSearchIndex == minimumSearchIndex && currentFileIndex >= 0)) {
       return;
     }
+    // A lower bound filters requests; it does not position the reader. A fresh/rebuilt iterator
+    // already has this bound but must still be allowed to skip fully covered historical files.
+    // Once scanning starts, equal-bound probes must not recheck the same uncovered file footers.
     minimumSearchIndex = targetSearchIndex;
 
     if (writerProgressCoverage == null || !canDiscardBufferedRequests(writerProgressCoverage)) {
@@ -447,6 +458,26 @@ boolean isWaitingForReadableWal() {
     return retryCurrentReader;
   }
 
+  @Override
+  public String toString() {
+    final String status =
+        retryCurrentReader
+            ? "WAITING_FOR_READABLE_WAL"
+            : scanYielded
+                ? "SCAN_YIELDED"
+                : hasBufferedNext() ? "REQUEST_BUFFERED" : "NO_BUFFERED_REQUEST";
+    return status
+        + "[fileVersion="
+        + currentReaderVersionId
+        + ", entryOffset="
+        + consumedEntryCountInCurrentFile
+        + ", bodyReadAttempts="
+        + bodyReadAttempts
+        + ", minimumSearchIndex="
+        + minimumSearchIndex
+        + "]";
+  }
+
   String getSkippedBrokenWalFileDetails(final int firstFileIndex) {
     final int firstAvailableIndex =
         getSkippedBrokenWalFileCount() - skippedBrokenWalFileDetails.size();
@@ -515,6 +546,7 @@ private IndexedConsensusRequest advance(final ScanBudget scanBudget) throws IOEx
             // hasNext() can otherwise read an arbitrary WAL prefix before returning a request.
             scanBudget.remainingEntries--;
           }
+          bodyReadAttempts++;
           final ByteBuffer buffer = currentReader.next();
           consumedEntryCountInCurrentFile = currentReader.getCurrentEntryIndex() + 1;
           final WALEntryType type = WALEntryType.valueOf(buffer.get());

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

@@ -35,6 +35,7 @@
 import org.apache.iotdb.consensus.iot.subscription.SubscriptionQueueAdmission;
 import org.apache.iotdb.consensus.iot.subscription.SubscriptionQueueRejectionReason;
 import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.i18n.DataNodePipeMessages;
 import org.apache.iotdb.db.queryengine.plan.statement.StatementTestUtils;
 import org.apache.iotdb.db.storageengine.dataregion.wal.io.ProgressWALReader;
 import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALFileVersion;
@@ -56,13 +57,18 @@
 import org.apache.iotdb.rpc.subscription.payload.poll.WriterId;
 import org.apache.iotdb.rpc.subscription.payload.poll.WriterProgress;
 
+import ch.qos.logback.classic.Level;
+import ch.qos.logback.classic.Logger;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.core.read.ListAppender;
 import org.apache.tsfile.enums.ColumnCategory;
 import org.apache.tsfile.enums.TSDataType;
 import org.apache.tsfile.utils.Pair;
 import org.apache.tsfile.write.record.Tablet;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.TemporaryFolder;
+import org.slf4j.LoggerFactory;
 
 import java.io.File;
 import java.io.IOException;
@@ -112,6 +118,99 @@ public class ConsensusPrefetchingQueueTest {
 
   @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder();
 
+  @Test
+  public void testReplayStallWarningIsThrottledAndResetsAfterProgress() throws Exception {
+    final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir();
+    final Logger logger = (Logger) LoggerFactory.getLogger(ConsensusPrefetchingQueue.class);
+    final Level originalLevel = logger.getLevel();
+    logger.setLevel(Level.WARN);
+    final ListAppender<ILoggingEvent> appender = new ListAppender<>();
+    appender.start();
+    logger.addAppender(appender);
+    ConsensusPrefetchingQueue queue = null;
+    try {
+      final FakeConsensusReqReader reader = new FakeConsensusReqReader();
+      reader.currentSearchIndex = 10L;
+      final IoTConsensusServerImpl server = mock(IoTConsensusServerImpl.class);
+      when(server.getConsensusReqReader()).thenReturn(reader);
+      when(server.getWriterSafeFrontierTracker()).thenReturn(new WriterSafeFrontierTracker());
+      queue =
+          new ConsensusPrefetchingQueue(
+              "consumerGroup",
+              "topic",
+              TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE,
+              new DataRegionId(1),
+              server,
+              new SubscriptionWalRetentionPolicy(
+                  "topic",
+                  SubscriptionWalRetentionPolicy.UNBOUNDED,
+                  SubscriptionWalRetentionPolicy.UNBOUNDED),
+              mock(ConsensusLogToTabletConverter.class),
+              newCommitManager(temporaryFolder.newFolder("replay-stall-warning")),
+              new RegionProgress(Collections.emptyMap()),
+              1L,
+              1L,
+              true);
+      final Field initialized =
+          ConsensusPrefetchingQueue.class.getDeclaredField("prefetchInitialized");
+      initialized.setAccessible(true);
+      initialized.setBoolean(queue, true);
+      final Method report =
+          ConsensusPrefetchingQueue.class.getDeclaredMethod(
+              "reportReplayStallIfNecessary", long.class);
+      report.setAccessible(true);
+      final Supplier<Long> warningCount =
+          () ->
+              appender.list.stream()
+                  .filter(
+                      event ->
+                          event
+                              .getMessage()
+                              .equals(
+                                  DataNodePipeMessages
+                                      .LOG_SUBSCRIPTION_REPLAY_HAS_MADE_NO_CURSOR_PROGRESS_FOR_ARG_MS_ARG_INSPECT_PREFETCH_WORKER_STACKS_AND_WAL_I_O_FAE1F64B))
+                  .count();
+      report.invoke(queue, 0L);
+      report.invoke(queue, TimeUnit.SECONDS.toNanos(59L));
+      assertEquals(Long.valueOf(0L), warningCount.get());
+      report.invoke(queue, TimeUnit.SECONDS.toNanos(60L));
+      report.invoke(queue, TimeUnit.SECONDS.toNanos(89L));
+      assertEquals(Long.valueOf(1L), warningCount.get());
+      report.invoke(queue, TimeUnit.SECONDS.toNanos(90L));
+      assertEquals(Long.valueOf(2L), warningCount.get());
+
+      final Field cursorField =
+          ConsensusPrefetchingQueue.class.getDeclaredField("nextExpectedSearchIndex");
+      cursorField.setAccessible(true);
+      ((AtomicLong) cursorField.get(queue)).incrementAndGet();
+      report.invoke(queue, TimeUnit.SECONDS.toNanos(91L));
+      report.invoke(queue, TimeUnit.SECONDS.toNanos(150L));
+      assertEquals(Long.valueOf(2L), warningCount.get());
+      report.invoke(queue, TimeUnit.SECONDS.toNanos(151L));
+      assertEquals(Long.valueOf(3L), warningCount.get());
+
+      reader.currentSearchIndex = 0L;
+      report.invoke(queue, TimeUnit.SECONDS.toNanos(250L));
+      report.invoke(queue, TimeUnit.SECONDS.toNanos(350L));
+      reader.currentSearchIndex = 10L;
+      // A backlog appearing after an idle period starts a new observation window.
+      report.invoke(queue, TimeUnit.SECONDS.toNanos(450L));
+      report.invoke(queue, TimeUnit.SECONDS.toNanos(509L));
+      assertEquals(Long.valueOf(3L), warningCount.get());
+      queue.setActive(false);
+      report.invoke(queue, TimeUnit.SECONDS.toNanos(600L));
+      assertEquals(Long.valueOf(3L), warningCount.get());
+    } finally {
+      logger.detachAppender(appender);
+      logger.setLevel(originalLevel);
+      appender.stop();
+      if (queue != null) {
+        queue.close();
+      }
+      IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir);
+    }
+  }
+
   @Test
   public void testLateOldPrefetchDeregistrationKeepsReplacementTask() {
     final ConsensusSubscriptionPrefetchExecutor executor =