From 8fe67c4935ec54330e1a33b2256f621b00edbc58 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Sat, 10 Oct 2026 16:00:21 +0800 Subject: [PATCH 1/2] [Subscription] Report retained WAL file gaps and retry transient reads --- .../base/AbstractSubscriptionConsumer.java | 18 +- .../SubscriptionConsumerLifecycleTest.java | 50 ++++ .../iotdb/db/i18n/DataNodePipeMessages.java | 16 ++ .../iotdb/db/i18n/DataNodePipeMessages.java | 16 ++ .../consensus/ConsensusPrefetchingQueue.java | 56 +++- .../broker/consensus/ProgressWALIterator.java | 260 ++++++++++++++---- ...usSubscriptionPrefetchingQueueMetrics.java | 20 ++ .../ConsensusPrefetchingQueueTest.java | 32 +++ .../consensus/ProgressWALIteratorTest.java | 207 +++++++++++++- ...bscriptionPrefetchingQueueMetricsTest.java | 65 +++++ .../commons/service/metric/enums/Metric.java | 2 + 11 files changed, 674 insertions(+), 68 deletions(-) diff --git a/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionConsumer.java b/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionConsumer.java index 4eecc4d028c9..113971ac1222 100644 --- a/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionConsumer.java +++ b/iotdb-client/subscription/src/main/java/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 fencedException = new AtomicReference<>(); + private final AtomicReference 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 multiplePoll( /* @NotNull */ final Set 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 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/iotdb-client/subscription/src/test/java/org/apache/iotdb/session/subscription/consumer/base/SubscriptionConsumerLifecycleTest.java b/iotdb-client/subscription/src/test/java/org/apache/iotdb/session/subscription/consumer/base/SubscriptionConsumerLifecycleTest.java index 75f1ed786503..51f35c5d8205 100644 --- a/iotdb-client/subscription/src/test/java/org/apache/iotdb/session/subscription/consumer/base/SubscriptionConsumerLifecycleTest.java +++ b/iotdb-client/subscription/src/test/java/org/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> failed = new CompletableFuture<>(); + failed.completeExceptionally(gap); + final CompletableFuture> 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> 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/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 index 3cd7ae145858..2d84f18203fa 100644 --- 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 index 23f58a00c51f..2e3c70384a2e 100644 --- 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/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java index b8cd6b258d36..192f94ae4fa8 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java +++ b/iotdb-core/datanode/src/main/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 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 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)); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIterator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIterator.java index 6572473cb58c..cbf0649f2243 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIterator.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIterator.java @@ -23,10 +23,12 @@ import org.apache.iotdb.commons.request.IConsensusRequest; import org.apache.iotdb.consensus.common.request.IndexedConsensusRequest; import org.apache.iotdb.consensus.common.request.IoTConsensusRequest; +import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.i18n.DataNodePipeMessages; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.SearchNode; import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntryType; import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALInfoEntry; +import org.apache.iotdb.db.storageengine.dataregion.wal.exception.BrokenWALFileException; import org.apache.iotdb.db.storageengine.dataregion.wal.io.ProgressWALReader; import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALFileVersion; import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALMetaData; @@ -64,6 +66,7 @@ public interface WriterProgressCoverage { } private static final Logger LOGGER = LoggerFactory.getLogger(ProgressWALIterator.class); + private static final int MAX_SKIPPED_FILE_DETAILS = 8; private static final int SEARCH_INDEX_OFFSET = WALInfoEntry.FIXED_SERIALIZED_SIZE + PlanNodeType.BYTES; @@ -82,10 +85,13 @@ public interface WriterProgressCoverage { private boolean currentReaderUsesLiveSnapshot = false; private int consumedEntryCountInCurrentFile = 0; private final Set skippedBrokenWalVersionIds = new HashSet<>(); - private int unreportedSkippedBrokenWalFileCount = 0; - private String firstUnreportedSkippedBrokenWalFile; - private String lastUnreportedSkippedBrokenWalFile; - private String firstUnreportedSkippedBrokenWalError; + private final List skippedBrokenWalFileDetails = new ArrayList<>(); + private long skippedBrokenWalEntryCount; + private boolean retryCurrentReader; + private long eofRetryVersionId = -1L; + private int eofRetryEntryOffset = -1; + private long eofRetryFileLength = -1L; + private long eofRetryFileModified = -1L; private IOException lastError; private boolean incompleteScan = false; private String incompleteScanDetail; @@ -194,7 +200,6 @@ public boolean hasNext() { DataNodePipeMessages.PIPE_LOG_PROGRESSWALITERATOR_ERROR_READING_WAL_2DB46D41, e); } if (nextReady == null) { - logSkippedBrokenWalFilesIfNecessary(); return false; } } @@ -366,6 +371,23 @@ int getSkippedBrokenWalFileCount() { return skippedBrokenWalVersionIds.size(); } + long getSkippedBrokenWalEntryCount() { + return skippedBrokenWalEntryCount; + } + + boolean isWaitingForReadableWal() { + return retryCurrentReader; + } + + String getSkippedBrokenWalFileDetails(final int firstFileIndex) { + final int firstAvailableIndex = + getSkippedBrokenWalFileCount() - skippedBrokenWalFileDetails.size(); + final int start = Math.max(0, firstFileIndex - firstAvailableIndex); + return (firstFileIndex < firstAvailableIndex ? "...; " : "") + + String.join( + "; ", skippedBrokenWalFileDetails.subList(start, skippedBrokenWalFileDetails.size())); + } + public boolean hasIncompleteScan() { return incompleteScan || hasReadError() || hasSkippedBrokenWalFiles(); } @@ -398,6 +420,21 @@ public void close() throws IOException { private IndexedConsensusRequest advance() throws IOException { while (true) { + if (retryCurrentReader) { + refresh(); + final int retryFileIndex = findFileIndexByVersion(currentReaderVersionId); + if (retryFileIndex < 0) { + // Do not let a temporarily invisible file make replay silently move to its successor. + return null; + } + if (!openReaderAtIndex(retryFileIndex, consumedEntryCountInCurrentFile)) { + if (retryCurrentReader) { + return null; + } + continue; + } + retryCurrentReader = false; + } if (currentReader != null && currentReader.hasNext()) { try { final ByteBuffer buffer = currentReader.next(); @@ -430,25 +467,21 @@ private IndexedConsensusRequest advance() throws IOException { } continue; } catch (final EOFException eofException) { - if (!currentReaderUsesLiveSnapshot) { - final IndexedConsensusRequest flushed = skipUnreadableCurrentWalFile(eofException); - if (flushed != null && !shouldSkip(flushed)) { - return flushed; - } - continue; + if (!isConfirmedRetainedWalTruncation()) { + deferCurrentReader(eofException); + return null; } - // Live snapshot metadata may get ahead of the bytes currently visible in the file. Treat - // EOF as "this snapshot is exhausted for now" instead of terminating the iterator. - final IndexedConsensusRequest flushed = flushPending(); + final IndexedConsensusRequest flushed = skipUnreadableCurrentWalFile(eofException); if (flushed != null && !shouldSkip(flushed)) { - closeCurrentReader(); return flushed; } - if (reopenLiveSnapshotReader()) { - continue; - } - return null; + continue; } catch (final IOException readException) { + if (!(readException instanceof BrokenWALFileException) + && !isCurrentEntryAboveSizeLimit()) { + deferCurrentReader(readException); + return null; + } final IndexedConsensusRequest flushed = skipUnreadableCurrentWalFile(readException); if (flushed != null && !shouldSkip(flushed)) { return flushed; @@ -479,6 +512,9 @@ private IndexedConsensusRequest advance() throws IOException { } if (!openNextReader()) { + if (retryCurrentReader) { + return null; + } final IndexedConsensusRequest flushed = flushPending(); if (flushed != null && !shouldSkip(flushed)) { return flushed; @@ -493,6 +529,9 @@ private boolean openNextReader() throws IOException { if (openReaderAtIndex(currentFileIndex, 0)) { return true; } + if (retryCurrentReader) { + return false; + } } return false; } @@ -522,10 +561,14 @@ private boolean reopenLiveSnapshotReader() throws IOException { if (previousFileIndex < 0) { return openFirstReaderAfterVersion(currentReaderVersionId); } + final long versionToReopen = currentReaderVersionId; if (openReaderAtIndex(previousFileIndex, consumedEntryCountInCurrentFile)) { return true; } - return openFirstReaderAfterVersion(currentReaderVersionId); + if (retryCurrentReader) { + return false; + } + return openFirstReaderAfterVersion(versionToReopen); } private boolean openReaderAtIndex(final int fileIndex, final int skipEntries) throws IOException { @@ -546,11 +589,13 @@ private boolean openReaderAtIndex( throws IOException { final File walFile = walFiles[fileIndex]; final long versionId = walFileVersionIds[fileIndex]; + currentFileIndex = fileIndex; final boolean useLiveSnapshot = liveWalNode != null && versionId == liveWalNode.getCurrentWALFileVersion(); + ProgressWALReader reader = null; try { - final ProgressWALReader reader = + reader = useLiveSnapshot ? new ProgressWALReader( walFile, @@ -562,10 +607,14 @@ private boolean openReaderAtIndex( reader.close(); markIncompleteScan( String.format( - "failed to reopen WAL file %s at entry offset %s: iterator could not skip to the requested position", - walFile.getName(), skipEntries), + DataNodePipeMessages + .MESSAGE_FAILED_TO_REOPEN_WAL_FILE_ARG_AT_ENTRY_OFFSET_ARG_ITERATOR_COULD_NOT_SKIP_TO_THE_REQUESTED_POSITION_332B3AD9, + walFile.getName(), + skipEntries), null); - resetCurrentFileTracking(); + currentReaderVersionId = versionId; + consumedEntryCountInCurrentFile = skipEntries; + retryCurrentReader = true; return false; } currentReader = reader; @@ -573,9 +622,19 @@ private boolean openReaderAtIndex( currentReaderVersionId = versionId; currentReaderUsesLiveSnapshot = useLiveSnapshot; consumedEntryCountInCurrentFile = skipEntries; + retryCurrentReader = false; return true; } catch (final IOException e) { - if (isNearLiveWalVersion(versionId)) { + if (reader != null) { + try { + reader.close(); + } catch (final IOException closeException) { + e.addSuppressed(closeException); + } + } + if (isNearLiveWalVersion(versionId) + || !(e instanceof BrokenWALFileException) + || (liveWalNode == null && fileIndex == walFiles.length - 1)) { LOGGER.debug( DataNodePipeMessages .PIPE_LOG_PROGRESSWALITERATOR_FAILED_TO_OPEN_NEAR_LIVE_WAL_FILE_RETRYING_5AEB94AC, @@ -592,33 +651,123 @@ private boolean openReaderAtIndex( } markIncompleteScan( String.format( - "failed to open near-live WAL file %s while replay scan was still in progress", - walFile.getName()), + DataNodePipeMessages + .MESSAGE_WAL_FILE_ARG_VERSIONID_ARG_ENTRYOFFSET_ARG_IS_TEMPORARILY_UNREADABLE_REPLAY_WILL_RETRY_WITHOUT_SKIPPING_ARG_EA11FBDD, + walFile.getAbsolutePath(), + versionId, + skipEntries, + summarizeException(e)), e); + currentReaderVersionId = versionId; + consumedEntryCountInCurrentFile = skipEntries; + retryCurrentReader = true; return false; } - recordSkippedBrokenWalFile(versionId, walFile, e); + recordSkippedBrokenWalFile(versionId, walFile, e, null, skipEntries); + pendingRequests.clear(); + pendingSearchIndex = Long.MIN_VALUE; + pendingLocalSeq = Long.MIN_VALUE; + resetCurrentFileTracking(); return false; } } private void recordSkippedBrokenWalFile( - final long versionId, final File walFile, final IOException error) { + final long versionId, + final File walFile, + final IOException error, + final WALMetaData metadata, + final int firstSkippedEntryOffset) { if (!skippedBrokenWalVersionIds.add(versionId)) { return; } - if (unreportedSkippedBrokenWalFileCount == 0) { - firstUnreportedSkippedBrokenWalFile = walFile.getName(); - firstUnreportedSkippedBrokenWalError = summarizeException(error); - } - lastUnreportedSkippedBrokenWalFile = walFile.getName(); - unreportedSkippedBrokenWalFileCount++; - LOGGER.debug( + final int entryCount = metadata == null ? -1 : metadata.getBuffersSize().size(); + final long skippedEntries = + entryCount < 0 ? -1L : Math.max(0L, (long) entryCount - firstSkippedEntryOffset); + if (skippedEntries > 0L) { + skippedBrokenWalEntryCount += skippedEntries; + } + // Filename boundaries are file-level search-index bounds, not an entry count. Footer entry + // ordinals also include request fragments and entries from other writers. + final long firstSearchIndex = WALFileUtils.parseStartSearchIndex(walFile.getName()); + final int fileIndex = findFileIndexByVersion(versionId); + final long lastSearchIndex = + fileIndex >= 0 && fileIndex + 1 < walFiles.length + ? WALFileUtils.parseStartSearchIndex(walFiles[fileIndex + 1].getName()) + : -1L; + final String detail = + String.format( + DataNodePipeMessages + .MESSAGE_FILE_ARG_VERSIONID_ARG_FILESEARCHINDEXRANGE_ARG_ARG_ENTRYRANGE_ARG_ARG_SKIPPEDENTRIES_ARG_ERROR_ARG_0D6D77B0, + walFile.getAbsolutePath(), + versionId, + formatKnownValue(firstSearchIndex), + formatKnownValue(lastSearchIndex), + firstSkippedEntryOffset, + formatKnownValue(entryCount), + formatKnownValue(skippedEntries), + summarizeException(error)); + if (skippedBrokenWalFileDetails.size() == MAX_SKIPPED_FILE_DETAILS) { + skippedBrokenWalFileDetails.remove(0); + } + skippedBrokenWalFileDetails.add(detail); + LOGGER.warn( DataNodePipeMessages - .PIPE_LOG_PROGRESSWALITERATOR_FAILED_TO_PROCESS_WAL_FILE_SKIPPING_REMAINING_ENTRIES_093F14A9, - walFile.getName(), + .LOG_PROGRESSWALITERATOR_SKIPPED_UNREADABLE_RETAINED_WAL_FILE_ARG_HISTORICAL_SUBSCRIPTION_DATA_MAY_BE_LOST_AE0DBAB1, + detail, + error); + } + + private static String formatKnownValue(final long value) { + return value < 0 ? DataNodePipeMessages.MESSAGE_UNKNOWN_AD921D60 : String.valueOf(value); + } + + private boolean isConfirmedRetainedWalTruncation() { + final File walFile = walFiles[currentFileIndex]; + final boolean sealed = + liveWalNode != null + ? !isNearLiveWalVersion(currentReaderVersionId) + : currentFileIndex + 1 < walFiles.length; + final boolean confirmed = + !currentReaderUsesLiveSnapshot + && sealed + && eofRetryVersionId == currentReaderVersionId + && eofRetryEntryOffset == consumedEntryCountInCurrentFile + && eofRetryFileLength == walFile.length() + && eofRetryFileModified == walFile.lastModified(); + eofRetryVersionId = currentReaderVersionId; + eofRetryEntryOffset = consumedEntryCountInCurrentFile; + eofRetryFileLength = walFile.length(); + eofRetryFileModified = walFile.lastModified(); + return confirmed; + } + + private void deferCurrentReader(final IOException error) throws IOException { + // Keep the incomplete request and the offset of the first unread entry. Flushing or reopening + // at offset zero here would either deliver a partial request or duplicate its fragments. + markIncompleteScan( + String.format( + DataNodePipeMessages + .MESSAGE_WAL_FILE_ARG_VERSIONID_ARG_ENTRYOFFSET_ARG_IS_TEMPORARILY_UNREADABLE_REPLAY_WILL_RETRY_WITHOUT_SKIPPING_ARG_EA11FBDD, + walFiles[currentFileIndex].getAbsolutePath(), + currentReaderVersionId, + consumedEntryCountInCurrentFile, + summarizeException(error)), error); + retryCurrentReader = true; + closeCurrentReader(); + } + + private boolean isCurrentEntryAboveSizeLimit() { + final int entryIndex = currentReader.getCurrentEntryIndex(); + final List sizes = currentReader.getMetaData().getBuffersSize(); + // An entry above the supported limit is deterministically unreadable, even after reopening. + // Preserve the existing large-entry escape path, but report the resulting data gap explicitly. + return entryIndex >= 0 + && entryIndex < sizes.size() + && sizes.get(entryIndex) + > IoTDBDescriptor.getInstance().getConfig().getWalEntrySizeLimitInByte(); } private IndexedConsensusRequest skipUnreadableCurrentWalFile(final IOException error) { @@ -631,6 +780,13 @@ private IndexedConsensusRequest skipUnreadableCurrentWalFile(final IOException e failedFileIndex >= 0 && failedFileIndex < walFileVersionIds.length ? walFileVersionIds[failedFileIndex] : currentReaderVersionId; + final WALMetaData failedFileMetadata = + currentReader == null ? null : currentReader.getMetaData(); + final int firstSkippedEntryOffset = + Math.max( + 0, + consumedEntryCountInCurrentFile + - (isFailedEntryPartOfPendingRequest() ? pendingRequests.size() : 0)); final IndexedConsensusRequest flushed = isFailedEntryPartOfPendingRequest() ? null : flushPending(); @@ -651,7 +807,8 @@ private IndexedConsensusRequest skipUnreadableCurrentWalFile(final IOException e error); return flushed; } - recordSkippedBrokenWalFile(versionId, walFile, error); + recordSkippedBrokenWalFile( + versionId, walFile, error, failedFileMetadata, firstSkippedEntryOffset); return flushed; } @@ -664,25 +821,6 @@ && isSamePendingRequest( currentReader.getCurrentEntryLocalSeq()); } - 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() @@ -707,6 +845,9 @@ private boolean openFirstReaderAfterVersion(final long versionId) throws IOExcep if (openReaderAtIndex(i, 0)) { return true; } + if (retryCurrentReader) { + return false; + } } resetCurrentFileTracking(); return false; @@ -783,6 +924,9 @@ private void resetCurrentFileTracking() { currentReaderVersionId = -1L; currentReaderUsesLiveSnapshot = false; consumedEntryCountInCurrentFile = 0; + retryCurrentReader = false; + eofRetryVersionId = -1L; + eofRetryEntryOffset = -1; } private void markIncompleteScan(final String detail, final IOException cause) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/metric/ConsensusSubscriptionPrefetchingQueueMetrics.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/metric/ConsensusSubscriptionPrefetchingQueueMetrics.java index 0a549565b06e..359da8f92397 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/metric/ConsensusSubscriptionPrefetchingQueueMetrics.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/metric/ConsensusSubscriptionPrefetchingQueueMetrics.java @@ -116,6 +116,18 @@ private void createAutoGauge(final QueueMetricsKey key) { queue, ConsensusPrefetchingQueue::getWalGapSkippedEntries, key.getTags()); + metricService.createAutoGauge( + Metric.SUBSCRIPTION_CONSENSUS_WAL_FILE_GAP.toString(), + MetricLevel.IMPORTANT, + queue, + ConsensusPrefetchingQueue::getUnreadableWalFileCount, + key.getTags()); + metricService.createAutoGauge( + Metric.SUBSCRIPTION_CONSENSUS_WAL_FILE_GAP_ENTRIES.toString(), + MetricLevel.IMPORTANT, + queue, + ConsensusPrefetchingQueue::getUnreadableWalEntryCount, + key.getTags()); metricService.createAutoGauge( Metric.SUBSCRIPTION_CONSENSUS_ROUTING_EPOCH_CHANGE.toString(), MetricLevel.IMPORTANT, @@ -309,6 +321,14 @@ private void removeAutoGauge(final QueueMetricsKey key) { MetricType.AUTO_GAUGE, Metric.SUBSCRIPTION_CONSENSUS_LAG.toString(), key.getTags()); metricService.remove( MetricType.AUTO_GAUGE, Metric.SUBSCRIPTION_CONSENSUS_WAL_GAP.toString(), key.getTags()); + metricService.remove( + MetricType.AUTO_GAUGE, + Metric.SUBSCRIPTION_CONSENSUS_WAL_FILE_GAP.toString(), + key.getTags()); + metricService.remove( + MetricType.AUTO_GAUGE, + Metric.SUBSCRIPTION_CONSENSUS_WAL_FILE_GAP_ENTRIES.toString(), + key.getTags()); metricService.remove( MetricType.AUTO_GAUGE, Metric.SUBSCRIPTION_CONSENSUS_ROUTING_EPOCH_CHANGE.toString(), diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueTest.java index 2b3dc9f0ef55..766577ac1a23 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueTest.java @@ -229,8 +229,15 @@ public void testUnreadableWalReplayIsVisibleAsPollError() throws Exception { final ProgressWALIterator iterator = mock(ProgressWALIterator.class); when(iterator.getSkippedBrokenWalFileCount()).thenReturn(1); + when(iterator.getSkippedBrokenWalEntryCount()).thenReturn(5L); + when(iterator.getSkippedBrokenWalFileDetails(0)) + .thenReturn("file=_0-0-1.wal, versionId=0, error=EOFException"); setSubscriptionWalIterator(queue, iterator); invokeReportUnreadableWalReplayIfNecessary(queue); + invokeReportUnreadableWalReplayIfNecessary(queue); + assertEquals(1L, queue.getUnreadableWalFileCount()); + assertEquals(5L, queue.getUnreadableWalEntryCount()); + assertEquals(0L, queue.getWalGapSkippedEntries()); final SubscriptionEvent event = queue.poll("consumer"); assertNotNull(event); @@ -238,7 +245,32 @@ public void testUnreadableWalReplayIsVisibleAsPollError() throws Exception { SubscriptionPollResponseType.ERROR.getType(), event.getCurrentResponse().getResponseType()); assertTrue(event.getCurrentResponse().getPayload() instanceof ErrorPayload); + final ErrorPayload error = (ErrorPayload) event.getCurrentResponse().getPayload(); + assertTrue(error.isCritical()); + assertTrue(error.getErrorMessage().contains("_0-0-1.wal")); + assertTrue(error.getErrorMessage().contains("EOFException")); + assertTrue(error.getErrorMessage().contains(queue.getPrefetchingQueueId())); + assertTrue(error.getErrorMessage().contains(queue.getConsensusGroupId().toString())); assertNull(queue.poll("consumer")); + + final Method reset = + ConsensusPrefetchingQueue.class.getDeclaredMethod( + "resetSubscriptionWALPosition", long.class); + reset.setAccessible(true); + reset.invoke(queue, 1L); + assertEquals(1L, queue.getUnreadableWalFileCount()); + assertEquals(5L, queue.getUnreadableWalEntryCount()); + final ProgressWALIterator replacement = mock(ProgressWALIterator.class); + when(replacement.getSkippedBrokenWalFileCount()).thenReturn(2); + // No readable footer: file skips remain observable even when entry counts are unknown. + when(replacement.getSkippedBrokenWalFileDetails(0)) + .thenReturn("file=_1-1-1.wal, skippedEntries=unknown, error=BrokenWALFileException"); + setSubscriptionWalIterator(queue, replacement); + invokeReportUnreadableWalReplayIfNecessary(queue); + assertEquals(3L, queue.getUnreadableWalFileCount()); + assertEquals(5L, queue.getUnreadableWalEntryCount()); + assertTrue( + ((ErrorPayload) queue.poll("consumer").getCurrentResponse().getPayload()).isCritical()); } finally { if (queue != null) { queue.close(); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIteratorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIteratorTest.java index c800a2327e90..8af97b319644 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIteratorTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIteratorTest.java @@ -32,13 +32,20 @@ import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALFileStatus; import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALFileUtils; +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.junit.Assume; import org.junit.Test; +import org.slf4j.LoggerFactory; import java.io.File; import java.nio.ByteBuffer; import java.nio.file.Files; import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; @@ -361,6 +368,13 @@ public void testIteratorAggregatesUnreadableRetainedWalFiles() throws Exception final File lastWal = dir.resolve(WALFileUtils.getLogFileName(2, 2, WALFileStatus.CONTAINS_SEARCH_INDEX)) .toFile(); + final Logger logger = (Logger) LoggerFactory.getLogger(ProgressWALIterator.class); + final Level originalLevel = logger.getLevel(); + logger.setLevel(Level.WARN); + final ListAppender appender = new ListAppender<>(); + appender.setContext(logger.getLoggerContext()); + appender.start(); + logger.addAppender(appender); try { Files.write(firstBrokenWal.toPath(), new byte[128]); @@ -372,6 +386,18 @@ public void testIteratorAggregatesUnreadableRetainedWalFiles() throws Exception try (ProgressWALIterator iterator = new ProgressWALIterator(dir.toFile(), Long.MIN_VALUE)) { assertTrue(iterator.hasNext()); + // A continuously readable iterator must log every skip before reaching exhaustion. + assertEquals(2, appender.list.size()); + assertTrue(appender.list.get(0).getFormattedMessage().contains(firstBrokenWal.getName())); + assertTrue(appender.list.get(1).getFormattedMessage().contains(secondBrokenWal.getName())); + assertTrue(appender.list.get(0).getFormattedMessage().contains("versionId=0")); + assertTrue( + appender.list.get(0).getFormattedMessage().contains("error=BrokenWALFileException")); + assertEquals(0L, iterator.getSkippedBrokenWalEntryCount()); + final String detail = iterator.getSkippedBrokenWalFileDetails(0); + assertTrue(detail.contains(firstBrokenWal.getName())); + assertTrue(detail.contains(secondBrokenWal.getName())); + assertTrue(detail.contains("skippedEntries=unknown")); assertEquals(2L, iterator.next().getSearchIndex()); assertFalse(iterator.hasNext()); assertTrue(iterator.hasSkippedBrokenWalFiles()); @@ -380,6 +406,9 @@ public void testIteratorAggregatesUnreadableRetainedWalFiles() throws Exception assertFalse(iterator.hasReadError()); } } finally { + logger.detachAppender(appender); + logger.setLevel(originalLevel); + appender.stop(); Files.deleteIfExists(firstBrokenWal.toPath()); Files.deleteIfExists(secondBrokenWal.toPath()); Files.deleteIfExists( @@ -421,6 +450,8 @@ public void testIteratorSkipsUnreadableEntryAndPreservesCompletedPendingRequest( assertEquals(3L, iterator.next().getSearchIndex()); assertFalse(iterator.hasNext()); assertEquals(1, iterator.getSkippedBrokenWalFileCount()); + assertEquals(1L, iterator.getSkippedBrokenWalEntryCount()); + assertTrue(iterator.getSkippedBrokenWalFileDetails(0).contains("entryRange=[1, 2)")); assertTrue(iterator.hasIncompleteScan()); } } finally { @@ -459,6 +490,7 @@ public void testIteratorDropsPendingRequestWhenUnreadableEntryContinuesSameWrite assertEquals(3L, iterator.next().getSearchIndex()); assertFalse(iterator.hasNext()); assertEquals(1, iterator.getSkippedBrokenWalFileCount()); + assertEquals(2L, iterator.getSkippedBrokenWalEntryCount()); } } finally { IoTDBDescriptor.getInstance().getConfig().setWalEntrySizeLimitInByte(originalEntrySizeLimit); @@ -468,6 +500,178 @@ public void testIteratorDropsPendingRequestWhenUnreadableEntryContinuesSameWrite } } + @Test + public void testRetainedEofRetriesAndPreservesPendingFragments() throws Exception { + verifyEofRetryPreservesPendingFragments(false); + } + + @Test + public void testNearLiveEofDoesNotBlacklistFileOnRepeatedReads() throws Exception { + verifyEofRetryPreservesPendingFragments(true); + } + + private void verifyEofRetryPreservesPendingFragments(final boolean nearLive) throws Exception { + final Path dir = Files.createTempDirectory("progress-wal-iterator-eof-retry"); + final File dataWal = + dir.resolve(WALFileUtils.getLogFileName(0, 0, WALFileStatus.CONTAINS_SEARCH_INDEX)) + .toFile(); + final File successorWal = + dir.resolve(WALFileUtils.getLogFileName(1, 1, WALFileStatus.CONTAINS_SEARCH_INDEX)) + .toFile(); + try { + writeFragmentedWal(dataWal, 38); + final WALMetaData successorMetadata = singleEntryMeta(19, 2L, 1L, 200L, 7, 2L); + try (WALWriter writer = new WALWriter(successorWal, WALFileVersion.V3)) { + writer.write(searchableEntry(2L), successorMetadata); + } + final WALNode walNode = mock(WALNode.class); + when(walNode.getLogDirectory()).thenReturn(dir.toFile()); + when(walNode.getCurrentWALFileVersion()).thenReturn(1L); + when(walNode.getCurrentWALMetaDataSnapshot()).thenReturn(successorMetadata); + try (ProgressWALIterator iterator = + nearLive + ? new ProgressWALIterator(walNode, 1L) + : new ProgressWALIterator(dir.toFile(), 1L)) { + assertFalse(iterator.hasNext()); + assertTrue(iterator.isWaitingForReadableWal()); + assertEquals(0, iterator.getSkippedBrokenWalFileCount()); + if (nearLive) { + assertFalse(iterator.hasNext()); + assertEquals(0, iterator.getSkippedBrokenWalFileCount()); + } + // Simulate bytes/footer becoming readable after the first EOF. The retry must resume at + // entry offset one, retaining the first fragment exactly once. + Files.delete(dataWal.toPath()); + writeFragmentedWal(dataWal, 19); + assertTrue(iterator.hasNext()); + final IndexedConsensusRequest recovered = iterator.next(); + assertEquals(1L, recovered.getSearchIndex()); + assertEquals(2, recovered.getRequests().size()); + assertTrue(iterator.hasNext()); + assertEquals(2L, iterator.next().getSearchIndex()); + assertFalse(iterator.hasNext()); + assertEquals(0, iterator.getSkippedBrokenWalFileCount()); + } + } finally { + Files.deleteIfExists(dataWal.toPath()); + Files.deleteIfExists(successorWal.toPath()); + Files.deleteIfExists(dir); + } + } + + @Test + public void testRepeatedEofInStableRetainedFileReportsTruncatedRequest() throws Exception { + final Path dir = Files.createTempDirectory("progress-wal-iterator-confirmed-truncation"); + final File dataWal = + dir.resolve(WALFileUtils.getLogFileName(0, 0, WALFileStatus.CONTAINS_SEARCH_INDEX)) + .toFile(); + final File successorWal = + dir.resolve(WALFileUtils.getLogFileName(1, 1, WALFileStatus.CONTAINS_SEARCH_INDEX)) + .toFile(); + try { + writeFragmentedWal(dataWal, 38); + try (WALWriter writer = new WALWriter(successorWal, WALFileVersion.V3)) { + writer.write(searchableEntry(2L), singleEntryMeta(19, 2L, 1L, 200L, 7, 2L)); + } + try (ProgressWALIterator iterator = new ProgressWALIterator(dir.toFile(), 1L)) { + assertFalse(iterator.hasNext()); + assertEquals(0, iterator.getSkippedBrokenWalFileCount()); + assertTrue(iterator.hasNext()); + assertEquals(2L, iterator.next().getSearchIndex()); + assertEquals(1, iterator.getSkippedBrokenWalFileCount()); + assertEquals(2L, iterator.getSkippedBrokenWalEntryCount()); + final String detail = iterator.getSkippedBrokenWalFileDetails(0); + assertTrue(detail.contains(dataWal.getName())); + assertTrue(detail.contains("fileSearchIndexRange=(0, 1]")); + assertTrue(detail.contains("entryRange=[0, 2)")); + assertTrue(detail.contains("error=EOFException")); + } + } finally { + Files.deleteIfExists(dataWal.toPath()); + Files.deleteIfExists(successorWal.toPath()); + Files.deleteIfExists(dir); + } + } + + @Test + public void testLiveSnapshotEofPreservesIncompleteRequestUntilBytesAreVisible() throws Exception { + final Path dir = Files.createTempDirectory("progress-wal-iterator-live-eof-retry"); + final File liveWal = + dir.resolve(WALFileUtils.getLogFileName(0, 0, WALFileStatus.CONTAINS_SEARCH_INDEX)) + .toFile(); + try { + final WALMetaData metadata = singleEntryMeta(19, 1L, 1L, 100L, 7, 1L); + metadata.add(19, 1L, 1L, 100L, 7, 1L); + final WALNode walNode = mock(WALNode.class); + when(walNode.getLogDirectory()).thenReturn(dir.toFile()); + when(walNode.getCurrentWALFileVersion()).thenReturn(0L); + when(walNode.getCurrentWALMetaDataSnapshot()).thenReturn(metadata); + try (WALWriter writer = new WALWriter(liveWal, WALFileVersion.V3)) { + writer.write(searchableEntry(1L), singleEntryMeta(19, 1L, 1L, 100L, 7, 1L)); + try (ProgressWALIterator iterator = new ProgressWALIterator(walNode, 1L)) { + assertFalse(iterator.hasNext()); + assertEquals(0, iterator.getSkippedBrokenWalFileCount()); + writer.write(searchableEntry(1L), singleEntryMeta(19, 1L, 1L, 100L, 7, 1L)); + assertTrue(iterator.hasNext()); + final IndexedConsensusRequest recovered = iterator.next(); + assertEquals(1L, recovered.getSearchIndex()); + assertEquals(2, recovered.getRequests().size()); + assertEquals(0, iterator.getSkippedBrokenWalFileCount()); + } + } + } finally { + Files.deleteIfExists(liveWal.toPath()); + Files.deleteIfExists(dir); + } + } + + private static void writeFragmentedWal(final File walFile, final int secondEntrySize) + throws Exception { + try (WALWriter writer = new WALWriter(walFile, WALFileVersion.V3)) { + writer.write(searchableEntry(1L), singleEntryMeta(19, 1L, 1L, 100L, 7, 1L)); + writer.write(searchableEntry(1L), singleEntryMeta(secondEntrySize, 1L, 1L, 100L, 7, 1L)); + } + } + + @Test + public void testSkippedFileDetailsRemainBoundedDuringContinuousReplay() throws Exception { + final Path dir = Files.createTempDirectory("progress-wal-iterator-bounded-errors"); + final List files = new ArrayList<>(); + try { + for (int version = 0; version < 10; version++) { + final File file = + dir.resolve( + WALFileUtils.getLogFileName( + version, version, WALFileStatus.CONTAINS_SEARCH_INDEX)) + .toFile(); + files.add(file); + Files.write(file.toPath(), new byte[128]); + } + final File successor = + dir.resolve(WALFileUtils.getLogFileName(10, 10, WALFileStatus.CONTAINS_SEARCH_INDEX)) + .toFile(); + files.add(successor); + try (WALWriter writer = new WALWriter(successor, WALFileVersion.V3)) { + writer.write(searchableEntry(11L), singleEntryMeta(19, 11L, 1L, 200L, 7, 11L)); + } + try (ProgressWALIterator iterator = new ProgressWALIterator(dir.toFile(), 1L)) { + assertTrue(iterator.hasNext()); + assertEquals(10, iterator.getSkippedBrokenWalFileCount()); + final String detail = iterator.getSkippedBrokenWalFileDetails(0); + assertEquals(8, detail.split("versionId=").length - 1); + assertFalse(detail.contains(files.get(0).getAbsolutePath())); + assertTrue(detail.contains(files.get(9).getAbsolutePath())); + assertEquals(2, iterator.getSkippedBrokenWalFileDetails(8).split("versionId=").length - 1); + } + } finally { + for (final File file : files) { + Files.deleteIfExists(file.toPath()); + Files.deleteIfExists(file.toPath().resolveSibling(file.getName() + ".broken")); + } + Files.deleteIfExists(dir); + } + } + @Test public void testLiveWalReopenReusesMetadataSnapshot() throws Exception { final Path dir = Files.createTempDirectory("progress-wal-iterator-live-snapshot"); @@ -522,7 +726,8 @@ public void testIteratorMarksIncompleteScanWhenNearLiveWalCannotBeOpened() throw assertFalse(iterator.hasNext()); assertTrue(iterator.hasIncompleteScan()); assertTrue(iterator.hasReadError()); - assertTrue(iterator.getIncompleteScanDetail().contains("near-live WAL file")); + assertTrue(iterator.getIncompleteScanDetail().contains(brokenLiveWal.getName())); + assertEquals(0, iterator.getSkippedBrokenWalFileCount()); } } finally { Files.deleteIfExists(brokenLiveWal.toPath()); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/metric/ConsensusSubscriptionPrefetchingQueueMetricsTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/metric/ConsensusSubscriptionPrefetchingQueueMetricsTest.java index 782cebe30cda..8038d9cb0ada 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/metric/ConsensusSubscriptionPrefetchingQueueMetricsTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/metric/ConsensusSubscriptionPrefetchingQueueMetricsTest.java @@ -30,9 +30,11 @@ import org.apache.iotdb.metrics.utils.MetricType; import org.junit.Test; +import org.mockito.ArgumentCaptor; import java.lang.reflect.Field; import java.util.Map; +import java.util.function.ToDoubleFunction; import static org.junit.Assert.assertEquals; import static org.mockito.ArgumentMatchers.any; @@ -60,6 +62,10 @@ public void testMetricsAreIsolatedByRegion() throws Exception { when(firstQueue.getConsensusGroupId()).thenReturn(firstRegionId); when(secondQueue.getPrefetchingQueueId()).thenReturn(queueId); when(secondQueue.getConsensusGroupId()).thenReturn(secondRegionId); + when(firstQueue.getUnreadableWalFileCount()).thenReturn(4L); + when(firstQueue.getUnreadableWalEntryCount()).thenReturn(7L); + when(secondQueue.getUnreadableWalFileCount()).thenReturn(5L); + when(secondQueue.getUnreadableWalEntryCount()).thenReturn(9L); when(staleFirstQueue.getPrefetchingQueueId()).thenReturn(queueId); when(staleFirstQueue.getConsensusGroupId()).thenReturn(firstRegionId); when(metricService.getOrCreateRate( @@ -103,6 +109,43 @@ public void testMetricsAreIsolatedByRegion() throws Exception { assertEquals(2, queueMap.size()); assertEquals(2, rateMap.size()); + for (final Metric metric : + new Metric[] { + Metric.SUBSCRIPTION_CONSENSUS_WAL_FILE_GAP, + Metric.SUBSCRIPTION_CONSENSUS_WAL_FILE_GAP_ENTRIES + }) { + @SuppressWarnings({"unchecked", "rawtypes"}) + final ArgumentCaptor> mapper = + ArgumentCaptor.forClass((Class) ToDoubleFunction.class); + verify(metricService) + .createAutoGauge( + eq(metric.toString()), + eq(MetricLevel.IMPORTANT), + eq(firstQueue), + mapper.capture(), + eq(Tag.NAME.toString()), + eq(queueId), + eq(Tag.REGION.toString()), + eq(firstRegionId.toString())); + verify(metricService) + .createAutoGauge( + eq(metric.toString()), + eq(MetricLevel.IMPORTANT), + eq(secondQueue), + mapper.capture(), + eq(Tag.NAME.toString()), + eq(queueId), + eq(Tag.REGION.toString()), + eq(secondRegionId.toString())); + assertEquals( + metric == Metric.SUBSCRIPTION_CONSENSUS_WAL_FILE_GAP ? 4.0 : 7.0, + mapper.getAllValues().get(0).applyAsDouble(firstQueue), + 0.0); + assertEquals( + metric == Metric.SUBSCRIPTION_CONSENSUS_WAL_FILE_GAP ? 5.0 : 9.0, + mapper.getAllValues().get(1).applyAsDouble(secondQueue), + 0.0); + } verify(metricService) .createAutoGauge( eq(Metric.SUBSCRIPTION_CONSENSUS_LAG.toString()), @@ -169,6 +212,28 @@ public void testMetricsAreIsolatedByRegion() throws Exception { metrics.deregister(firstQueue); assertEquals(1, queueMap.size()); assertEquals(1, rateMap.size()); + for (final Metric metric : + new Metric[] { + Metric.SUBSCRIPTION_CONSENSUS_WAL_FILE_GAP, + Metric.SUBSCRIPTION_CONSENSUS_WAL_FILE_GAP_ENTRIES + }) { + verify(metricService) + .remove( + MetricType.AUTO_GAUGE, + metric.toString(), + Tag.NAME.toString(), + queueId, + Tag.REGION.toString(), + firstRegionId.toString()); + verify(metricService, never()) + .remove( + MetricType.AUTO_GAUGE, + metric.toString(), + Tag.NAME.toString(), + queueId, + Tag.REGION.toString(), + secondRegionId.toString()); + } verify(metricService) .remove( MetricType.AUTO_GAUGE, diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java index d412d74d0ccb..e20919801e42 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java @@ -216,6 +216,8 @@ public enum Metric { SUBSCRIPTION_EVENT_TRANSFER("subscription_event_transfer"), SUBSCRIPTION_CONSENSUS_LAG("subscription_consensus_lag"), SUBSCRIPTION_CONSENSUS_WAL_GAP("subscription_consensus_wal_gap"), + SUBSCRIPTION_CONSENSUS_WAL_FILE_GAP("subscription_consensus_wal_file_gap"), + SUBSCRIPTION_CONSENSUS_WAL_FILE_GAP_ENTRIES("subscription_consensus_wal_file_gap_entries"), SUBSCRIPTION_CONSENSUS_ROUTING_EPOCH_CHANGE("subscription_consensus_routing_epoch_change"), SUBSCRIPTION_CONSENSUS_WATERMARK("subscription_consensus_watermark"), SUBSCRIPTION_CONSENSUS_SEEK_GENERATION("subscription_consensus_seek_generation"), From 5e834bafccdeb841ce1b6858921f9062c08d1be4 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Sat, 10 Oct 2026 16:36:38 +0800 Subject: [PATCH 2/2] [Subscription] Declare runtime Logback dependencies for WAL log assertions --- iotdb-core/datanode/pom.xml | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/iotdb-core/datanode/pom.xml b/iotdb-core/datanode/pom.xml index 5cd76ae2ac22..48bc0490611d 100644 --- a/iotdb-core/datanode/pom.xml +++ b/iotdb-core/datanode/pom.xml @@ -177,6 +177,16 @@ org.slf4j slf4j-api + + ch.qos.logback + logback-classic + runtime + + + ch.qos.logback + logback-core + runtime + net.java.dev.jna jna @@ -462,6 +472,12 @@ org.apache.iotdb:isession at.yawk.lz4:lz4-java + + + ch.qos.logback:logback-classic + ch.qos.logback:logback-core +