jt2594838 · Oct 10, 2026 · Oct 10, 2026 · Oct 10, 2026
diff --git a/...ava/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionConsumer.java b/...ava/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionConsumer.java
@@ -130,6 +130,8 @@ abstract class AbstractSubscriptionConsumer implements AutoCloseable {
   private final AtomicBoolean isClosed = new AtomicBoolean(true);
   private final AtomicReference<SubscriptionConsumerFencedException> fencedException =
       new AtomicReference<>();
+  private final AtomicReference<SubscriptionRuntimeCriticalException> pendingCriticalPollError =
+      new AtomicReference<>();
   // This variable indicates whether the consumer has ever been closed.
   private final AtomicBoolean isReleased = new AtomicBoolean(false);
 
@@ -779,6 +781,11 @@ public long getLatestWatermarkTimestamp() {
   protected List<SubscriptionMessage> multiplePoll(
       /* @NotNull */ final Set<String> topicNames, final long timeoutMs) {
     checkIfFenced();
+    final SubscriptionRuntimeCriticalException pendingError =
+        pendingCriticalPollError.getAndSet(null);
+    if (pendingError != null) {
+      throw pendingError;
+    }
     if (topicNames.isEmpty()) {
       return Collections.emptyList();
     }
@@ -865,10 +872,13 @@ List<SubscriptionMessage> collectMultiplePollResults(
     // case.
     checkIfFenced();
 
-    // even if a SubscriptionRuntimeCriticalException is encountered, try to deliver the message to
-    // the client
-    if (messages.isEmpty() && Objects.nonNull(lastSubscriptionRuntimeCriticalException)) {
-      throw lastSubscriptionRuntimeCriticalException;
+    if (Objects.nonNull(lastSubscriptionRuntimeCriticalException)) {
+      if (messages.isEmpty()) {
+        throw lastSubscriptionRuntimeCriticalException;
+      }
+      // Deliver healthy sibling results once, then surface the critical error on the next poll.
+      // Otherwise a busy topic can indefinitely mask another topic's permanent WAL replay gap.
+      pendingCriticalPollError.compareAndSet(null, lastSubscriptionRuntimeCriticalException);
     }
 
     return messages;

diff --git a/...rg/apache/iotdb/session/subscription/consumer/base/SubscriptionConsumerLifecycleTest.java b/...rg/apache/iotdb/session/subscription/consumer/base/SubscriptionConsumerLifecycleTest.java
@@ -22,6 +22,7 @@
 import org.apache.iotdb.common.rpc.thrift.TEndPoint;
 import org.apache.iotdb.rpc.subscription.exception.SubscriptionConsumerFencedException;
 import org.apache.iotdb.rpc.subscription.exception.SubscriptionException;
+import org.apache.iotdb.rpc.subscription.exception.SubscriptionRuntimeCriticalException;
 import org.apache.iotdb.rpc.subscription.exception.SubscriptionRuntimeNonCriticalException;
 import org.apache.iotdb.rpc.subscription.payload.poll.SubscriptionCommitContext;
 import org.apache.iotdb.rpc.subscription.payload.poll.SubscriptionPollResponse;
@@ -230,6 +231,55 @@ public void testFencedParallelPollDoesNotDeliverSiblingMessages() throws Excepti
     Assert.assertTrue(consumer.isFenced());
   }
 
+  @Test
+  public void testCriticalParallelPollErrorIsDeliveredAfterHealthySiblingMessages()
+      throws Exception {
+    for (final boolean failedTaskFirst : new boolean[] {false, true}) {
+      final TestPullConsumer consumer = new TestPullConsumer();
+      final SubscriptionRuntimeCriticalException gap =
+          new SubscriptionRuntimeCriticalException("WAL replay gap");
+      final SubscriptionMessage message =
+          new SubscriptionMessage(
+              new SubscriptionCommitContext(0, 0, "healthy_topic", CONSUMER_GROUP_ID, 1L), 1L);
+      final CompletableFuture<List<SubscriptionMessage>> failed = new CompletableFuture<>();
+      failed.completeExceptionally(gap);
+      final CompletableFuture<List<SubscriptionMessage>> healthy =
+          CompletableFuture.completedFuture(Collections.singletonList(message));
+      Assert.assertEquals(
+          Collections.singletonList(message),
+          consumer.collectMultiplePollResults(
+              failedTaskFirst ? Arrays.asList(failed, healthy) : Arrays.asList(healthy, failed),
+              new HashSet<>(Arrays.asList("healthy_topic", "gap_topic"))));
+      try {
+        consumer.multiplePoll(Collections.singleton("healthy_topic"), 100L);
+        Assert.fail("A healthy topic must not hide the pending critical WAL replay error");
+      } catch (final SubscriptionRuntimeCriticalException expected) {
+        Assert.assertSame(gap, expected);
+      }
+      Assert.assertTrue(consumer.multiplePoll(Collections.emptySet(), 100L).isEmpty());
+      Assert.assertFalse(consumer.isFenced());
+    }
+  }
+
+  @Test
+  public void testCriticalParallelPollErrorWithoutMessagesIsDeliveredImmediately()
+      throws Exception {
+    final TestPullConsumer consumer = new TestPullConsumer();
+    final SubscriptionRuntimeCriticalException gap =
+        new SubscriptionRuntimeCriticalException("WAL replay gap");
+    final CompletableFuture<List<SubscriptionMessage>> failed = new CompletableFuture<>();
+    failed.completeExceptionally(gap);
+    try {
+      consumer.collectMultiplePollResults(
+          Arrays.asList(CompletableFuture.completedFuture(Collections.emptyList()), failed),
+          Collections.singleton("gap_topic"));
+      Assert.fail("A critical WAL replay error must fail an otherwise empty poll");
+    } catch (final SubscriptionRuntimeCriticalException expected) {
+      Assert.assertSame(gap, expected);
+    }
+    Assert.assertTrue(consumer.multiplePoll(Collections.emptySet(), 100L).isEmpty());
+  }
+
   @Test
   public void testFencedAsyncCommitFailsBeforeReadingMessages() throws Exception {
     final TestPullConsumer consumer = new TestPullConsumer();

diff --git a/iotdb-core/datanode/pom.xml b/iotdb-core/datanode/pom.xml
@@ -177,6 +177,16 @@
             <groupId>org.slf4j</groupId>
             <artifactId>slf4j-api</artifactId>
         </dependency>
+        <dependency>
+            <groupId>ch.qos.logback</groupId>
+            <artifactId>logback-classic</artifactId>
+            <scope>runtime</scope>
+        </dependency>
+        <dependency>
+            <groupId>ch.qos.logback</groupId>
+            <artifactId>logback-core</artifactId>
+            <scope>runtime</scope>
+        </dependency>
         <dependency>
             <groupId>net.java.dev.jna</groupId>
             <artifactId>jna</artifactId>
@@ -462,6 +472,12 @@
                         <ignoredDependency>org.apache.iotdb:isession</ignoredDependency>
                         <ignoredDependency>at.yawk.lz4:lz4-java</ignoredDependency>
                     </ignoredDependencies>
+                    <ignoredNonTestScopedDependencies>
+                        <!-- Logback supplies the server's runtime SLF4J backend and LogbackMetrics.
+                             Direct references in this module also support test log assertions. -->
+                        <ignoredNonTestScopedDependency>ch.qos.logback:logback-classic</ignoredNonTestScopedDependency>
+                        <ignoredNonTestScopedDependency>ch.qos.logback:logback-core</ignoredNonTestScopedDependency>
+                    </ignoredNonTestScopedDependencies>
                 </configuration>
             </plugin>
             <plugin>

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
@@ -2216,6 +2216,22 @@ private DataNodePipeMessages() {}
       "ProgressWALIterator: skipped {} unreadable retained WAL files in directory {}, "
           + "firstFile={}, lastFile={}, firstError={}; historical subscription data in these "
           + "files cannot be replayed";
+  public static final String
+      LOG_PROGRESSWALITERATOR_SKIPPED_UNREADABLE_RETAINED_WAL_FILE_ARG_HISTORICAL_SUBSCRIPTION_DATA_MAY_BE_LOST_AE0DBAB1 =
+          "ProgressWALIterator: skipped unreadable retained WAL file {}; historical subscription data may be lost";
+  public static final String
+      MESSAGE_FILE_ARG_VERSIONID_ARG_FILESEARCHINDEXRANGE_ARG_ARG_ENTRYRANGE_ARG_ARG_SKIPPEDENTRIES_ARG_ERROR_ARG_0D6D77B0 =
+          "file=%s, versionId=%d, fileSearchIndexRange=(%s, %s], entryRange=[%d, %s), skippedEntries=%s, error=%s";
+  public static final String
+      MESSAGE_SUBSCRIPTION_WAL_REPLAY_SKIPPED_ARG_UNREADABLE_RETAINED_WAL_FILE_S_IN_QUEUE_ARG_REGION_ARG_HISTORICAL_DATA_MAY_BE_LOST_DETAILS_RECENT_FILES_FULL_DETAILS_IN_SERVER_LOG_ARG_FEDEFF7E =
+          "Subscription WAL replay skipped %d unreadable retained WAL file(s) in queue %s, region %s. Historical data may be lost. Details (recent files; full details in server log): %s";
+  public static final String MESSAGE_UNKNOWN_AD921D60 = "unknown";
+  public static final String
+      MESSAGE_WAL_FILE_ARG_VERSIONID_ARG_ENTRYOFFSET_ARG_IS_TEMPORARILY_UNREADABLE_REPLAY_WILL_RETRY_WITHOUT_SKIPPING_ARG_EA11FBDD =
+          "WAL file %s (versionId=%d, entryOffset=%d) is temporarily unreadable; replay will retry without skipping: %s";
+  public static final String
+      MESSAGE_FAILED_TO_REOPEN_WAL_FILE_ARG_AT_ENTRY_OFFSET_ARG_ITERATOR_COULD_NOT_SKIP_TO_THE_REQUESTED_POSITION_332B3AD9 =
+          "failed to reopen WAL file %s at entry offset %s: iterator could not skip to the requested position";
   public static final String
       LOG_ARG_WAL_GAP_ARG_ARG_SKIPPED_ARG_TOTAL_SKIPPED_ARG_CAUSE_UNDETERMINED_D4E6491A =
           "{}: WAL gap [{}, {}), skipped {}, total skipped {}; cause undetermined";

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
@@ -2057,6 +2057,22 @@ private DataNodePipeMessages() {}
   public static final String PIPE_LOG_PROGRESSWALITERATOR_SKIPPED_UNREADABLE_RETAINED_WAL_FILES_FFC8455E =
       "ProgressWALIterator：跳过了 {} 个无法读取的保留 WAL 文件，directory={}，firstFile={}，"
           + "lastFile={}，firstError={}；这些文件中的历史订阅数据无法重放";
+  public static final String
+      LOG_PROGRESSWALITERATOR_SKIPPED_UNREADABLE_RETAINED_WAL_FILE_ARG_HISTORICAL_SUBSCRIPTION_DATA_MAY_BE_LOST_AE0DBAB1 =
+          "ProgressWALIterator：跳过了无法读取的保留 WAL 文件 {}；历史订阅数据可能丢失";
+  public static final String
+      MESSAGE_FILE_ARG_VERSIONID_ARG_FILESEARCHINDEXRANGE_ARG_ARG_ENTRYRANGE_ARG_ARG_SKIPPEDENTRIES_ARG_ERROR_ARG_0D6D77B0 =
+          "file=%s，versionId=%d，fileSearchIndexRange=(%s, %s]，entryRange=[%d, %s)，skippedEntries=%s，error=%s";
+  public static final String
+      MESSAGE_SUBSCRIPTION_WAL_REPLAY_SKIPPED_ARG_UNREADABLE_RETAINED_WAL_FILE_S_IN_QUEUE_ARG_REGION_ARG_HISTORICAL_DATA_MAY_BE_LOST_DETAILS_RECENT_FILES_FULL_DETAILS_IN_SERVER_LOG_ARG_FEDEFF7E =
+          "订阅 WAL 重放累计跳过了 %d 个无法读取的保留 WAL 文件，队列 %s，region %s。历史数据可能丢失。详情（近期文件；完整信息见服务端日志）：%s";
+  public static final String MESSAGE_UNKNOWN_AD921D60 = "未知";
+  public static final String
+      MESSAGE_WAL_FILE_ARG_VERSIONID_ARG_ENTRYOFFSET_ARG_IS_TEMPORARILY_UNREADABLE_REPLAY_WILL_RETRY_WITHOUT_SKIPPING_ARG_EA11FBDD =
+          "WAL 文件 %s（versionId=%d，entryOffset=%d）暂不可读；重放将重试而不跳过：%s";
+  public static final String
+      MESSAGE_FAILED_TO_REOPEN_WAL_FILE_ARG_AT_ENTRY_OFFSET_ARG_ITERATOR_COULD_NOT_SKIP_TO_THE_REQUESTED_POSITION_332B3AD9 =
+          "重新打开 WAL 文件 %s（条目偏移 %s）失败：迭代器无法跳到请求的位置";
   public static final String
       LOG_ARG_WAL_GAP_ARG_ARG_SKIPPED_ARG_TOTAL_SKIPPED_ARG_CAUSE_UNDETERMINED_D4E6491A =
           "{}：WAL 缺口 [{}, {})，跳过 {} 条，累计跳过 {} 条；原因尚未确定";

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
@@ -213,10 +213,16 @@ public class ConsensusPrefetchingQueue {
 
   private final AtomicLong walGapSkippedEntries = new AtomicLong(0);
 
+  private final AtomicLong unreadableWalFileCount = new AtomicLong(0);
+
+  private final AtomicLong unreadableWalEntryCount = new AtomicLong(0);
+
   private final AtomicReference<String> pendingWalReplayError = new AtomicReference<>();
 
   private int reportedSkippedBrokenWalFileCount = 0;
 
+  private long reportedSkippedBrokenWalEntryCount = 0L;
+
   /**
    * Guards queue state transitions that touch replay positioning, seek state, and writer buffers.
    */
@@ -849,7 +855,7 @@ public SubscriptionEvent poll(final String consumerId, final RegionProgress regi
       }
       final String walReplayError = pendingWalReplayError.getAndSet(null);
       if (Objects.nonNull(walReplayError)) {
-        return generateErrorResponse(walReplayError);
+        return generateErrorResponse(walReplayError, true);
       }
       if (!prefetchInitialized && !initPrefetch(regionProgress)) {
         return null;
@@ -2135,12 +2141,24 @@ private void reportUnreadableWalReplayIfNecessary() {
     if (skippedBrokenWalFileCount <= reportedSkippedBrokenWalFileCount) {
       return;
     }
+    final long skippedBrokenWalEntryCount = subscriptionWALIterator.getSkippedBrokenWalEntryCount();
+    final long totalSkippedFiles =
+        unreadableWalFileCount.addAndGet(
+            skippedBrokenWalFileCount - reportedSkippedBrokenWalFileCount);
+    unreadableWalEntryCount.addAndGet(
+        skippedBrokenWalEntryCount - reportedSkippedBrokenWalEntryCount);
+    final String detail =
+        subscriptionWALIterator.getSkippedBrokenWalFileDetails(reportedSkippedBrokenWalFileCount);
     reportedSkippedBrokenWalFileCount = skippedBrokenWalFileCount;
+    reportedSkippedBrokenWalEntryCount = skippedBrokenWalEntryCount;
     pendingWalReplayError.set(
         String.format(
             DataNodePipeMessages
-                .SUBSCRIPTION_ERROR_WAL_REPLAY_SKIPPED_UNREADABLE_RETAINED_WAL_FILES_D60C5FB2,
-            skippedBrokenWalFileCount));
+                .MESSAGE_SUBSCRIPTION_WAL_REPLAY_SKIPPED_ARG_UNREADABLE_RETAINED_WAL_FILE_S_IN_QUEUE_ARG_REGION_ARG_HISTORICAL_DATA_MAY_BE_LOST_DETAILS_RECENT_FILES_FULL_DETAILS_IN_SERVER_LOG_ARG_FEDEFF7E,
+            totalSkippedFiles,
+            getPrefetchingQueueId(),
+            consensusGroupId,
+            detail));
   }
 
   private void advanceWalReplayCursorIfPresent(final IndexedConsensusRequest request) {
@@ -2183,6 +2201,10 @@ private void ensureSubscriptionWalReadable() {
       return;
     }
     reportUnreadableWalReplayIfNecessary();
+    if (subscriptionWALIterator.isWaitingForReadableWal()) {
+      // A retry must preserve the incomplete writer request and its file/entry offset.
+      return;
+    }
     if (!(consensusReqReader instanceof WALNode)) {
       return;
     }
@@ -2205,6 +2227,7 @@ private void ensureSubscriptionWalReadable() {
   private void resetSubscriptionWALPosition(final long startSearchIndex) {
     closeSubscriptionWALIterator();
     reportedSkippedBrokenWalFileCount = 0;
+    reportedSkippedBrokenWalEntryCount = 0L;
     subscriptionWALIterator = createSubscriptionWALIterator(startSearchIndex);
   }
 
@@ -2223,7 +2246,12 @@ private boolean hasReadableWalEntries() {
       // applies the pending realignment. Returning true keeps the worker scheduled for that round.
       return true;
     }
-    return Objects.nonNull(subscriptionWALIterator) && subscriptionWALIterator.hasNext();
+    if (Objects.isNull(subscriptionWALIterator)) {
+      return false;
+    }
+    final boolean readable = subscriptionWALIterator.hasNext();
+    reportUnreadableWalReplayIfNecessary();
+    return readable;
   }
 
   private void requestSubscriptionWalReset(
@@ -2257,6 +2285,7 @@ private void closeSubscriptionWALIterator() {
     if (Objects.isNull(subscriptionWALIterator)) {
       return;
     }
+    reportUnreadableWalReplayIfNecessary();
     try {
       subscriptionWALIterator.close();
     } catch (final IOException e) {
@@ -4096,9 +4125,14 @@ private void flushLingeringBatchOnCloseUnderWriteLock() {
   }
 
   private SubscriptionEvent generateErrorResponse(final String errorMessage) {
+    return generateErrorResponse(errorMessage, false);
+  }
+
+  private SubscriptionEvent generateErrorResponse(
+      final String errorMessage, final boolean critical) {
     return new SubscriptionEvent(
         SubscriptionPollResponseType.ERROR.getType(),
-        new ErrorPayload(errorMessage, false),
+        new ErrorPayload(errorMessage, critical),
         createNonCommittableContext(IoTDBDescriptor.getInstance().getConfig().getDataNodeId()));
   }
 
@@ -4155,6 +4189,16 @@ public long getWalGapSkippedEntries() {
     return walGapSkippedEntries.get();
   }
 
+  /** Cumulative file-skip occurrences, including replay after an iterator reset or seek. */
+  public long getUnreadableWalFileCount() {
+    return unreadableWalFileCount.get();
+  }
+
+  /** Known skipped WAL entries/fragments; unreadable footers leave their entry counts unknown. */
+  public long getUnreadableWalEntryCount() {
+    return unreadableWalEntryCount.get();
+  }
+
   public long getEpochChangeCount() {
     return runtimeVersionChangeCount.get();
   }
@@ -4659,6 +4703,8 @@ public Map<String, String> coreReportMessage() {
     result.put("walPathAcceptedEntries", String.valueOf(getWalPathAcceptedEntries()));
     result.put("seekGeneration", String.valueOf(seekGeneration.get()));
     result.put("walGapSkippedEntries", String.valueOf(walGapSkippedEntries.get()));
+    result.put("unreadableWalFileCount", String.valueOf(getUnreadableWalFileCount()));
+    result.put("unreadableWalEntryCount", String.valueOf(getUnreadableWalEntryCount()));
     result.put("bufferedRealtimeEntryCount", String.valueOf(getRealtimeBufferedEntryCount()));
     result.put("lag", String.valueOf(getLag()));
     result.put("isClosed", String.valueOf(isClosed));