[Subscription] Fix stalled WAL replay retries across file refresh and rotation

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

@@ -2270,6 +2270,12 @@ private DataNodePipeMessages() {}
       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_METADATA_SNAPSHOT_FOR_FILE_ARG_VERSIONID_ARG_IS_UNAVAILABLE_AFTER_ROTATION_REPLAY_WILL_RETRY_WITHOUT_SKIPPING_D99C12A0 =
+          "WAL metadata snapshot for file %s (versionId=%d) is unavailable after rotation; replay will retry without skipping";
+  public static final String
+      EXCEPTION_WAL_FILE_ARG_VERSIONID_ARG_IS_MISSING_FROM_TWO_SUCCESSIVE_DIRECTORY_SCANS_AFTER_A_NEWER_WAL_FILE_BECAME_VISIBLE_1E170B4B =
+          "WAL file %s (versionId=%d) is missing from two successive directory scans after a newer WAL file became visible";
   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";

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

@@ -2111,6 +2111,12 @@ private DataNodePipeMessages() {}
       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_METADATA_SNAPSHOT_FOR_FILE_ARG_VERSIONID_ARG_IS_UNAVAILABLE_AFTER_ROTATION_REPLAY_WILL_RETRY_WITHOUT_SKIPPING_D99C12A0 =
+          "WAL 文件 %s（versionId=%d）在轮转后无法获取对应的元数据快照；回放将重试，不跳过该文件";
+  public static final String
+      EXCEPTION_WAL_FILE_ARG_VERSIONID_ARG_IS_MISSING_FROM_TWO_SUCCESSIVE_DIRECTORY_SCANS_AFTER_A_NEWER_WAL_FILE_BECAME_VISIBLE_1E170B4B =
+          "后继 WAL 文件可见后，WAL 文件 %s（versionId=%d）在连续两次目录扫描中均不存在";
   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";

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java

@@ -893,6 +893,17 @@ public WALMetaData getCurrentWALMetaDataSnapshot() {
     return writer == null ? new WALMetaData() : writer.snapshotMetaData();
   }
 
+  /** Returns null if rotation has already installed a writer for another file version. */
+  public WALMetaData getWALMetaDataSnapshot(final long fileVersionId) {
+    final WALWriter writer = currentWALFileWriter;
+    // The writer and currentWALFileVersion are published separately during rotation. Bind the
+    // snapshot to the captured writer itself, including while the version field is still old.
+    return writer != null
+            && WALFileUtils.parseVersionId(writer.getLogFile().getName()) == fileVersionId
+        ? writer.snapshotMetaData()
+        : null;
+  }
+
   public CheckpointManager getCheckpointManager() {
     return checkpointManager;
   }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNode.java

@@ -1090,6 +1090,10 @@ public WALMetaData getCurrentWALMetaDataSnapshot() {
     return buffer.getCurrentWALMetaDataSnapshot();
   }
 
+  public WALMetaData getWALMetaDataSnapshot(final long fileVersionId) {
+    return buffer.getWALMetaDataSnapshot(fileVersionId);
+  }
+
   @Override
   public long getTotalSize() {
     return WALManager.getInstance().getTotalDiskUsage();

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

@@ -4736,8 +4736,8 @@ ProgressWALReader openProgressWALReader(final File walFile) throws IOException {
   private static LiveWALMetaDataSnapshot captureLiveWALMetaDataSnapshot(final WALNode walNode) {
     while (true) {
       final long versionId = walNode.getCurrentWALFileVersion();
-      final WALMetaData metadata = walNode.getCurrentWALMetaDataSnapshot();
-      if (versionId == walNode.getCurrentWALFileVersion()) {
+      final WALMetaData metadata = walNode.getWALMetaDataSnapshot(versionId);
+      if (metadata != null && versionId == walNode.getCurrentWALFileVersion()) {
         return new LiveWALMetaDataSnapshot(versionId, metadata);
       }
     }