From 9a74f69c7998a490ebc3366afb734f460b470600 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 9 Oct 2026 11:57:16 +0800 Subject: [PATCH 1/8] Fix subscription WAL replay gaps across seek --- .../iotdb/db/i18n/DataNodePipeMessages.java | 11 +- .../iotdb/db/i18n/DataNodePipeMessages.java | 9 +- .../consensus/ConsensusPrefetchingQueue.java | 153 ++++++- .../ConsensusPrefetchingQueueTest.java | 387 +++++++++++++++++- 4 files changed, 529 insertions(+), 31 deletions(-) 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 1c0362b627b3a..95583ded3a923 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,10 +2216,13 @@ 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 PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_WAL_REPLAY_SKIPPED_UNAVAILABLE_SEARCH_INDEXES_B8023B64 = - "ConsensusPrefetchingQueue {}: WAL replay skipped unavailable search indexes [{}, {}), " - + "skippedEntries={}, totalWalGapSkippedEntries={}; the missing WAL data may have been " - + "reclaimed before subscription consumption"; + public static final String PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_WAL_REPLAY_FOUND_UNAVAILABLE_SEARCH_INDEXES_E0CBFFFA = + "ConsensusPrefetchingQueue {}: WAL replay found unavailable search indexes [{}, {}) before " + + "searchIndex {}; forcing WAL refresh before retry"; + public static final String MESSAGE_CONSENSUSPREFETCHINGQUEUE_WAL_REPLAY_CANNOT_RECOVER_SEARCH_INDEXES_70781B22 = + "ConsensusPrefetchingQueue %s: WAL replay cannot recover search indexes [%s, %s), " + + "unavailableEntries=%s, totalWalGapSkippedEntries=%s; subscription delivery is " + + "stalled to prevent silent data loss"; public static final String PIPE_LOG_PIPE_TERMINATE_EVENT_COMMITTED_FOR_HISTORICAL_TRANSFER_CREATIONTIME_9B807B28 = "Pipe {}@{}: terminate event committed for historical transfer. creationTime: {}, " + "shouldMark: {}. {}"; diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java index 49096659ade1b..ccb9cd42da20d 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,9 +2057,12 @@ 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 PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_WAL_REPLAY_SKIPPED_UNAVAILABLE_SEARCH_INDEXES_B8023B64 = - "ConsensusPrefetchingQueue {}：WAL 重放跳过了不可用的 searchIndex 区间 [{}, {})，" - + "skippedEntries={}，totalWalGapSkippedEntries={}；缺失的 WAL 数据可能已在订阅消费前被回收"; + public static final String PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_WAL_REPLAY_FOUND_UNAVAILABLE_SEARCH_INDEXES_E0CBFFFA = + "ConsensusPrefetchingQueue {}：WAL 回放发现不可用 searchIndex 区间 [{}, {})，下一个可见 " + + "searchIndex={}；正在强制刷新 WAL 后重试"; + public static final String MESSAGE_CONSENSUSPREFETCHINGQUEUE_WAL_REPLAY_CANNOT_RECOVER_SEARCH_INDEXES_70781B22 = + "ConsensusPrefetchingQueue %s：WAL 回放无法恢复 searchIndex 区间 [%s, %s)，不可用条目数=%s，" + + "walGapSkippedEntries 总数=%s；为防止静默丢数，订阅投递已停滞"; public static final String PIPE_LOG_PIPE_TERMINATE_EVENT_COMMITTED_FOR_HISTORICAL_TRANSFER_CREATIONTIME_9B807B28 = "Pipe {}@{}：历史传输的终止事件已提交。creationTime：{}，shouldMark：{}。{}"; public static final String PIPE_LOG_PIPE_HISTORICAL_SOURCE_HAS_SUPPLIED_ALL_EVENTS_EMITTING_8B58DE19 = diff --git a/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 b692ec94ed046..be194c986ba7d 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 @@ -297,6 +297,12 @@ public class ConsensusPrefetchingQueue { private volatile long lastWalGapWaitLogTimeMs = 0L; + /** Search index whose WAL visibility gap is being retried after forcing a WAL refresh. */ + private volatile long walGapRetryExpectedSearchIndex = Long.MIN_VALUE; + + /** Critical replay failure returned to consumers instead of silently advancing past data. */ + private volatile String walReplayFailureMessage; + /** Fallback committed region progress from local persisted state. */ private final RegionProgress fallbackCommittedRegionProgress; @@ -771,6 +777,9 @@ public SubscriptionEvent poll(final String consumerId, final RegionProgress regi if (pendingSeekRequest != null) { return null; } + if (Objects.nonNull(walReplayFailureMessage)) { + return generateCriticalErrorResponse(walReplayFailureMessage); + } final SubscriptionEvent event = pollInternal(consumerId); if (Objects.nonNull(event) && prefetchingQueue.size() < MAX_PREFETCHING_QUEUE_SIZE) { requestPrefetch(); @@ -837,6 +846,7 @@ private boolean initPrefetchUnderInitializationLock(final RegionProgress regionP // readers without WAL support. this.subscriptionWALIterator = createSubscriptionWALIterator(resolvedStart.getStartSearchIndex()); + resetWalReplayFailureState(); this.prefetchInitialized = true; this.observedSeekGeneration = seekGeneration.get(); discardBatch(this.lingerBatch); @@ -882,7 +892,9 @@ private ReplayLocateDecision resolveInitReplayStartDecision( case AT_END: final long resolvedReplayStartSearchIndex = replayTarget.getStatus() == ReplayLocateStatus.AT_END - ? Math.min(replayTarget.getStartSearchIndex(), fallbackTailSearchIndex) + ? Math.min( + nextSearchIndexAfter(replayTarget.getStartSearchIndex()), + fallbackTailSearchIndex) : replayTarget.getStartSearchIndex(); return new ReplayLocateDecision( replayTarget.getStatus(), @@ -1427,6 +1439,11 @@ public PrefetchRoundResult drivePrefetchOnce() { applyPendingSubscriptionWalReset(observedSeekGeneration); recycleInFlightEvents(); + if (Objects.nonNull(walReplayFailureMessage)) { + blockRealtimeAdmission(); + return PrefetchRoundResult.dormant(); + } + if (!isActive) { blockRealtimeAdmission(); return computeIdleRoundResult(); @@ -1515,6 +1532,11 @@ public PrefetchRoundResult drivePrefetchOnce() { if (batch.isEmpty() && lingerBatch.isEmpty()) { final MaterializationResult walResult = tryCatchUpFromWAL(observedSeekGeneration); + if (walResult == MaterializationResult.WAL_GAP) { + return Objects.nonNull(walReplayFailureMessage) + ? PrefetchRoundResult.dormant() + : PrefetchRoundResult.rescheduleAfter(WAL_GAP_RETRY_SLEEP_MS); + } if (walResult == MaterializationResult.MEMORY_BLOCKED) { blockRealtimeAdmission(); return PrefetchRoundResult.rescheduleAfter(MEMORY_RETRY_SLEEP_MS); @@ -1930,7 +1952,9 @@ private MaterializationResult tryCatchUpFromWAL(final long expectedSeekGeneratio pumpFromSubscriptionWAL( batchState, expectedSeekGeneration, maxWalEntries, maxTablets, maxBatchBytes); if (materializationResult != MaterializationResult.SUCCESS) { - if (materializationResult == MaterializationResult.MEMORY_BLOCKED && !batchState.isEmpty()) { + if ((materializationResult == MaterializationResult.MEMORY_BLOCKED + || materializationResult == MaterializationResult.WAL_GAP) + && !batchState.isEmpty()) { if (!flushBatch(batchState, expectedSeekGeneration)) { discardBatch(batchState); return MaterializationResult.STALE; @@ -1973,6 +1997,11 @@ private MaterializationResult pumpFromSubscriptionWAL( if (isBeforeLocalCursor(walEntry)) { continue; } + final MaterializationResult continuityResult = + validateWalReplayContinuity(walEntry, expectedSeekGeneration); + if (continuityResult != MaterializationResult.SUCCESS) { + return continuityResult; + } if (shouldSkipForRecoveryProgress(walEntry)) { advanceWalReplayCursorIfPresent(walEntry); continue; @@ -2035,22 +2064,80 @@ private void advanceWalReplayCursorIfPresent(final IndexedConsensusRequest reque if (!hasLocalSearchIndex(request)) { return; } + nextExpectedSearchIndex.set(request.getSearchIndex() + 1); + } + + private MaterializationResult validateWalReplayContinuity( + final IndexedConsensusRequest request, final long expectedSeekGeneration) { + if (seekGeneration.get() != expectedSeekGeneration || isClosed) { + return MaterializationResult.STALE; + } + if (!hasLocalSearchIndex(request)) { + return MaterializationResult.SUCCESS; + } final long actualSearchIndex = request.getSearchIndex(); final long expectedSearchIndex = nextExpectedSearchIndex.get(); - if (actualSearchIndex > expectedSearchIndex) { - final long skippedEntries = actualSearchIndex - expectedSearchIndex; - final long totalSkippedEntries = walGapSkippedEntries.addAndGet(skippedEntries); + if (actualSearchIndex <= expectedSearchIndex) { + if (actualSearchIndex == expectedSearchIndex + && walGapRetryExpectedSearchIndex == expectedSearchIndex) { + walGapRetryExpectedSearchIndex = Long.MIN_VALUE; + pendingWalGapRetryRequested = false; + } + return MaterializationResult.SUCCESS; + } + + if (walGapRetryExpectedSearchIndex != expectedSearchIndex) { + if (seekGeneration.get() != expectedSeekGeneration || isClosed) { + return MaterializationResult.STALE; + } + walGapRetryExpectedSearchIndex = expectedSearchIndex; LOGGER.warn( DataNodePipeMessages - .PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_WAL_REPLAY_SKIPPED_UNAVAILABLE_SEARCH_INDEXES_B8023B64, + .PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_WAL_REPLAY_FOUND_UNAVAILABLE_SEARCH_INDEXES_E0CBFFFA, this, expectedSearchIndex, actualSearchIndex, - skippedEntries, - totalSkippedEntries); + actualSearchIndex); + if (consensusReqReader instanceof WALNode) { + ((WALNode) consensusReqReader).rollWALFile(); + } + if (seekGeneration.get() != expectedSeekGeneration || isClosed) { + return MaterializationResult.STALE; + } + resetSubscriptionWALPosition(expectedSearchIndex); + onWalGapRetryScheduled(); + pendingWalGapRetryRequested = true; + return MaterializationResult.WAL_GAP; + } + + if (seekGeneration.get() != expectedSeekGeneration || isClosed) { + return MaterializationResult.STALE; + } + if (Objects.isNull(walReplayFailureMessage)) { + final long unavailableEntries = actualSearchIndex - expectedSearchIndex; + final long totalUnavailableEntries = walGapSkippedEntries.addAndGet(unavailableEntries); + walReplayFailureMessage = + String.format( + DataNodePipeMessages + .MESSAGE_CONSENSUSPREFETCHINGQUEUE_WAL_REPLAY_CANNOT_RECOVER_SEARCH_INDEXES_70781B22, + this, + expectedSearchIndex, + actualSearchIndex, + unavailableEntries, + totalUnavailableEntries); + LOGGER.error(walReplayFailureMessage); + blockRealtimeAdmission(); } - nextExpectedSearchIndex.set(actualSearchIndex + 1); + return MaterializationResult.WAL_GAP; + } + + private void resetWalReplayFailureState() { + pendingWalGapRetryRequested = false; + walGapWaitStartTimeMs = 0L; + lastWalGapWaitLogTimeMs = 0L; + walGapRetryExpectedSearchIndex = Long.MIN_VALUE; + walReplayFailureMessage = null; } private void ensureSubscriptionWalReadable() { @@ -3053,9 +3140,7 @@ public void cleanUp() { reconcileRetainedTabletMemoryAfterCleanup(); memoryBlockedEntryBytes = -1L; resetBatchWriterProgress(); - pendingWalGapRetryRequested = false; - walGapWaitStartTimeMs = 0L; - lastWalGapWaitLogTimeMs = 0L; + resetWalReplayFailureState(); pendingSubscriptionWalResetSearchIndex = Long.MIN_VALUE; pendingSubscriptionWalResetGeneration = Long.MIN_VALUE; closeSubscriptionWALIterator(); @@ -3072,15 +3157,28 @@ public void cleanUp() { * old files have been reclaimed, the earliest available position may be later than 0. */ public void seekToBeginning() { - seekToResolvedPosition(0L, new RegionProgress(Collections.emptyMap()), "beginning"); + final RegionProgress beginningProgress = new RegionProgress(Collections.emptyMap()); + if (consensusReqReader instanceof WALNode) { + // The first retained WAL index can be greater than one after older files are reclaimed. + seekToRegionProgress(beginningProgress); + } else { + seekToResolvedPosition(FIRST_CONSENSUS_SEARCH_INDEX, beginningProgress, "beginning"); + } } /** * Seeks to the current WAL write position. After this, only newly written data will be consumed. */ public void seekToEnd() { + final RegionProgress tailProgress = computeTailRegionProgress(); seekToResolvedPosition( - consensusReqReader.getCurrentSearchIndex(), computeTailRegionProgress(), "end"); + nextSearchIndexAfter(consensusReqReader.getCurrentSearchIndex()), tailProgress, "end"); + } + + private static long nextSearchIndexAfter(final long searchIndex) { + return searchIndex == Long.MAX_VALUE + ? Long.MAX_VALUE + : Math.max(FIRST_CONSENSUS_SEARCH_INDEX, searchIndex + 1L); } public void seekToRegionProgress(final RegionProgress regionProgress) { @@ -3108,7 +3206,9 @@ public void seekToRegionProgress(final RegionProgress regionProgress) { replayTarget.getStatus(), replayTarget.getStartSearchIndex()); seekToResolvedPosition( - replayTarget.getStartSearchIndex(), + replayTarget.getStatus() == ReplayLocateStatus.AT_END + ? nextSearchIndexAfter(replayTarget.getStartSearchIndex()) + : replayTarget.getStartSearchIndex(), replayTarget.getRecoveryRegionProgress(), "regionProgress"); return; @@ -3149,7 +3249,9 @@ public void seekAfterRegionProgress(final RegionProgress regionProgress) { replayTarget.getStatus(), replayTarget.getStartSearchIndex()); seekToResolvedPosition( - replayTarget.getStartSearchIndex(), + replayTarget.getStatus() == ReplayLocateStatus.AT_END + ? nextSearchIndexAfter(replayTarget.getStartSearchIndex()) + : replayTarget.getStartSearchIndex(), replayTarget.getRecoveryRegionProgress(), "regionProgressAfter"); return; @@ -3349,9 +3451,7 @@ private void applySeekResetUnderWriteLock(final PendingSeekRequest request) { reconcileRetainedTabletMemoryAfterCleanup(); resetBatchWriterProgress(); observedSeekGeneration = seekGeneration.get(); - pendingWalGapRetryRequested = false; - walGapWaitStartTimeMs = 0L; - lastWalGapWaitLogTimeMs = 0L; + resetWalReplayFailureState(); // 5. Reset commit state to the writer progress immediately before the first re-delivered // entry so seek/rebind resumes from the intended frontier. @@ -3452,6 +3552,7 @@ static long findReplayRetainedMinVersionId( return 0L; } + WALFileUtils.ascSortByVersionId(walFiles); final int replayFileIndex = Math.max(0, WALFileUtils.binarySearchFileBySearchIndex(walFiles, nextExpectedSearchIndex)); return WALFileUtils.parseVersionId(walFiles[replayFileIndex].getName()); @@ -3964,6 +4065,13 @@ private SubscriptionEvent generateErrorResponse(final String errorMessage) { createNonCommittableContext(IoTDBDescriptor.getInstance().getConfig().getDataNodeId())); } + private SubscriptionEvent generateCriticalErrorResponse(final String errorMessage) { + return new SubscriptionEvent( + SubscriptionPollResponseType.ERROR.getType(), + new ErrorPayload(errorMessage, true), + createNonCommittableContext(IoTDBDescriptor.getInstance().getConfig().getDataNodeId())); + } + private SubscriptionEvent generateOutdatedErrorResponse() { return new SubscriptionEvent( SubscriptionPollResponseType.ERROR.getType(), @@ -4068,9 +4176,7 @@ private void setActiveUnderRuntimeLock(final boolean active) { memoryBlockedEntryBytes = -1L; prefetchInitialized = false; observedSeekGeneration = seekGeneration.get(); - pendingWalGapRetryRequested = false; - walGapWaitStartTimeMs = 0L; - lastWalGapWaitLogTimeMs = 0L; + resetWalReplayFailureState(); pendingSubscriptionWalResetSearchIndex = Long.MIN_VALUE; pendingSubscriptionWalResetGeneration = Long.MIN_VALUE; closeSubscriptionWALIterator(); @@ -4158,6 +4264,9 @@ public String getProgressStatusName() { if (!isActive) { return SubscriptionProgressSnapshot.STATUS_INACTIVE; } + if (Objects.nonNull(walReplayFailureMessage)) { + return SubscriptionProgressSnapshot.STATUS_STALLED; + } if (getLag() <= 0L) { return SubscriptionProgressSnapshot.STATUS_CAUGHT_UP; } 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 0f321341401ed..850704f723393 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 @@ -79,6 +79,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.locks.ReentrantReadWriteLock; import java.util.function.BooleanSupplier; @@ -925,6 +926,7 @@ public void testFilteredEmptyEntryAdvancesProgressWithoutEvent() throws Exceptio 1L, 1L, true); + queue.setSubscriptionMemoryManager(new SubscriptionMemoryManager(16L * 1024 * 1024)); final IndexedConsensusRequest request = new IndexedConsensusRequest( @@ -988,6 +990,7 @@ public void testFilteredEmptyEntryDoesNotAdvanceAcrossUncommittedData() throws E 1L, 1L, true); + queue.setSubscriptionMemoryManager(new SubscriptionMemoryManager(16L * 1024 * 1024)); final IndexedConsensusRequest dataRequest = new IndexedConsensusRequest( @@ -1182,6 +1185,7 @@ protected ReplayLocateDecision locateReplayStartForRegionProgress( walNode.getCurrentSearchIndex(), regionProgress, "empty sealed WAL"); } }; + queue.setSubscriptionMemoryManager(new SubscriptionMemoryManager(16L * 1024 * 1024)); pendingEntries(queue) .offer( @@ -1210,6 +1214,67 @@ protected ReplayLocateDecision locateReplayStartForRegionProgress( } } + @Test + public void testAtEndInitializationStartsAfterLastWalIndex() throws Exception { + final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); + final File systemDir = temporaryFolder.newFolder("atEndAfterTail"); + final File walDirectory = temporaryFolder.newFolder("atEndAfterTailWal"); + ConsensusPrefetchingQueue queue = null; + try { + final DataRegionId regionId = new DataRegionId(8); + final WALNode walNode = mock(WALNode.class); + when(walNode.getLogDirectory()).thenReturn(walDirectory); + when(walNode.getCurrentSearchIndex()).thenReturn(4L); + final IoTConsensusServerImpl serverImpl = mock(IoTConsensusServerImpl.class); + when(serverImpl.getConsensusReqReader()).thenReturn(walNode); + when(serverImpl.getWriterSafeFrontierTracker()).thenReturn(new WriterSafeFrontierTracker()); + final ConsensusLogToTabletConverter converter = mock(ConsensusLogToTabletConverter.class); + when(converter.convert(any())).thenReturn(Collections.singletonList(createTablet())); + when(converter.getDatabaseName()).thenReturn("db"); + final RegionProgress committedProgress = + new RegionProgress( + Collections.singletonMap( + new WriterId(regionId.toString(), 7), new WriterProgress(1000L, 2L))); + queue = + new ConsensusPrefetchingQueue( + "consumerGroup", + "topic", + TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE, + regionId, + serverImpl, + new SubscriptionWalRetentionPolicy( + "topic", + SubscriptionWalRetentionPolicy.UNBOUNDED, + SubscriptionWalRetentionPolicy.UNBOUNDED), + converter, + newCommitManager(systemDir), + committedProgress, + 5L, + 1L, + true) { + @Override + protected ReplayLocateDecision locateReplayStartForRegionProgress( + final RegionProgress regionProgress, final boolean seekAfter) { + return ReplayLocateDecision.atEnd( + walNode.getCurrentSearchIndex(), regionProgress, "empty sealed WAL"); + } + }; + queue.setSubscriptionMemoryManager(new SubscriptionMemoryManager(16L * 1024 * 1024)); + + assertNull(queue.poll("consumer")); + assertEquals(5L, queue.getCurrentReadSearchIndex()); + assertTrue(pendingEntries(queue).offer(createRequest(5L))); + queue.drivePrefetchOnce(); + assertEquals(6L, queue.getCurrentReadSearchIndex()); + assertEquals(1L, queue.getPendingPathAcceptedEntries()); + } finally { + if (queue != null) { + queue.close(); + } + IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir); + } + } + @Test public void testZeroPhysicalTimeUsesWriterLocalSequenceForReplayLookup() throws Exception { final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); @@ -1382,6 +1447,92 @@ ProgressWALReader openProgressWALReader(final File walFile) throws IOException { } } + @Test + public void testWalReplayRetriesGapWithoutSkippingEntries() throws Exception { + final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); + final File systemDir = temporaryFolder.newFolder("wal-replay-gap-counter"); + ConsensusPrefetchingQueue queue = null; + try { + final DataRegionId regionId = new DataRegionId(9); + final WALNode walNode = mock(WALNode.class); + when(walNode.getCurrentSearchIndex()).thenReturn(4L); + when(walNode.getLogDirectory()).thenReturn(systemDir); + final IoTConsensusServerImpl serverImpl = mock(IoTConsensusServerImpl.class); + when(serverImpl.getConsensusReqReader()).thenReturn(walNode); + when(serverImpl.getWriterSafeFrontierTracker()).thenReturn(new WriterSafeFrontierTracker()); + + final AtomicInteger conversionCount = new AtomicInteger(); + final ConsensusLogToTabletConverter converter = mock(ConsensusLogToTabletConverter.class); + when(converter.convert(any())) + .thenAnswer( + ignored -> { + conversionCount.incrementAndGet(); + return Collections.singletonList(createTablet()); + }); + when(converter.getDatabaseName()).thenReturn("db"); + + final Iterator initiallyVisibleEntries = + Arrays.asList(createRequest(1L), createRequest(4L)).iterator(); + final ProgressWALIterator initialIterator = mock(ProgressWALIterator.class); + when(initialIterator.hasNext()).thenAnswer(ignored -> initiallyVisibleEntries.hasNext()); + when(initialIterator.next()).thenAnswer(ignored -> initiallyVisibleEntries.next()); + final Iterator refreshedEntries = + Arrays.asList(createRequest(2L), createRequest(3L), createRequest(4L)).iterator(); + final ProgressWALIterator refreshedIterator = mock(ProgressWALIterator.class); + when(refreshedIterator.hasNext()).thenAnswer(ignored -> refreshedEntries.hasNext()); + when(refreshedIterator.next()).thenAnswer(ignored -> refreshedEntries.next()); + final AtomicInteger iteratorCreationCount = new AtomicInteger(); + + queue = + new ConsensusPrefetchingQueue( + "consumerGroup", + "topic", + TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE, + regionId, + serverImpl, + new SubscriptionWalRetentionPolicy( + "topic", + SubscriptionWalRetentionPolicy.UNBOUNDED, + SubscriptionWalRetentionPolicy.UNBOUNDED), + converter, + newCommitManager(systemDir), + new RegionProgress(Collections.emptyMap()), + 1L, + 1L, + true) { + @Override + protected ProgressWALIterator createSubscriptionWALIterator( + final long startSearchIndex) { + return iteratorCreationCount.getAndIncrement() == 0 + ? initialIterator + : refreshedIterator; + } + }; + queue.setSubscriptionMemoryManager(new SubscriptionMemoryManager(16L * 1024 * 1024)); + + assertNull(queue.poll("consumer")); + queue.drivePrefetchOnce(); + + assertEquals(1L, queue.getWalPathAcceptedEntries()); + assertEquals(1, conversionCount.get()); + assertEquals(0L, queue.getWalGapSkippedEntries()); + assertEquals(2L, queue.getCurrentReadSearchIndex()); + verify(walNode).rollWALFile(); + + queue.drivePrefetchOnce(); + + assertEquals(4L, queue.getWalPathAcceptedEntries()); + assertEquals(4, conversionCount.get()); + assertEquals(0L, queue.getWalGapSkippedEntries()); + assertEquals(5L, queue.getCurrentReadSearchIndex()); + } finally { + if (queue != null) { + queue.close(); + } + IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir); + } + } + @Test public void testPendingGapReplayHonorsPerRoundWalEntryLimit() throws Exception { final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); @@ -1507,14 +1658,24 @@ public void testWalReplayCountsOnlyUnavailableSearchIndexes() throws Exception { ConsensusPrefetchingQueue.class.getDeclaredMethod( "advanceWalReplayCursorIfPresent", IndexedConsensusRequest.class); advanceWalReplayCursorIfPresent.setAccessible(true); + final Method validateWalReplayContinuity = + ConsensusPrefetchingQueue.class.getDeclaredMethod( + "validateWalReplayContinuity", IndexedConsensusRequest.class, long.class); + validateWalReplayContinuity.setAccessible(true); advanceWalReplayCursorIfPresent.invoke(queue, createRequest(1L)); assertEquals(0L, queue.getWalGapSkippedEntries()); assertEquals(2L, queue.getCurrentReadSearchIndex()); - advanceWalReplayCursorIfPresent.invoke(queue, createRequest(4L)); + assertEquals( + "WAL_GAP", validateWalReplayContinuity.invoke(queue, createRequest(4L), 0L).toString()); + assertEquals(0L, queue.getWalGapSkippedEntries()); + assertEquals(2L, queue.getCurrentReadSearchIndex()); + + assertEquals( + "WAL_GAP", validateWalReplayContinuity.invoke(queue, createRequest(4L), 0L).toString()); assertEquals(2L, queue.getWalGapSkippedEntries()); - assertEquals(5L, queue.getCurrentReadSearchIndex()); + assertEquals(2L, queue.getCurrentReadSearchIndex()); } finally { if (queue != null) { queue.close(); @@ -1611,6 +1772,191 @@ protected ProgressWALIterator createSubscriptionWALIterator( } } + @Test + public void testWalReplayFailsCriticallyWhenGapRemainsUnavailable() throws Exception { + final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); + final File systemDir = temporaryFolder.newFolder("wal-replay-unrecoverable-gap"); + ConsensusPrefetchingQueue queue = null; + try { + final DataRegionId regionId = new DataRegionId(10); + final FakeConsensusReqReader reader = new FakeConsensusReqReader(); + reader.currentSearchIndex = 4L; + final IoTConsensusServerImpl serverImpl = mock(IoTConsensusServerImpl.class); + when(serverImpl.getConsensusReqReader()).thenReturn(reader); + when(serverImpl.getWriterSafeFrontierTracker()).thenReturn(new WriterSafeFrontierTracker()); + + final AtomicInteger conversionCount = new AtomicInteger(); + final ConsensusLogToTabletConverter converter = mock(ConsensusLogToTabletConverter.class); + when(converter.convert(any())) + .thenAnswer( + ignored -> { + conversionCount.incrementAndGet(); + return Collections.singletonList(createTablet()); + }); + when(converter.getDatabaseName()).thenReturn("db"); + + final Iterator initiallyVisibleEntries = + Arrays.asList(createRequest(1L), createRequest(4L)).iterator(); + final ProgressWALIterator initialIterator = mock(ProgressWALIterator.class); + when(initialIterator.hasNext()).thenAnswer(ignored -> initiallyVisibleEntries.hasNext()); + when(initialIterator.next()).thenAnswer(ignored -> initiallyVisibleEntries.next()); + final Iterator refreshedEntries = + Collections.singletonList(createRequest(4L)).iterator(); + final ProgressWALIterator refreshedIterator = mock(ProgressWALIterator.class); + when(refreshedIterator.hasNext()).thenAnswer(ignored -> refreshedEntries.hasNext()); + when(refreshedIterator.next()).thenAnswer(ignored -> refreshedEntries.next()); + final AtomicInteger iteratorCreationCount = new AtomicInteger(); + + queue = + new ConsensusPrefetchingQueue( + "consumerGroup", + "topic", + TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE, + regionId, + serverImpl, + new SubscriptionWalRetentionPolicy( + "topic", + SubscriptionWalRetentionPolicy.UNBOUNDED, + SubscriptionWalRetentionPolicy.UNBOUNDED), + converter, + newCommitManager(systemDir), + new RegionProgress(Collections.emptyMap()), + 1L, + 1L, + true) { + @Override + protected ProgressWALIterator createSubscriptionWALIterator( + final long startSearchIndex) { + return iteratorCreationCount.getAndIncrement() == 0 + ? initialIterator + : refreshedIterator; + } + }; + queue.setSubscriptionMemoryManager(new SubscriptionMemoryManager(16L * 1024 * 1024)); + + assertNull(queue.poll("consumer")); + queue.drivePrefetchOnce(); + queue.drivePrefetchOnce(); + + assertEquals(1L, queue.getWalPathAcceptedEntries()); + assertEquals(1, conversionCount.get()); + assertEquals(2L, queue.getWalGapSkippedEntries()); + assertEquals(2L, queue.getCurrentReadSearchIndex()); + assertEquals(4L, queue.getProgressStatus()); + + final SubscriptionEvent errorEvent = queue.poll("consumer"); + assertNotNull(errorEvent); + assertEquals( + SubscriptionPollResponseType.ERROR.getType(), + errorEvent.getCurrentResponse().getResponseType()); + assertTrue(((ErrorPayload) errorEvent.getCurrentResponse().getPayload()).isCritical()); + } finally { + if (queue != null) { + queue.close(); + } + IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir); + } + } + + @Test + public void testSeekClearsCriticalWalGapAndResumesAtTarget() throws Exception { + final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); + final File systemDir = temporaryFolder.newFolder("wal-gap-seek-recovery"); + ConsensusPrefetchingQueue queue = null; + try { + final FakeConsensusReqReader reader = new FakeConsensusReqReader(); + reader.currentSearchIndex = 7L; + final IoTConsensusServerImpl serverImpl = mock(IoTConsensusServerImpl.class); + when(serverImpl.getConsensusReqReader()).thenReturn(reader); + when(serverImpl.getWriterSafeFrontierTracker()).thenReturn(new WriterSafeFrontierTracker()); + + final ConsensusLogToTabletConverter converter = mock(ConsensusLogToTabletConverter.class); + when(converter.convert(any())).thenReturn(Collections.singletonList(createTablet())); + when(converter.getDatabaseName()).thenReturn("db"); + + final AtomicInteger iteratorCreationCount = new AtomicInteger(); + queue = + new ConsensusPrefetchingQueue( + "consumerGroup", + "topic", + TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE, + new DataRegionId(14), + serverImpl, + new SubscriptionWalRetentionPolicy( + "topic", + SubscriptionWalRetentionPolicy.UNBOUNDED, + SubscriptionWalRetentionPolicy.UNBOUNDED), + converter, + newCommitManager(systemDir), + new RegionProgress(Collections.emptyMap()), + 1L, + 1L, + true) { + @Override + protected ProgressWALIterator createSubscriptionWALIterator( + final long startSearchIndex) { + final List entries; + switch (iteratorCreationCount.getAndIncrement()) { + case 0: + entries = Arrays.asList(createRequest(1L), createRequest(4L)); + break; + case 1: + entries = Collections.singletonList(createRequest(4L)); + break; + default: + assertEquals(7L, startSearchIndex); + entries = Collections.singletonList(createRequest(7L)); + } + final Iterator walEntries = entries.iterator(); + final ProgressWALIterator iterator = mock(ProgressWALIterator.class); + when(iterator.hasNext()).thenAnswer(ignored -> walEntries.hasNext()); + when(iterator.next()).thenAnswer(ignored -> walEntries.next()); + return iterator; + } + }; + queue.setSubscriptionMemoryManager(new SubscriptionMemoryManager(16L * 1024 * 1024)); + + assertNull(queue.poll("consumer")); + queue.drivePrefetchOnce(); + queue.drivePrefetchOnce(); + + assertEquals(2L, queue.getCurrentReadSearchIndex()); + assertEquals(4L, queue.getProgressStatus()); + final SubscriptionEvent errorEvent = queue.poll("consumer"); + assertNotNull(errorEvent); + assertEquals( + SubscriptionPollResponseType.ERROR.getType(), + errorEvent.getCurrentResponse().getResponseType()); + assertTrue(((ErrorPayload) errorEvent.getCurrentResponse().getPayload()).isCritical()); + + applySeekForTest(queue, 7L); + + assertEquals(7L, queue.getCurrentReadSearchIndex()); + assertEquals(3L, queue.getProgressStatus()); + final Method validateWalReplayContinuity = + ConsensusPrefetchingQueue.class.getDeclaredMethod( + "validateWalReplayContinuity", IndexedConsensusRequest.class, long.class); + validateWalReplayContinuity.setAccessible(true); + assertEquals( + "STALE", validateWalReplayContinuity.invoke(queue, createRequest(4L), 0L).toString()); + assertEquals(7L, queue.getCurrentReadSearchIndex()); + assertNull(queue.poll("consumer")); + queue.drivePrefetchOnce(); + + assertEquals(8L, queue.getCurrentReadSearchIndex()); + final SubscriptionEvent resumedEvent = queue.poll("consumer"); + assertNotNull(resumedEvent); + assertEquals( + SubscriptionPollResponseType.TABLETS.getType(), + resumedEvent.getCurrentResponse().getResponseType()); + } finally { + if (queue != null) { + queue.close(); + } + IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir); + } + } + @Test public void testActivationRetriesUntilConfigNodeProgressIsExplicitlyAvailable() throws Exception { final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); @@ -1760,6 +2106,7 @@ public void testTabletMemoryReleasedAfterAck() throws Exception { 1L, 1L, true); + queue.setSubscriptionMemoryManager(new SubscriptionMemoryManager(16L * 1024 * 1024)); final IndexedConsensusRequest dataRequest = new IndexedConsensusRequest( @@ -1828,6 +2175,7 @@ public void testLateAckDoesNotWaitForPrefetchReadLock() throws Exception { 1L, 1L, true); + queue.setSubscriptionMemoryManager(new SubscriptionMemoryManager(16L * 1024 * 1024)); final IndexedConsensusRequest dataRequest = new IndexedConsensusRequest( @@ -2636,6 +2984,7 @@ public void testRequeueDoesNotIncrementNackCount() throws Exception { 1L, 1L, true); + queue.setSubscriptionMemoryManager(new SubscriptionMemoryManager(16L * 1024 * 1024)); reader.currentSearchIndex = 2L; assertTrue(pendingEntries(queue).offer(createRequest(1L))); @@ -2768,6 +3117,40 @@ private static void setPrefetchBinding( subtaskField.set(queue, subtask); } + private static void applySeekForTest( + final ConsensusPrefetchingQueue queue, final long targetSearchIndex) throws Exception { + final Field seekGenerationField = + ConsensusPrefetchingQueue.class.getDeclaredField("seekGeneration"); + seekGenerationField.setAccessible(true); + final AtomicLong seekGeneration = (AtomicLong) seekGenerationField.get(queue); + final long previousSeekGeneration = seekGeneration.get(); + final long targetSeekGeneration = seekGeneration.incrementAndGet(); + + final Class pendingSeekRequestClass = + Class.forName(ConsensusPrefetchingQueue.class.getName() + "$PendingSeekRequest"); + final Constructor constructor = + pendingSeekRequestClass.getDeclaredConstructor( + long.class, RegionProgress.class, String.class, boolean.class, long.class, long.class); + constructor.setAccessible(true); + final Object pendingSeekRequest = + constructor.newInstance( + targetSearchIndex, + new RegionProgress(Collections.emptyMap()), + "test", + true, + previousSeekGeneration, + targetSeekGeneration); + final Field pendingSeekRequestField = + ConsensusPrefetchingQueue.class.getDeclaredField("pendingSeekRequest"); + pendingSeekRequestField.setAccessible(true); + pendingSeekRequestField.set(queue, pendingSeekRequest); + + final Method applyPendingSeekRequest = + ConsensusPrefetchingQueue.class.getDeclaredMethod("applyPendingSeekRequestIfNecessary"); + applyPendingSeekRequest.setAccessible(true); + assertTrue((Boolean) applyPendingSeekRequest.invoke(queue)); + } + private static ProgressWALIterator subscriptionWalIterator(final ConsensusPrefetchingQueue queue) throws Exception { final Field field = ConsensusPrefetchingQueue.class.getDeclaredField("subscriptionWALIterator"); From 1aa29d8d199244526b062800efa5cccdf315cb6c Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 9 Oct 2026 13:00:53 +0800 Subject: [PATCH 2/8] Fix subscription replay across transient WAL rotation visibility --- .../iotdb/db/i18n/DataNodePipeMessages.java | 2 + .../iotdb/db/i18n/DataNodePipeMessages.java | 2 + .../broker/consensus/ProgressWALIterator.java | 41 +++++-- .../consensus/ProgressWALIteratorTest.java | 108 +++++++++++++++++- 4 files changed, 143 insertions(+), 10 deletions(-) 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 95583ded3a923..dc3054a5eb873 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 @@ -2202,6 +2202,8 @@ private DataNodePipeMessages() {} "ProgressWALIterator: failed to open near-live WAL file {}, retrying without blacklisting"; public static final String PIPE_LOG_PROGRESSWALITERATOR_ERROR_READING_WAL_2DB46D41 = "ProgressWALIterator: error reading WAL"; + public static final String MESSAGE_PREVIOUS_LIVE_WAL_FILE_VERSION_ARG_IS_TEMPORARILY_ABSENT_DURING_ROTATION_CB8C8CE6 = + "previous live WAL file version %s is temporarily absent during rotation"; public static final String PIPE_LOG_PROGRESSWALITERATOR_FAILED_TO_PROCESS_WAL_FILE_SKIPPING_REMAINING_ENTRIES_093F14A9 = "ProgressWALIterator: failed to process WAL file {}, skipping remaining entries"; 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 ccb9cd42da20d..49540844456a3 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 @@ -2045,6 +2045,8 @@ private DataNodePipeMessages() {} "ProgressWALIterator：打开 near-live WAL 文件 {} 失败，不加入黑名单并重试"; public static final String PIPE_LOG_PROGRESSWALITERATOR_ERROR_READING_WAL_2DB46D41 = "ProgressWALIterator：读取 WAL 出错"; + public static final String MESSAGE_PREVIOUS_LIVE_WAL_FILE_VERSION_ARG_IS_TEMPORARILY_ABSENT_DURING_ROTATION_CB8C8CE6 = + "上一个活跃 WAL 文件版本 %s 在轮转期间暂时不可见"; public static final String PIPE_LOG_PROGRESSWALITERATOR_FAILED_TO_PROCESS_WAL_FILE_SKIPPING_REMAINING_ENTRIES_093F14A9 = "ProgressWALIterator：处理 WAL 文件 {} 失败，跳过该文件的剩余条目"; 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 ee61c105c1507..6c9ffecaea55d 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 @@ -79,6 +79,7 @@ public interface WriterProgressCoverage { private int currentFileIndex = -1; private ProgressWALReader currentReader; private long currentReaderVersionId = -1L; + private long pendingNearLiveWalVersionId = -1L; private boolean currentReaderUsesLiveSnapshot = false; private int consumedEntryCountInCurrentFile = 0; private final Set skippedBrokenWalVersionIds = new HashSet<>(); @@ -390,6 +391,7 @@ public void close() throws IOException { pendingRequests.clear(); pendingSearchIndex = Long.MIN_VALUE; pendingLocalSeq = Long.MIN_VALUE; + pendingNearLiveWalVersionId = -1L; lastError = null; incompleteScan = false; incompleteScanDetail = null; @@ -489,9 +491,27 @@ private IndexedConsensusRequest advance() throws IOException { } private boolean openNextReader() throws IOException { + if (pendingNearLiveWalVersionId >= 0L) { + refresh(); + final int retryFileIndex = findFileIndexByVersion(pendingNearLiveWalVersionId); + if (retryFileIndex < 0) { + return false; + } + currentFileIndex = retryFileIndex - 1; + pendingNearLiveWalVersionId = -1L; + } while (++currentFileIndex < walFiles.length) { - if (openReaderAtIndex(currentFileIndex, 0)) { - return true; + final int candidateFileIndex = currentFileIndex; + final long candidateVersionId = walFileVersionIds[candidateFileIndex]; + try { + if (openReaderAtIndex(candidateFileIndex, 0)) { + return true; + } + } catch (final IOException e) { + // A near-live WAL file can be temporarily unavailable while it is renamed or sealed. + // Retry this version on the next scan instead of jumping to its successor. + pendingNearLiveWalVersionId = candidateVersionId; + throw e; } } return false; @@ -520,7 +540,15 @@ private boolean reopenLiveSnapshotReader() throws IOException { final int previousFileIndex = findFileIndexByVersion(currentReaderVersionId); if (previousFileIndex < 0) { - return openFirstReaderAfterVersion(currentReaderVersionId); + // The old live file may be between its seal and rename. Its last snapshot does not prove + // that every entry in the sealed file was consumed, so wait for the old version to reappear. + markIncompleteScan( + String.format( + DataNodePipeMessages + .MESSAGE_PREVIOUS_LIVE_WAL_FILE_VERSION_ARG_IS_TEMPORARILY_ABSENT_DURING_ROTATION_CB8C8CE6, + currentReaderVersionId), + null); + return false; } if (openReaderAtIndex(previousFileIndex, consumedEntryCountInCurrentFile)) { return true; @@ -590,12 +618,7 @@ private boolean openReaderAtIndex( } } } - markIncompleteScan( - String.format( - "failed to open near-live WAL file %s while replay scan was still in progress", - walFile.getName()), - e); - return false; + throw e; } recordSkippedBrokenWalFile(versionId, walFile, e); return false; 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 45bff39d6ba78..f621a727d160d 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 @@ -38,6 +38,7 @@ import java.nio.ByteBuffer; import java.nio.file.Files; import java.nio.file.Path; +import java.util.concurrent.atomic.AtomicLong; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; @@ -452,7 +453,7 @@ public void testIteratorMarksIncompleteScanWhenNearLiveWalCannotBeOpened() throw assertFalse(iterator.hasNext()); assertTrue(iterator.hasIncompleteScan()); assertTrue(iterator.hasReadError()); - assertTrue(iterator.getIncompleteScanDetail().contains("near-live WAL file")); + assertTrue(iterator.getLastError() != null); } } finally { Files.deleteIfExists(brokenLiveWal.toPath()); @@ -460,6 +461,111 @@ public void testIteratorMarksIncompleteScanWhenNearLiveWalCannotBeOpened() throw } } + @Test + public void testNearLiveWalOpenFailureDoesNotSkipToSuccessor() throws Exception { + final Path dir = Files.createTempDirectory("progress-wal-iterator-near-live-retry"); + final File firstWal = + dir.resolve(WALFileUtils.getLogFileName(0, 0, WALFileStatus.CONTAINS_SEARCH_INDEX)) + .toFile(); + final File temporarilyUnavailableWal = + dir.resolve(WALFileUtils.getLogFileName(1, 1, WALFileStatus.CONTAINS_SEARCH_INDEX)) + .toFile(); + final File successorWal = + dir.resolve(WALFileUtils.getLogFileName(2, 2, WALFileStatus.CONTAINS_SEARCH_INDEX)) + .toFile(); + + try { + try (WALWriter writer = new WALWriter(firstWal, WALFileVersion.V3)) { + writer.write(searchableEntry(1L), singleEntryMeta(19, 1L, 1L, 100L, 7, 1L)); + } + Files.write(temporarilyUnavailableWal.toPath(), new byte[128]); + try (WALWriter writer = new WALWriter(successorWal, WALFileVersion.V3)) { + writer.write(searchableEntry(3L), singleEntryMeta(19, 3L, 1L, 300L, 7, 3L)); + } + + final WALNode walNode = mock(WALNode.class); + when(walNode.getLogDirectory()).thenReturn(dir.toFile()); + when(walNode.getCurrentWALFileVersion()).thenReturn(2L); + when(walNode.getCurrentWALMetaDataSnapshot()) + .thenReturn(singleEntryMeta(19, 3L, 1L, 300L, 7, 3L)); + + try (ProgressWALIterator iterator = new ProgressWALIterator(walNode, 1L)) { + assertTrue(iterator.hasNext()); + assertEquals(1L, iterator.next().getSearchIndex()); + assertFalse(iterator.hasNext()); + assertTrue(iterator.hasReadError()); + + Files.delete(temporarilyUnavailableWal.toPath()); + assertFalse(iterator.hasNext()); + try (WALWriter writer = new WALWriter(temporarilyUnavailableWal, WALFileVersion.V3)) { + writer.write(searchableEntry(2L), singleEntryMeta(19, 2L, 1L, 200L, 7, 2L)); + } + + assertTrue(iterator.hasNext()); + assertEquals(2L, iterator.next().getSearchIndex()); + assertFalse(iterator.hasReadError()); + assertTrue(iterator.hasNext()); + assertEquals(3L, iterator.next().getSearchIndex()); + } + } finally { + Files.deleteIfExists(firstWal.toPath()); + Files.deleteIfExists(temporarilyUnavailableWal.toPath()); + Files.deleteIfExists(successorWal.toPath()); + Files.deleteIfExists(dir); + } + } + + @Test + public void testRotatedLiveWalMustReappearBeforeReadingSuccessor() throws Exception { + final Path dir = Files.createTempDirectory("progress-wal-iterator-rotation-rename"); + final File firstWal = + dir.resolve(WALFileUtils.getLogFileName(0, 0, WALFileStatus.CONTAINS_SEARCH_INDEX)) + .toFile(); + final Path temporarilyHiddenWal = dir.resolve(firstWal.getName() + ".hidden"); + final File successorWal = + dir.resolve(WALFileUtils.getLogFileName(1, 2, WALFileStatus.CONTAINS_SEARCH_INDEX)) + .toFile(); + + try { + try (WALWriter writer = new WALWriter(firstWal, WALFileVersion.V3)) { + writer.write(searchableEntry(1L), singleEntryMeta(19, 1L, 1L, 100L, 7, 1L)); + writer.write(searchableEntry(2L), singleEntryMeta(19, 2L, 1L, 200L, 7, 2L)); + } + + final AtomicLong currentWalVersion = new AtomicLong(0L); + final WALNode walNode = mock(WALNode.class); + when(walNode.getLogDirectory()).thenReturn(dir.toFile()); + when(walNode.getCurrentWALFileVersion()).thenAnswer(ignored -> currentWalVersion.get()); + when(walNode.getCurrentWALMetaDataSnapshot()) + .thenReturn(singleEntryMeta(19, 1L, 1L, 100L, 7, 1L)); + + try (ProgressWALIterator iterator = new ProgressWALIterator(walNode, 1L)) { + assertTrue(iterator.hasNext()); + assertEquals(1L, iterator.next().getSearchIndex()); + assertFalse(iterator.hasNext()); + + Files.move(firstWal.toPath(), temporarilyHiddenWal); + try (WALWriter writer = new WALWriter(successorWal, WALFileVersion.V3)) { + writer.write(searchableEntry(3L), singleEntryMeta(19, 3L, 1L, 300L, 7, 3L)); + } + currentWalVersion.set(1L); + assertFalse(iterator.hasNext()); + assertTrue(iterator.hasIncompleteScan()); + + Files.move(temporarilyHiddenWal, firstWal.toPath()); + assertTrue(iterator.hasNext()); + assertEquals(2L, iterator.next().getSearchIndex()); + assertTrue(iterator.hasNext()); + assertEquals(3L, iterator.next().getSearchIndex()); + } + } finally { + Files.deleteIfExists(firstWal.toPath()); + Files.deleteIfExists(temporarilyHiddenWal); + Files.deleteIfExists(successorWal.toPath()); + Files.deleteIfExists(dir); + } + } + @Test public void testAdvanceToKeepsUncoveredFollowerRequestInCurrentFile() throws Exception { final Path dir = Files.createTempDirectory("progress-wal-iterator-current-file-advance"); From 22f4b02d76cbf25c0e763f7c67475e915c864685 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 9 Oct 2026 13:15:36 +0800 Subject: [PATCH 3/8] Hide rotated WAL under a non-WAL filename in regression test --- .../subscription/broker/consensus/ProgressWALIteratorTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 f621a727d160d..5377c48510c3f 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 @@ -521,7 +521,7 @@ public void testRotatedLiveWalMustReappearBeforeReadingSuccessor() throws Except final File firstWal = dir.resolve(WALFileUtils.getLogFileName(0, 0, WALFileStatus.CONTAINS_SEARCH_INDEX)) .toFile(); - final Path temporarilyHiddenWal = dir.resolve(firstWal.getName() + ".hidden"); + final Path temporarilyHiddenWal = dir.resolve("temporarily-hidden"); final File successorWal = dir.resolve(WALFileUtils.getLogFileName(1, 2, WALFileStatus.CONTAINS_SEARCH_INDEX)) .toFile(); From e1c9daf5f7c86b9694ca9e0105d57ccf7ca94c09 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 9 Oct 2026 15:22:01 +0800 Subject: [PATCH 4/8] Fix subscription WAL gap recovery behavior --- .../iotdb/db/i18n/DataNodePipeMessages.java | 17 ++--- .../iotdb/db/i18n/DataNodePipeMessages.java | 16 ++-- .../consensus/ConsensusPrefetchingQueue.java | 73 +++++++------------ .../broker/consensus/ProgressWALIterator.java | 10 +-- .../ConsensusPrefetchingQueueTest.java | 35 ++++----- .../consensus/ProgressWALIteratorTest.java | 7 +- 6 files changed, 65 insertions(+), 93 deletions(-) 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 dc3054a5eb873..2f680887368b7 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 @@ -2202,8 +2202,8 @@ private DataNodePipeMessages() {} "ProgressWALIterator: failed to open near-live WAL file {}, retrying without blacklisting"; public static final String PIPE_LOG_PROGRESSWALITERATOR_ERROR_READING_WAL_2DB46D41 = "ProgressWALIterator: error reading WAL"; - public static final String MESSAGE_PREVIOUS_LIVE_WAL_FILE_VERSION_ARG_IS_TEMPORARILY_ABSENT_DURING_ROTATION_CB8C8CE6 = - "previous live WAL file version %s is temporarily absent during rotation"; + public static final String MESSAGE_PREVIOUS_LIVE_WAL_FILE_VERSION_ARG_IS_ABSENT_FROM_CURRENT_SCAN_RETRYING_5484DEF8 = + "previous live WAL file version %s is absent from current scan; retrying"; public static final String PIPE_LOG_PROGRESSWALITERATOR_FAILED_TO_PROCESS_WAL_FILE_SKIPPING_REMAINING_ENTRIES_093F14A9 = "ProgressWALIterator: failed to process WAL file {}, skipping remaining entries"; @@ -2218,13 +2218,12 @@ 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 PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_WAL_REPLAY_FOUND_UNAVAILABLE_SEARCH_INDEXES_E0CBFFFA = - "ConsensusPrefetchingQueue {}: WAL replay found unavailable search indexes [{}, {}) before " - + "searchIndex {}; forcing WAL refresh before retry"; - public static final String MESSAGE_CONSENSUSPREFETCHINGQUEUE_WAL_REPLAY_CANNOT_RECOVER_SEARCH_INDEXES_70781B22 = - "ConsensusPrefetchingQueue %s: WAL replay cannot recover search indexes [%s, %s), " - + "unavailableEntries=%s, totalWalGapSkippedEntries=%s; subscription delivery is " - + "stalled to prevent silent data loss"; + public static final String + LOG_ARG_WAL_REPLAY_OBSERVED_GAP_ARG_ARG_BEFORE_SEARCHINDEX_ARG_REFRESHING_WAL_AND_RETRYING_0669FB51 = + "{}: WAL replay observed gap [{}, {}) before searchIndex {}; refreshing WAL and retrying"; + public static final String + LOG_ARG_WAL_GAP_ARG_ARG_PERSISTS_AFTER_REFRESH_SKIPPING_ARG_TOTAL_SKIPPED_ARG_16FE0680 = + "{}: WAL gap [{}, {}) persists after refresh; skipping {}, total skipped {}"; public static final String PIPE_LOG_PIPE_TERMINATE_EVENT_COMMITTED_FOR_HISTORICAL_TRANSFER_CREATIONTIME_9B807B28 = "Pipe {}@{}: terminate event committed for historical transfer. creationTime: {}, " + "shouldMark: {}. {}"; diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java index 49540844456a3..8adb0976fcc9f 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 @@ -2045,8 +2045,8 @@ private DataNodePipeMessages() {} "ProgressWALIterator：打开 near-live WAL 文件 {} 失败，不加入黑名单并重试"; public static final String PIPE_LOG_PROGRESSWALITERATOR_ERROR_READING_WAL_2DB46D41 = "ProgressWALIterator：读取 WAL 出错"; - public static final String MESSAGE_PREVIOUS_LIVE_WAL_FILE_VERSION_ARG_IS_TEMPORARILY_ABSENT_DURING_ROTATION_CB8C8CE6 = - "上一个活跃 WAL 文件版本 %s 在轮转期间暂时不可见"; + public static final String MESSAGE_PREVIOUS_LIVE_WAL_FILE_VERSION_ARG_IS_ABSENT_FROM_CURRENT_SCAN_RETRYING_5484DEF8 = + "上一个活跃 WAL 文件版本 %s 未出现在本次扫描结果中；稍后重试"; public static final String PIPE_LOG_PROGRESSWALITERATOR_FAILED_TO_PROCESS_WAL_FILE_SKIPPING_REMAINING_ENTRIES_093F14A9 = "ProgressWALIterator：处理 WAL 文件 {} 失败，跳过该文件的剩余条目"; @@ -2059,12 +2059,12 @@ 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 PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_WAL_REPLAY_FOUND_UNAVAILABLE_SEARCH_INDEXES_E0CBFFFA = - "ConsensusPrefetchingQueue {}：WAL 回放发现不可用 searchIndex 区间 [{}, {})，下一个可见 " - + "searchIndex={}；正在强制刷新 WAL 后重试"; - public static final String MESSAGE_CONSENSUSPREFETCHINGQUEUE_WAL_REPLAY_CANNOT_RECOVER_SEARCH_INDEXES_70781B22 = - "ConsensusPrefetchingQueue %s：WAL 回放无法恢复 searchIndex 区间 [%s, %s)，不可用条目数=%s，" - + "walGapSkippedEntries 总数=%s；为防止静默丢数，订阅投递已停滞"; + public static final String + LOG_ARG_WAL_REPLAY_OBSERVED_GAP_ARG_ARG_BEFORE_SEARCHINDEX_ARG_REFRESHING_WAL_AND_RETRYING_0669FB51 = + "{}：WAL 回放观察到缺口 [{}, {})，下一个可见 searchIndex={}；正在刷新 WAL 并重试"; + public static final String + LOG_ARG_WAL_GAP_ARG_ARG_PERSISTS_AFTER_REFRESH_SKIPPING_ARG_TOTAL_SKIPPED_ARG_16FE0680 = + "{}：WAL 缺口 [{}, {}) 在刷新后仍存在；跳过 {} 条，累计跳过 {} 条"; public static final String PIPE_LOG_PIPE_TERMINATE_EVENT_COMMITTED_FOR_HISTORICAL_TRANSFER_CREATIONTIME_9B807B28 = "Pipe {}@{}：历史传输的终止事件已提交。creationTime：{}，shouldMark：{}。{}"; public static final String PIPE_LOG_PIPE_HISTORICAL_SOURCE_HAS_SUPPLIED_ALL_EVENTS_EMITTING_8B58DE19 = diff --git a/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 be194c986ba7d..83252aaad6bbd 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 @@ -300,9 +300,6 @@ public class ConsensusPrefetchingQueue { /** Search index whose WAL visibility gap is being retried after forcing a WAL refresh. */ private volatile long walGapRetryExpectedSearchIndex = Long.MIN_VALUE; - /** Critical replay failure returned to consumers instead of silently advancing past data. */ - private volatile String walReplayFailureMessage; - /** Fallback committed region progress from local persisted state. */ private final RegionProgress fallbackCommittedRegionProgress; @@ -777,9 +774,6 @@ public SubscriptionEvent poll(final String consumerId, final RegionProgress regi if (pendingSeekRequest != null) { return null; } - if (Objects.nonNull(walReplayFailureMessage)) { - return generateCriticalErrorResponse(walReplayFailureMessage); - } final SubscriptionEvent event = pollInternal(consumerId); if (Objects.nonNull(event) && prefetchingQueue.size() < MAX_PREFETCHING_QUEUE_SIZE) { requestPrefetch(); @@ -846,7 +840,7 @@ private boolean initPrefetchUnderInitializationLock(final RegionProgress regionP // readers without WAL support. this.subscriptionWALIterator = createSubscriptionWALIterator(resolvedStart.getStartSearchIndex()); - resetWalReplayFailureState(); + resetWalGapRetryState(); this.prefetchInitialized = true; this.observedSeekGeneration = seekGeneration.get(); discardBatch(this.lingerBatch); @@ -1439,11 +1433,6 @@ public PrefetchRoundResult drivePrefetchOnce() { applyPendingSubscriptionWalReset(observedSeekGeneration); recycleInFlightEvents(); - if (Objects.nonNull(walReplayFailureMessage)) { - blockRealtimeAdmission(); - return PrefetchRoundResult.dormant(); - } - if (!isActive) { blockRealtimeAdmission(); return computeIdleRoundResult(); @@ -1533,9 +1522,7 @@ public PrefetchRoundResult drivePrefetchOnce() { if (batch.isEmpty() && lingerBatch.isEmpty()) { final MaterializationResult walResult = tryCatchUpFromWAL(observedSeekGeneration); if (walResult == MaterializationResult.WAL_GAP) { - return Objects.nonNull(walReplayFailureMessage) - ? PrefetchRoundResult.dormant() - : PrefetchRoundResult.rescheduleAfter(WAL_GAP_RETRY_SLEEP_MS); + return PrefetchRoundResult.rescheduleAfter(WAL_GAP_RETRY_SLEEP_MS); } if (walResult == MaterializationResult.MEMORY_BLOCKED) { blockRealtimeAdmission(); @@ -2064,7 +2051,21 @@ private void advanceWalReplayCursorIfPresent(final IndexedConsensusRequest reque if (!hasLocalSearchIndex(request)) { return; } - nextExpectedSearchIndex.set(request.getSearchIndex() + 1); + final long actualSearchIndex = request.getSearchIndex(); + final long expectedSearchIndex = nextExpectedSearchIndex.get(); + if (actualSearchIndex > expectedSearchIndex) { + final long skippedEntries = actualSearchIndex - expectedSearchIndex; + final long totalSkippedEntries = walGapSkippedEntries.addAndGet(skippedEntries); + LOGGER.warn( + DataNodePipeMessages + .LOG_ARG_WAL_GAP_ARG_ARG_PERSISTS_AFTER_REFRESH_SKIPPING_ARG_TOTAL_SKIPPED_ARG_16FE0680, + this, + expectedSearchIndex, + actualSearchIndex, + skippedEntries, + totalSkippedEntries); + } + nextExpectedSearchIndex.set(actualSearchIndex + 1); } private MaterializationResult validateWalReplayContinuity( @@ -2094,7 +2095,7 @@ private MaterializationResult validateWalReplayContinuity( walGapRetryExpectedSearchIndex = expectedSearchIndex; LOGGER.warn( DataNodePipeMessages - .PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_WAL_REPLAY_FOUND_UNAVAILABLE_SEARCH_INDEXES_E0CBFFFA, + .LOG_ARG_WAL_REPLAY_OBSERVED_GAP_ARG_ARG_BEFORE_SEARCHINDEX_ARG_REFRESHING_WAL_AND_RETRYING_0669FB51, this, expectedSearchIndex, actualSearchIndex, @@ -2114,30 +2115,16 @@ private MaterializationResult validateWalReplayContinuity( if (seekGeneration.get() != expectedSeekGeneration || isClosed) { return MaterializationResult.STALE; } - if (Objects.isNull(walReplayFailureMessage)) { - final long unavailableEntries = actualSearchIndex - expectedSearchIndex; - final long totalUnavailableEntries = walGapSkippedEntries.addAndGet(unavailableEntries); - walReplayFailureMessage = - String.format( - DataNodePipeMessages - .MESSAGE_CONSENSUSPREFETCHINGQUEUE_WAL_REPLAY_CANNOT_RECOVER_SEARCH_INDEXES_70781B22, - this, - expectedSearchIndex, - actualSearchIndex, - unavailableEntries, - totalUnavailableEntries); - LOGGER.error(walReplayFailureMessage); - blockRealtimeAdmission(); - } - return MaterializationResult.WAL_GAP; + walGapRetryExpectedSearchIndex = Long.MIN_VALUE; + pendingWalGapRetryRequested = false; + return MaterializationResult.SUCCESS; } - private void resetWalReplayFailureState() { + private void resetWalGapRetryState() { pendingWalGapRetryRequested = false; walGapWaitStartTimeMs = 0L; lastWalGapWaitLogTimeMs = 0L; walGapRetryExpectedSearchIndex = Long.MIN_VALUE; - walReplayFailureMessage = null; } private void ensureSubscriptionWalReadable() { @@ -3140,7 +3127,7 @@ public void cleanUp() { reconcileRetainedTabletMemoryAfterCleanup(); memoryBlockedEntryBytes = -1L; resetBatchWriterProgress(); - resetWalReplayFailureState(); + resetWalGapRetryState(); pendingSubscriptionWalResetSearchIndex = Long.MIN_VALUE; pendingSubscriptionWalResetGeneration = Long.MIN_VALUE; closeSubscriptionWALIterator(); @@ -3451,7 +3438,7 @@ private void applySeekResetUnderWriteLock(final PendingSeekRequest request) { reconcileRetainedTabletMemoryAfterCleanup(); resetBatchWriterProgress(); observedSeekGeneration = seekGeneration.get(); - resetWalReplayFailureState(); + resetWalGapRetryState(); // 5. Reset commit state to the writer progress immediately before the first re-delivered // entry so seek/rebind resumes from the intended frontier. @@ -4065,13 +4052,6 @@ private SubscriptionEvent generateErrorResponse(final String errorMessage) { createNonCommittableContext(IoTDBDescriptor.getInstance().getConfig().getDataNodeId())); } - private SubscriptionEvent generateCriticalErrorResponse(final String errorMessage) { - return new SubscriptionEvent( - SubscriptionPollResponseType.ERROR.getType(), - new ErrorPayload(errorMessage, true), - createNonCommittableContext(IoTDBDescriptor.getInstance().getConfig().getDataNodeId())); - } - private SubscriptionEvent generateOutdatedErrorResponse() { return new SubscriptionEvent( SubscriptionPollResponseType.ERROR.getType(), @@ -4176,7 +4156,7 @@ private void setActiveUnderRuntimeLock(final boolean active) { memoryBlockedEntryBytes = -1L; prefetchInitialized = false; observedSeekGeneration = seekGeneration.get(); - resetWalReplayFailureState(); + resetWalGapRetryState(); pendingSubscriptionWalResetSearchIndex = Long.MIN_VALUE; pendingSubscriptionWalResetGeneration = Long.MIN_VALUE; closeSubscriptionWALIterator(); @@ -4264,9 +4244,6 @@ public String getProgressStatusName() { if (!isActive) { return SubscriptionProgressSnapshot.STATUS_INACTIVE; } - if (Objects.nonNull(walReplayFailureMessage)) { - return SubscriptionProgressSnapshot.STATUS_STALLED; - } if (getLag() <= 0L) { return SubscriptionProgressSnapshot.STATUS_CAUGHT_UP; } 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 6c9ffecaea55d..2ce6f09a3495b 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 @@ -508,8 +508,8 @@ private boolean openNextReader() throws IOException { return true; } } catch (final IOException e) { - // A near-live WAL file can be temporarily unavailable while it is renamed or sealed. - // Retry this version on the next scan instead of jumping to its successor. + // Retry this version on the next scan instead of jumping to its successor when the + // discovered path cannot be opened. pendingNearLiveWalVersionId = candidateVersionId; throw e; } @@ -540,12 +540,12 @@ private boolean reopenLiveSnapshotReader() throws IOException { final int previousFileIndex = findFileIndexByVersion(currentReaderVersionId); if (previousFileIndex < 0) { - // The old live file may be between its seal and rename. Its last snapshot does not prove - // that every entry in the sealed file was consumed, so wait for the old version to reappear. + // The old live version is absent from this scan. Its last snapshot does not prove that every + // entry in the sealed file was consumed, so wait for the old version to reappear. markIncompleteScan( String.format( DataNodePipeMessages - .MESSAGE_PREVIOUS_LIVE_WAL_FILE_VERSION_ARG_IS_TEMPORARILY_ABSENT_DURING_ROTATION_CB8C8CE6, + .MESSAGE_PREVIOUS_LIVE_WAL_FILE_VERSION_ARG_IS_ABSENT_FROM_CURRENT_SCAN_RETRYING_5484DEF8, currentReaderVersionId), null); return false; 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 850704f723393..6a5f513e9b85f 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 @@ -1673,9 +1673,12 @@ public void testWalReplayCountsOnlyUnavailableSearchIndexes() throws Exception { assertEquals(2L, queue.getCurrentReadSearchIndex()); assertEquals( - "WAL_GAP", validateWalReplayContinuity.invoke(queue, createRequest(4L), 0L).toString()); - assertEquals(2L, queue.getWalGapSkippedEntries()); + "SUCCESS", validateWalReplayContinuity.invoke(queue, createRequest(4L), 0L).toString()); + assertEquals(0L, queue.getWalGapSkippedEntries()); assertEquals(2L, queue.getCurrentReadSearchIndex()); + advanceWalReplayCursorIfPresent.invoke(queue, createRequest(4L)); + assertEquals(2L, queue.getWalGapSkippedEntries()); + assertEquals(5L, queue.getCurrentReadSearchIndex()); } finally { if (queue != null) { queue.close(); @@ -1773,7 +1776,7 @@ protected ProgressWALIterator createSubscriptionWALIterator( } @Test - public void testWalReplayFailsCriticallyWhenGapRemainsUnavailable() throws Exception { + public void testWalReplayContinuesWhenGapRemainsUnavailable() throws Exception { final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); final File systemDir = temporaryFolder.newFolder("wal-replay-unrecoverable-gap"); ConsensusPrefetchingQueue queue = null; @@ -1838,18 +1841,16 @@ protected ProgressWALIterator createSubscriptionWALIterator( queue.drivePrefetchOnce(); queue.drivePrefetchOnce(); - assertEquals(1L, queue.getWalPathAcceptedEntries()); - assertEquals(1, conversionCount.get()); + assertEquals(2L, queue.getWalPathAcceptedEntries()); + assertEquals(2, conversionCount.get()); assertEquals(2L, queue.getWalGapSkippedEntries()); - assertEquals(2L, queue.getCurrentReadSearchIndex()); - assertEquals(4L, queue.getProgressStatus()); + assertEquals(5L, queue.getCurrentReadSearchIndex()); - final SubscriptionEvent errorEvent = queue.poll("consumer"); - assertNotNull(errorEvent); + final SubscriptionEvent deliveredEvent = queue.poll("consumer"); + assertNotNull(deliveredEvent); assertEquals( - SubscriptionPollResponseType.ERROR.getType(), - errorEvent.getCurrentResponse().getResponseType()); - assertTrue(((ErrorPayload) errorEvent.getCurrentResponse().getPayload()).isCritical()); + SubscriptionPollResponseType.TABLETS.getType(), + deliveredEvent.getCurrentResponse().getResponseType()); } finally { if (queue != null) { queue.close(); @@ -1859,7 +1860,7 @@ protected ProgressWALIterator createSubscriptionWALIterator( } @Test - public void testSeekClearsCriticalWalGapAndResumesAtTarget() throws Exception { + public void testSeekClearsPendingWalGapRetryAndResumesAtTarget() throws Exception { final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); final File systemDir = temporaryFolder.newFolder("wal-gap-seek-recovery"); ConsensusPrefetchingQueue queue = null; @@ -1918,16 +1919,8 @@ protected ProgressWALIterator createSubscriptionWALIterator( assertNull(queue.poll("consumer")); queue.drivePrefetchOnce(); - queue.drivePrefetchOnce(); assertEquals(2L, queue.getCurrentReadSearchIndex()); - assertEquals(4L, queue.getProgressStatus()); - final SubscriptionEvent errorEvent = queue.poll("consumer"); - assertNotNull(errorEvent); - assertEquals( - SubscriptionPollResponseType.ERROR.getType(), - errorEvent.getCurrentResponse().getResponseType()); - assertTrue(((ErrorPayload) errorEvent.getCurrentResponse().getPayload()).isCritical()); applySeekForTest(queue, 7L); 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 5377c48510c3f..941c7f49321bd 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 @@ -516,8 +516,8 @@ public void testNearLiveWalOpenFailureDoesNotSkipToSuccessor() throws Exception } @Test - public void testRotatedLiveWalMustReappearBeforeReadingSuccessor() throws Exception { - final Path dir = Files.createTempDirectory("progress-wal-iterator-rotation-rename"); + public void testTemporarilyAbsentLiveWalMustReappearBeforeReadingSuccessor() throws Exception { + final Path dir = Files.createTempDirectory("progress-wal-iterator-absent-version"); final File firstWal = dir.resolve(WALFileUtils.getLogFileName(0, 0, WALFileStatus.CONTAINS_SEARCH_INDEX)) .toFile(); @@ -544,6 +544,8 @@ public void testRotatedLiveWalMustReappearBeforeReadingSuccessor() throws Except assertEquals(1L, iterator.next().getSearchIndex()); assertFalse(iterator.hasNext()); + // Simulate a listing in which an unread version is absent. Normal rotation leaves the + // name of a WAL with search-index entries unchanged. Files.move(firstWal.toPath(), temporarilyHiddenWal); try (WALWriter writer = new WALWriter(successorWal, WALFileVersion.V3)) { writer.write(searchableEntry(3L), singleEntryMeta(19, 3L, 1L, 300L, 7, 3L)); @@ -551,6 +553,7 @@ public void testRotatedLiveWalMustReappearBeforeReadingSuccessor() throws Except currentWalVersion.set(1L); assertFalse(iterator.hasNext()); assertTrue(iterator.hasIncompleteScan()); + assertFalse(iterator.hasNext()); Files.move(temporarilyHiddenWal, firstWal.toPath()); assertTrue(iterator.hasNext()); From f689fdb5549becef1b2dd80265387d6623103721 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 9 Oct 2026 15:34:18 +0800 Subject: [PATCH 5/8] Preserve absent local index in subscription WAL replay --- .../db/subscription/broker/consensus/ProgressWALIterator.java | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) 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 2ce6f09a3495b..c56b522a305aa 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 @@ -757,7 +757,9 @@ private long getSearchIndexOrMetadataFallback( } buffer.position(SEARCH_INDEX_OFFSET); final long bodySearchIndex = SearchNode.extractSearchIndex(buffer.getLong()); - return bodySearchIndex >= 0 ? bodySearchIndex : metadataSearchIndex; + // A replicated request has no local search index. The metadata fallback is based on the + // WAL entry offset, which also counts fragments and cannot assign a local index to it. + return bodySearchIndex; } private void startPending( From 4422f4cdabedde8c2e925f3f067d76056733e6fc Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 9 Oct 2026 15:43:53 +0800 Subject: [PATCH 6/8] Remove speculative WAL gap recovery from subscription fix --- .../iotdb/db/i18n/DataNodePipeMessages.java | 9 +- .../iotdb/db/i18n/DataNodePipeMessages.java | 9 +- .../consensus/ConsensusPrefetchingQueue.java | 90 +---- .../broker/consensus/ProgressWALIterator.java | 41 +-- .../ConsensusPrefetchingQueueTest.java | 315 ------------------ .../consensus/ProgressWALIteratorTest.java | 111 +----- 6 files changed, 25 insertions(+), 550 deletions(-) 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 2f680887368b7..5e92afb5f9cf0 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 @@ -2202,8 +2202,6 @@ private DataNodePipeMessages() {} "ProgressWALIterator: failed to open near-live WAL file {}, retrying without blacklisting"; public static final String PIPE_LOG_PROGRESSWALITERATOR_ERROR_READING_WAL_2DB46D41 = "ProgressWALIterator: error reading WAL"; - public static final String MESSAGE_PREVIOUS_LIVE_WAL_FILE_VERSION_ARG_IS_ABSENT_FROM_CURRENT_SCAN_RETRYING_5484DEF8 = - "previous live WAL file version %s is absent from current scan; retrying"; public static final String PIPE_LOG_PROGRESSWALITERATOR_FAILED_TO_PROCESS_WAL_FILE_SKIPPING_REMAINING_ENTRIES_093F14A9 = "ProgressWALIterator: failed to process WAL file {}, skipping remaining entries"; @@ -2219,11 +2217,8 @@ private DataNodePipeMessages() {} + "firstFile={}, lastFile={}, firstError={}; historical subscription data in these " + "files cannot be replayed"; public static final String - LOG_ARG_WAL_REPLAY_OBSERVED_GAP_ARG_ARG_BEFORE_SEARCHINDEX_ARG_REFRESHING_WAL_AND_RETRYING_0669FB51 = - "{}: WAL replay observed gap [{}, {}) before searchIndex {}; refreshing WAL and retrying"; - public static final String - LOG_ARG_WAL_GAP_ARG_ARG_PERSISTS_AFTER_REFRESH_SKIPPING_ARG_TOTAL_SKIPPED_ARG_16FE0680 = - "{}: WAL gap [{}, {}) persists after refresh; skipping {}, total skipped {}"; + LOG_ARG_WAL_GAP_ARG_ARG_SKIPPED_ARG_TOTAL_SKIPPED_ARG_CAUSE_UNDETERMINED_D4E6491A = + "{}: WAL gap [{}, {}), skipped {}, total skipped {}; cause undetermined"; public static final String PIPE_LOG_PIPE_TERMINATE_EVENT_COMMITTED_FOR_HISTORICAL_TRANSFER_CREATIONTIME_9B807B28 = "Pipe {}@{}: terminate event committed for historical transfer. creationTime: {}, " + "shouldMark: {}. {}"; diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java index 8adb0976fcc9f..14bc9cd4b266b 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 @@ -2045,8 +2045,6 @@ private DataNodePipeMessages() {} "ProgressWALIterator：打开 near-live WAL 文件 {} 失败，不加入黑名单并重试"; public static final String PIPE_LOG_PROGRESSWALITERATOR_ERROR_READING_WAL_2DB46D41 = "ProgressWALIterator：读取 WAL 出错"; - public static final String MESSAGE_PREVIOUS_LIVE_WAL_FILE_VERSION_ARG_IS_ABSENT_FROM_CURRENT_SCAN_RETRYING_5484DEF8 = - "上一个活跃 WAL 文件版本 %s 未出现在本次扫描结果中；稍后重试"; public static final String PIPE_LOG_PROGRESSWALITERATOR_FAILED_TO_PROCESS_WAL_FILE_SKIPPING_REMAINING_ENTRIES_093F14A9 = "ProgressWALIterator：处理 WAL 文件 {} 失败，跳过该文件的剩余条目"; @@ -2060,11 +2058,8 @@ private DataNodePipeMessages() {} "ProgressWALIterator：跳过了 {} 个无法读取的保留 WAL 文件，directory={}，firstFile={}，" + "lastFile={}，firstError={}；这些文件中的历史订阅数据无法重放"; public static final String - LOG_ARG_WAL_REPLAY_OBSERVED_GAP_ARG_ARG_BEFORE_SEARCHINDEX_ARG_REFRESHING_WAL_AND_RETRYING_0669FB51 = - "{}：WAL 回放观察到缺口 [{}, {})，下一个可见 searchIndex={}；正在刷新 WAL 并重试"; - public static final String - LOG_ARG_WAL_GAP_ARG_ARG_PERSISTS_AFTER_REFRESH_SKIPPING_ARG_TOTAL_SKIPPED_ARG_16FE0680 = - "{}：WAL 缺口 [{}, {}) 在刷新后仍存在；跳过 {} 条，累计跳过 {} 条"; + LOG_ARG_WAL_GAP_ARG_ARG_SKIPPED_ARG_TOTAL_SKIPPED_ARG_CAUSE_UNDETERMINED_D4E6491A = + "{}：WAL 缺口 [{}, {})，跳过 {} 条，累计跳过 {} 条；原因尚未确定"; public static final String PIPE_LOG_PIPE_TERMINATE_EVENT_COMMITTED_FOR_HISTORICAL_TRANSFER_CREATIONTIME_9B807B28 = "Pipe {}@{}：历史传输的终止事件已提交。creationTime：{}，shouldMark：{}。{}"; public static final String PIPE_LOG_PIPE_HISTORICAL_SOURCE_HAS_SUPPLIED_ALL_EVENTS_EMITTING_8B58DE19 = diff --git a/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 83252aaad6bbd..925c74c82f2fc 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 @@ -297,9 +297,6 @@ public class ConsensusPrefetchingQueue { private volatile long lastWalGapWaitLogTimeMs = 0L; - /** Search index whose WAL visibility gap is being retried after forcing a WAL refresh. */ - private volatile long walGapRetryExpectedSearchIndex = Long.MIN_VALUE; - /** Fallback committed region progress from local persisted state. */ private final RegionProgress fallbackCommittedRegionProgress; @@ -840,7 +837,6 @@ private boolean initPrefetchUnderInitializationLock(final RegionProgress regionP // readers without WAL support. this.subscriptionWALIterator = createSubscriptionWALIterator(resolvedStart.getStartSearchIndex()); - resetWalGapRetryState(); this.prefetchInitialized = true; this.observedSeekGeneration = seekGeneration.get(); discardBatch(this.lingerBatch); @@ -1521,9 +1517,6 @@ public PrefetchRoundResult drivePrefetchOnce() { if (batch.isEmpty() && lingerBatch.isEmpty()) { final MaterializationResult walResult = tryCatchUpFromWAL(observedSeekGeneration); - if (walResult == MaterializationResult.WAL_GAP) { - return PrefetchRoundResult.rescheduleAfter(WAL_GAP_RETRY_SLEEP_MS); - } if (walResult == MaterializationResult.MEMORY_BLOCKED) { blockRealtimeAdmission(); return PrefetchRoundResult.rescheduleAfter(MEMORY_RETRY_SLEEP_MS); @@ -1939,9 +1932,7 @@ private MaterializationResult tryCatchUpFromWAL(final long expectedSeekGeneratio pumpFromSubscriptionWAL( batchState, expectedSeekGeneration, maxWalEntries, maxTablets, maxBatchBytes); if (materializationResult != MaterializationResult.SUCCESS) { - if ((materializationResult == MaterializationResult.MEMORY_BLOCKED - || materializationResult == MaterializationResult.WAL_GAP) - && !batchState.isEmpty()) { + if (materializationResult == MaterializationResult.MEMORY_BLOCKED && !batchState.isEmpty()) { if (!flushBatch(batchState, expectedSeekGeneration)) { discardBatch(batchState); return MaterializationResult.STALE; @@ -1984,11 +1975,6 @@ private MaterializationResult pumpFromSubscriptionWAL( if (isBeforeLocalCursor(walEntry)) { continue; } - final MaterializationResult continuityResult = - validateWalReplayContinuity(walEntry, expectedSeekGeneration); - if (continuityResult != MaterializationResult.SUCCESS) { - return continuityResult; - } if (shouldSkipForRecoveryProgress(walEntry)) { advanceWalReplayCursorIfPresent(walEntry); continue; @@ -2058,7 +2044,7 @@ private void advanceWalReplayCursorIfPresent(final IndexedConsensusRequest reque final long totalSkippedEntries = walGapSkippedEntries.addAndGet(skippedEntries); LOGGER.warn( DataNodePipeMessages - .LOG_ARG_WAL_GAP_ARG_ARG_PERSISTS_AFTER_REFRESH_SKIPPING_ARG_TOTAL_SKIPPED_ARG_16FE0680, + .LOG_ARG_WAL_GAP_ARG_ARG_SKIPPED_ARG_TOTAL_SKIPPED_ARG_CAUSE_UNDETERMINED_D4E6491A, this, expectedSearchIndex, actualSearchIndex, @@ -2068,65 +2054,6 @@ private void advanceWalReplayCursorIfPresent(final IndexedConsensusRequest reque nextExpectedSearchIndex.set(actualSearchIndex + 1); } - private MaterializationResult validateWalReplayContinuity( - final IndexedConsensusRequest request, final long expectedSeekGeneration) { - if (seekGeneration.get() != expectedSeekGeneration || isClosed) { - return MaterializationResult.STALE; - } - if (!hasLocalSearchIndex(request)) { - return MaterializationResult.SUCCESS; - } - - final long actualSearchIndex = request.getSearchIndex(); - final long expectedSearchIndex = nextExpectedSearchIndex.get(); - if (actualSearchIndex <= expectedSearchIndex) { - if (actualSearchIndex == expectedSearchIndex - && walGapRetryExpectedSearchIndex == expectedSearchIndex) { - walGapRetryExpectedSearchIndex = Long.MIN_VALUE; - pendingWalGapRetryRequested = false; - } - return MaterializationResult.SUCCESS; - } - - if (walGapRetryExpectedSearchIndex != expectedSearchIndex) { - if (seekGeneration.get() != expectedSeekGeneration || isClosed) { - return MaterializationResult.STALE; - } - walGapRetryExpectedSearchIndex = expectedSearchIndex; - LOGGER.warn( - DataNodePipeMessages - .LOG_ARG_WAL_REPLAY_OBSERVED_GAP_ARG_ARG_BEFORE_SEARCHINDEX_ARG_REFRESHING_WAL_AND_RETRYING_0669FB51, - this, - expectedSearchIndex, - actualSearchIndex, - actualSearchIndex); - if (consensusReqReader instanceof WALNode) { - ((WALNode) consensusReqReader).rollWALFile(); - } - if (seekGeneration.get() != expectedSeekGeneration || isClosed) { - return MaterializationResult.STALE; - } - resetSubscriptionWALPosition(expectedSearchIndex); - onWalGapRetryScheduled(); - pendingWalGapRetryRequested = true; - return MaterializationResult.WAL_GAP; - } - - if (seekGeneration.get() != expectedSeekGeneration || isClosed) { - return MaterializationResult.STALE; - } - walGapRetryExpectedSearchIndex = Long.MIN_VALUE; - pendingWalGapRetryRequested = false; - return MaterializationResult.SUCCESS; - } - - private void resetWalGapRetryState() { - pendingWalGapRetryRequested = false; - walGapWaitStartTimeMs = 0L; - lastWalGapWaitLogTimeMs = 0L; - walGapRetryExpectedSearchIndex = Long.MIN_VALUE; - } - private void ensureSubscriptionWalReadable() { if (Objects.isNull(subscriptionWALIterator)) { return; @@ -3127,7 +3054,9 @@ public void cleanUp() { reconcileRetainedTabletMemoryAfterCleanup(); memoryBlockedEntryBytes = -1L; resetBatchWriterProgress(); - resetWalGapRetryState(); + pendingWalGapRetryRequested = false; + walGapWaitStartTimeMs = 0L; + lastWalGapWaitLogTimeMs = 0L; pendingSubscriptionWalResetSearchIndex = Long.MIN_VALUE; pendingSubscriptionWalResetGeneration = Long.MIN_VALUE; closeSubscriptionWALIterator(); @@ -3438,7 +3367,9 @@ private void applySeekResetUnderWriteLock(final PendingSeekRequest request) { reconcileRetainedTabletMemoryAfterCleanup(); resetBatchWriterProgress(); observedSeekGeneration = seekGeneration.get(); - resetWalGapRetryState(); + pendingWalGapRetryRequested = false; + walGapWaitStartTimeMs = 0L; + lastWalGapWaitLogTimeMs = 0L; // 5. Reset commit state to the writer progress immediately before the first re-delivered // entry so seek/rebind resumes from the intended frontier. @@ -3539,7 +3470,6 @@ static long findReplayRetainedMinVersionId( return 0L; } - WALFileUtils.ascSortByVersionId(walFiles); final int replayFileIndex = Math.max(0, WALFileUtils.binarySearchFileBySearchIndex(walFiles, nextExpectedSearchIndex)); return WALFileUtils.parseVersionId(walFiles[replayFileIndex].getName()); @@ -4156,7 +4086,9 @@ private void setActiveUnderRuntimeLock(final boolean active) { memoryBlockedEntryBytes = -1L; prefetchInitialized = false; observedSeekGeneration = seekGeneration.get(); - resetWalGapRetryState(); + pendingWalGapRetryRequested = false; + walGapWaitStartTimeMs = 0L; + lastWalGapWaitLogTimeMs = 0L; pendingSubscriptionWalResetSearchIndex = Long.MIN_VALUE; pendingSubscriptionWalResetGeneration = Long.MIN_VALUE; closeSubscriptionWALIterator(); 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 c56b522a305aa..6572473cb58c2 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 @@ -79,7 +79,6 @@ public interface WriterProgressCoverage { private int currentFileIndex = -1; private ProgressWALReader currentReader; private long currentReaderVersionId = -1L; - private long pendingNearLiveWalVersionId = -1L; private boolean currentReaderUsesLiveSnapshot = false; private int consumedEntryCountInCurrentFile = 0; private final Set skippedBrokenWalVersionIds = new HashSet<>(); @@ -391,7 +390,6 @@ public void close() throws IOException { pendingRequests.clear(); pendingSearchIndex = Long.MIN_VALUE; pendingLocalSeq = Long.MIN_VALUE; - pendingNearLiveWalVersionId = -1L; lastError = null; incompleteScan = false; incompleteScanDetail = null; @@ -491,27 +489,9 @@ private IndexedConsensusRequest advance() throws IOException { } private boolean openNextReader() throws IOException { - if (pendingNearLiveWalVersionId >= 0L) { - refresh(); - final int retryFileIndex = findFileIndexByVersion(pendingNearLiveWalVersionId); - if (retryFileIndex < 0) { - return false; - } - currentFileIndex = retryFileIndex - 1; - pendingNearLiveWalVersionId = -1L; - } while (++currentFileIndex < walFiles.length) { - final int candidateFileIndex = currentFileIndex; - final long candidateVersionId = walFileVersionIds[candidateFileIndex]; - try { - if (openReaderAtIndex(candidateFileIndex, 0)) { - return true; - } - } catch (final IOException e) { - // Retry this version on the next scan instead of jumping to its successor when the - // discovered path cannot be opened. - pendingNearLiveWalVersionId = candidateVersionId; - throw e; + if (openReaderAtIndex(currentFileIndex, 0)) { + return true; } } return false; @@ -540,15 +520,7 @@ private boolean reopenLiveSnapshotReader() throws IOException { final int previousFileIndex = findFileIndexByVersion(currentReaderVersionId); if (previousFileIndex < 0) { - // The old live version is absent from this scan. Its last snapshot does not prove that every - // entry in the sealed file was consumed, so wait for the old version to reappear. - markIncompleteScan( - String.format( - DataNodePipeMessages - .MESSAGE_PREVIOUS_LIVE_WAL_FILE_VERSION_ARG_IS_ABSENT_FROM_CURRENT_SCAN_RETRYING_5484DEF8, - currentReaderVersionId), - null); - return false; + return openFirstReaderAfterVersion(currentReaderVersionId); } if (openReaderAtIndex(previousFileIndex, consumedEntryCountInCurrentFile)) { return true; @@ -618,7 +590,12 @@ private boolean openReaderAtIndex( } } } - throw e; + markIncompleteScan( + String.format( + "failed to open near-live WAL file %s while replay scan was still in progress", + walFile.getName()), + e); + return false; } recordSkippedBrokenWalFile(versionId, walFile, e); return false; 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 6a5f513e9b85f..9edaef2fb7ee1 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 @@ -79,7 +79,6 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; -import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.locks.ReentrantReadWriteLock; import java.util.function.BooleanSupplier; @@ -926,7 +925,6 @@ public void testFilteredEmptyEntryAdvancesProgressWithoutEvent() throws Exceptio 1L, 1L, true); - queue.setSubscriptionMemoryManager(new SubscriptionMemoryManager(16L * 1024 * 1024)); final IndexedConsensusRequest request = new IndexedConsensusRequest( @@ -990,7 +988,6 @@ public void testFilteredEmptyEntryDoesNotAdvanceAcrossUncommittedData() throws E 1L, 1L, true); - queue.setSubscriptionMemoryManager(new SubscriptionMemoryManager(16L * 1024 * 1024)); final IndexedConsensusRequest dataRequest = new IndexedConsensusRequest( @@ -1185,7 +1182,6 @@ protected ReplayLocateDecision locateReplayStartForRegionProgress( walNode.getCurrentSearchIndex(), regionProgress, "empty sealed WAL"); } }; - queue.setSubscriptionMemoryManager(new SubscriptionMemoryManager(16L * 1024 * 1024)); pendingEntries(queue) .offer( @@ -1447,92 +1443,6 @@ ProgressWALReader openProgressWALReader(final File walFile) throws IOException { } } - @Test - public void testWalReplayRetriesGapWithoutSkippingEntries() throws Exception { - final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); - final File systemDir = temporaryFolder.newFolder("wal-replay-gap-counter"); - ConsensusPrefetchingQueue queue = null; - try { - final DataRegionId regionId = new DataRegionId(9); - final WALNode walNode = mock(WALNode.class); - when(walNode.getCurrentSearchIndex()).thenReturn(4L); - when(walNode.getLogDirectory()).thenReturn(systemDir); - final IoTConsensusServerImpl serverImpl = mock(IoTConsensusServerImpl.class); - when(serverImpl.getConsensusReqReader()).thenReturn(walNode); - when(serverImpl.getWriterSafeFrontierTracker()).thenReturn(new WriterSafeFrontierTracker()); - - final AtomicInteger conversionCount = new AtomicInteger(); - final ConsensusLogToTabletConverter converter = mock(ConsensusLogToTabletConverter.class); - when(converter.convert(any())) - .thenAnswer( - ignored -> { - conversionCount.incrementAndGet(); - return Collections.singletonList(createTablet()); - }); - when(converter.getDatabaseName()).thenReturn("db"); - - final Iterator initiallyVisibleEntries = - Arrays.asList(createRequest(1L), createRequest(4L)).iterator(); - final ProgressWALIterator initialIterator = mock(ProgressWALIterator.class); - when(initialIterator.hasNext()).thenAnswer(ignored -> initiallyVisibleEntries.hasNext()); - when(initialIterator.next()).thenAnswer(ignored -> initiallyVisibleEntries.next()); - final Iterator refreshedEntries = - Arrays.asList(createRequest(2L), createRequest(3L), createRequest(4L)).iterator(); - final ProgressWALIterator refreshedIterator = mock(ProgressWALIterator.class); - when(refreshedIterator.hasNext()).thenAnswer(ignored -> refreshedEntries.hasNext()); - when(refreshedIterator.next()).thenAnswer(ignored -> refreshedEntries.next()); - final AtomicInteger iteratorCreationCount = new AtomicInteger(); - - queue = - new ConsensusPrefetchingQueue( - "consumerGroup", - "topic", - TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE, - regionId, - serverImpl, - new SubscriptionWalRetentionPolicy( - "topic", - SubscriptionWalRetentionPolicy.UNBOUNDED, - SubscriptionWalRetentionPolicy.UNBOUNDED), - converter, - newCommitManager(systemDir), - new RegionProgress(Collections.emptyMap()), - 1L, - 1L, - true) { - @Override - protected ProgressWALIterator createSubscriptionWALIterator( - final long startSearchIndex) { - return iteratorCreationCount.getAndIncrement() == 0 - ? initialIterator - : refreshedIterator; - } - }; - queue.setSubscriptionMemoryManager(new SubscriptionMemoryManager(16L * 1024 * 1024)); - - assertNull(queue.poll("consumer")); - queue.drivePrefetchOnce(); - - assertEquals(1L, queue.getWalPathAcceptedEntries()); - assertEquals(1, conversionCount.get()); - assertEquals(0L, queue.getWalGapSkippedEntries()); - assertEquals(2L, queue.getCurrentReadSearchIndex()); - verify(walNode).rollWALFile(); - - queue.drivePrefetchOnce(); - - assertEquals(4L, queue.getWalPathAcceptedEntries()); - assertEquals(4, conversionCount.get()); - assertEquals(0L, queue.getWalGapSkippedEntries()); - assertEquals(5L, queue.getCurrentReadSearchIndex()); - } finally { - if (queue != null) { - queue.close(); - } - IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir); - } - } - @Test public void testPendingGapReplayHonorsPerRoundWalEntryLimit() throws Exception { final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); @@ -1658,24 +1568,11 @@ public void testWalReplayCountsOnlyUnavailableSearchIndexes() throws Exception { ConsensusPrefetchingQueue.class.getDeclaredMethod( "advanceWalReplayCursorIfPresent", IndexedConsensusRequest.class); advanceWalReplayCursorIfPresent.setAccessible(true); - final Method validateWalReplayContinuity = - ConsensusPrefetchingQueue.class.getDeclaredMethod( - "validateWalReplayContinuity", IndexedConsensusRequest.class, long.class); - validateWalReplayContinuity.setAccessible(true); advanceWalReplayCursorIfPresent.invoke(queue, createRequest(1L)); assertEquals(0L, queue.getWalGapSkippedEntries()); assertEquals(2L, queue.getCurrentReadSearchIndex()); - assertEquals( - "WAL_GAP", validateWalReplayContinuity.invoke(queue, createRequest(4L), 0L).toString()); - assertEquals(0L, queue.getWalGapSkippedEntries()); - assertEquals(2L, queue.getCurrentReadSearchIndex()); - - assertEquals( - "SUCCESS", validateWalReplayContinuity.invoke(queue, createRequest(4L), 0L).toString()); - assertEquals(0L, queue.getWalGapSkippedEntries()); - assertEquals(2L, queue.getCurrentReadSearchIndex()); advanceWalReplayCursorIfPresent.invoke(queue, createRequest(4L)); assertEquals(2L, queue.getWalGapSkippedEntries()); assertEquals(5L, queue.getCurrentReadSearchIndex()); @@ -1775,181 +1672,6 @@ protected ProgressWALIterator createSubscriptionWALIterator( } } - @Test - public void testWalReplayContinuesWhenGapRemainsUnavailable() throws Exception { - final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); - final File systemDir = temporaryFolder.newFolder("wal-replay-unrecoverable-gap"); - ConsensusPrefetchingQueue queue = null; - try { - final DataRegionId regionId = new DataRegionId(10); - final FakeConsensusReqReader reader = new FakeConsensusReqReader(); - reader.currentSearchIndex = 4L; - final IoTConsensusServerImpl serverImpl = mock(IoTConsensusServerImpl.class); - when(serverImpl.getConsensusReqReader()).thenReturn(reader); - when(serverImpl.getWriterSafeFrontierTracker()).thenReturn(new WriterSafeFrontierTracker()); - - final AtomicInteger conversionCount = new AtomicInteger(); - final ConsensusLogToTabletConverter converter = mock(ConsensusLogToTabletConverter.class); - when(converter.convert(any())) - .thenAnswer( - ignored -> { - conversionCount.incrementAndGet(); - return Collections.singletonList(createTablet()); - }); - when(converter.getDatabaseName()).thenReturn("db"); - - final Iterator initiallyVisibleEntries = - Arrays.asList(createRequest(1L), createRequest(4L)).iterator(); - final ProgressWALIterator initialIterator = mock(ProgressWALIterator.class); - when(initialIterator.hasNext()).thenAnswer(ignored -> initiallyVisibleEntries.hasNext()); - when(initialIterator.next()).thenAnswer(ignored -> initiallyVisibleEntries.next()); - final Iterator refreshedEntries = - Collections.singletonList(createRequest(4L)).iterator(); - final ProgressWALIterator refreshedIterator = mock(ProgressWALIterator.class); - when(refreshedIterator.hasNext()).thenAnswer(ignored -> refreshedEntries.hasNext()); - when(refreshedIterator.next()).thenAnswer(ignored -> refreshedEntries.next()); - final AtomicInteger iteratorCreationCount = new AtomicInteger(); - - queue = - new ConsensusPrefetchingQueue( - "consumerGroup", - "topic", - TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE, - regionId, - serverImpl, - new SubscriptionWalRetentionPolicy( - "topic", - SubscriptionWalRetentionPolicy.UNBOUNDED, - SubscriptionWalRetentionPolicy.UNBOUNDED), - converter, - newCommitManager(systemDir), - new RegionProgress(Collections.emptyMap()), - 1L, - 1L, - true) { - @Override - protected ProgressWALIterator createSubscriptionWALIterator( - final long startSearchIndex) { - return iteratorCreationCount.getAndIncrement() == 0 - ? initialIterator - : refreshedIterator; - } - }; - queue.setSubscriptionMemoryManager(new SubscriptionMemoryManager(16L * 1024 * 1024)); - - assertNull(queue.poll("consumer")); - queue.drivePrefetchOnce(); - queue.drivePrefetchOnce(); - - assertEquals(2L, queue.getWalPathAcceptedEntries()); - assertEquals(2, conversionCount.get()); - assertEquals(2L, queue.getWalGapSkippedEntries()); - assertEquals(5L, queue.getCurrentReadSearchIndex()); - - final SubscriptionEvent deliveredEvent = queue.poll("consumer"); - assertNotNull(deliveredEvent); - assertEquals( - SubscriptionPollResponseType.TABLETS.getType(), - deliveredEvent.getCurrentResponse().getResponseType()); - } finally { - if (queue != null) { - queue.close(); - } - IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir); - } - } - - @Test - public void testSeekClearsPendingWalGapRetryAndResumesAtTarget() throws Exception { - final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); - final File systemDir = temporaryFolder.newFolder("wal-gap-seek-recovery"); - ConsensusPrefetchingQueue queue = null; - try { - final FakeConsensusReqReader reader = new FakeConsensusReqReader(); - reader.currentSearchIndex = 7L; - final IoTConsensusServerImpl serverImpl = mock(IoTConsensusServerImpl.class); - when(serverImpl.getConsensusReqReader()).thenReturn(reader); - when(serverImpl.getWriterSafeFrontierTracker()).thenReturn(new WriterSafeFrontierTracker()); - - final ConsensusLogToTabletConverter converter = mock(ConsensusLogToTabletConverter.class); - when(converter.convert(any())).thenReturn(Collections.singletonList(createTablet())); - when(converter.getDatabaseName()).thenReturn("db"); - - final AtomicInteger iteratorCreationCount = new AtomicInteger(); - queue = - new ConsensusPrefetchingQueue( - "consumerGroup", - "topic", - TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE, - new DataRegionId(14), - serverImpl, - new SubscriptionWalRetentionPolicy( - "topic", - SubscriptionWalRetentionPolicy.UNBOUNDED, - SubscriptionWalRetentionPolicy.UNBOUNDED), - converter, - newCommitManager(systemDir), - new RegionProgress(Collections.emptyMap()), - 1L, - 1L, - true) { - @Override - protected ProgressWALIterator createSubscriptionWALIterator( - final long startSearchIndex) { - final List entries; - switch (iteratorCreationCount.getAndIncrement()) { - case 0: - entries = Arrays.asList(createRequest(1L), createRequest(4L)); - break; - case 1: - entries = Collections.singletonList(createRequest(4L)); - break; - default: - assertEquals(7L, startSearchIndex); - entries = Collections.singletonList(createRequest(7L)); - } - final Iterator walEntries = entries.iterator(); - final ProgressWALIterator iterator = mock(ProgressWALIterator.class); - when(iterator.hasNext()).thenAnswer(ignored -> walEntries.hasNext()); - when(iterator.next()).thenAnswer(ignored -> walEntries.next()); - return iterator; - } - }; - queue.setSubscriptionMemoryManager(new SubscriptionMemoryManager(16L * 1024 * 1024)); - - assertNull(queue.poll("consumer")); - queue.drivePrefetchOnce(); - - assertEquals(2L, queue.getCurrentReadSearchIndex()); - - applySeekForTest(queue, 7L); - - assertEquals(7L, queue.getCurrentReadSearchIndex()); - assertEquals(3L, queue.getProgressStatus()); - final Method validateWalReplayContinuity = - ConsensusPrefetchingQueue.class.getDeclaredMethod( - "validateWalReplayContinuity", IndexedConsensusRequest.class, long.class); - validateWalReplayContinuity.setAccessible(true); - assertEquals( - "STALE", validateWalReplayContinuity.invoke(queue, createRequest(4L), 0L).toString()); - assertEquals(7L, queue.getCurrentReadSearchIndex()); - assertNull(queue.poll("consumer")); - queue.drivePrefetchOnce(); - - assertEquals(8L, queue.getCurrentReadSearchIndex()); - final SubscriptionEvent resumedEvent = queue.poll("consumer"); - assertNotNull(resumedEvent); - assertEquals( - SubscriptionPollResponseType.TABLETS.getType(), - resumedEvent.getCurrentResponse().getResponseType()); - } finally { - if (queue != null) { - queue.close(); - } - IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir); - } - } - @Test public void testActivationRetriesUntilConfigNodeProgressIsExplicitlyAvailable() throws Exception { final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); @@ -2099,7 +1821,6 @@ public void testTabletMemoryReleasedAfterAck() throws Exception { 1L, 1L, true); - queue.setSubscriptionMemoryManager(new SubscriptionMemoryManager(16L * 1024 * 1024)); final IndexedConsensusRequest dataRequest = new IndexedConsensusRequest( @@ -2168,7 +1889,6 @@ public void testLateAckDoesNotWaitForPrefetchReadLock() throws Exception { 1L, 1L, true); - queue.setSubscriptionMemoryManager(new SubscriptionMemoryManager(16L * 1024 * 1024)); final IndexedConsensusRequest dataRequest = new IndexedConsensusRequest( @@ -2977,7 +2697,6 @@ public void testRequeueDoesNotIncrementNackCount() throws Exception { 1L, 1L, true); - queue.setSubscriptionMemoryManager(new SubscriptionMemoryManager(16L * 1024 * 1024)); reader.currentSearchIndex = 2L; assertTrue(pendingEntries(queue).offer(createRequest(1L))); @@ -3110,40 +2829,6 @@ private static void setPrefetchBinding( subtaskField.set(queue, subtask); } - private static void applySeekForTest( - final ConsensusPrefetchingQueue queue, final long targetSearchIndex) throws Exception { - final Field seekGenerationField = - ConsensusPrefetchingQueue.class.getDeclaredField("seekGeneration"); - seekGenerationField.setAccessible(true); - final AtomicLong seekGeneration = (AtomicLong) seekGenerationField.get(queue); - final long previousSeekGeneration = seekGeneration.get(); - final long targetSeekGeneration = seekGeneration.incrementAndGet(); - - final Class pendingSeekRequestClass = - Class.forName(ConsensusPrefetchingQueue.class.getName() + "$PendingSeekRequest"); - final Constructor constructor = - pendingSeekRequestClass.getDeclaredConstructor( - long.class, RegionProgress.class, String.class, boolean.class, long.class, long.class); - constructor.setAccessible(true); - final Object pendingSeekRequest = - constructor.newInstance( - targetSearchIndex, - new RegionProgress(Collections.emptyMap()), - "test", - true, - previousSeekGeneration, - targetSeekGeneration); - final Field pendingSeekRequestField = - ConsensusPrefetchingQueue.class.getDeclaredField("pendingSeekRequest"); - pendingSeekRequestField.setAccessible(true); - pendingSeekRequestField.set(queue, pendingSeekRequest); - - final Method applyPendingSeekRequest = - ConsensusPrefetchingQueue.class.getDeclaredMethod("applyPendingSeekRequestIfNecessary"); - applyPendingSeekRequest.setAccessible(true); - assertTrue((Boolean) applyPendingSeekRequest.invoke(queue)); - } - private static ProgressWALIterator subscriptionWalIterator(final ConsensusPrefetchingQueue queue) throws Exception { final Field field = ConsensusPrefetchingQueue.class.getDeclaredField("subscriptionWALIterator"); 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 941c7f49321bd..45bff39d6ba78 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 @@ -38,7 +38,6 @@ import java.nio.ByteBuffer; import java.nio.file.Files; import java.nio.file.Path; -import java.util.concurrent.atomic.AtomicLong; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; @@ -453,7 +452,7 @@ public void testIteratorMarksIncompleteScanWhenNearLiveWalCannotBeOpened() throw assertFalse(iterator.hasNext()); assertTrue(iterator.hasIncompleteScan()); assertTrue(iterator.hasReadError()); - assertTrue(iterator.getLastError() != null); + assertTrue(iterator.getIncompleteScanDetail().contains("near-live WAL file")); } } finally { Files.deleteIfExists(brokenLiveWal.toPath()); @@ -461,114 +460,6 @@ public void testIteratorMarksIncompleteScanWhenNearLiveWalCannotBeOpened() throw } } - @Test - public void testNearLiveWalOpenFailureDoesNotSkipToSuccessor() throws Exception { - final Path dir = Files.createTempDirectory("progress-wal-iterator-near-live-retry"); - final File firstWal = - dir.resolve(WALFileUtils.getLogFileName(0, 0, WALFileStatus.CONTAINS_SEARCH_INDEX)) - .toFile(); - final File temporarilyUnavailableWal = - dir.resolve(WALFileUtils.getLogFileName(1, 1, WALFileStatus.CONTAINS_SEARCH_INDEX)) - .toFile(); - final File successorWal = - dir.resolve(WALFileUtils.getLogFileName(2, 2, WALFileStatus.CONTAINS_SEARCH_INDEX)) - .toFile(); - - try { - try (WALWriter writer = new WALWriter(firstWal, WALFileVersion.V3)) { - writer.write(searchableEntry(1L), singleEntryMeta(19, 1L, 1L, 100L, 7, 1L)); - } - Files.write(temporarilyUnavailableWal.toPath(), new byte[128]); - try (WALWriter writer = new WALWriter(successorWal, WALFileVersion.V3)) { - writer.write(searchableEntry(3L), singleEntryMeta(19, 3L, 1L, 300L, 7, 3L)); - } - - final WALNode walNode = mock(WALNode.class); - when(walNode.getLogDirectory()).thenReturn(dir.toFile()); - when(walNode.getCurrentWALFileVersion()).thenReturn(2L); - when(walNode.getCurrentWALMetaDataSnapshot()) - .thenReturn(singleEntryMeta(19, 3L, 1L, 300L, 7, 3L)); - - try (ProgressWALIterator iterator = new ProgressWALIterator(walNode, 1L)) { - assertTrue(iterator.hasNext()); - assertEquals(1L, iterator.next().getSearchIndex()); - assertFalse(iterator.hasNext()); - assertTrue(iterator.hasReadError()); - - Files.delete(temporarilyUnavailableWal.toPath()); - assertFalse(iterator.hasNext()); - try (WALWriter writer = new WALWriter(temporarilyUnavailableWal, WALFileVersion.V3)) { - writer.write(searchableEntry(2L), singleEntryMeta(19, 2L, 1L, 200L, 7, 2L)); - } - - assertTrue(iterator.hasNext()); - assertEquals(2L, iterator.next().getSearchIndex()); - assertFalse(iterator.hasReadError()); - assertTrue(iterator.hasNext()); - assertEquals(3L, iterator.next().getSearchIndex()); - } - } finally { - Files.deleteIfExists(firstWal.toPath()); - Files.deleteIfExists(temporarilyUnavailableWal.toPath()); - Files.deleteIfExists(successorWal.toPath()); - Files.deleteIfExists(dir); - } - } - - @Test - public void testTemporarilyAbsentLiveWalMustReappearBeforeReadingSuccessor() throws Exception { - final Path dir = Files.createTempDirectory("progress-wal-iterator-absent-version"); - final File firstWal = - dir.resolve(WALFileUtils.getLogFileName(0, 0, WALFileStatus.CONTAINS_SEARCH_INDEX)) - .toFile(); - final Path temporarilyHiddenWal = dir.resolve("temporarily-hidden"); - final File successorWal = - dir.resolve(WALFileUtils.getLogFileName(1, 2, WALFileStatus.CONTAINS_SEARCH_INDEX)) - .toFile(); - - try { - try (WALWriter writer = new WALWriter(firstWal, WALFileVersion.V3)) { - writer.write(searchableEntry(1L), singleEntryMeta(19, 1L, 1L, 100L, 7, 1L)); - writer.write(searchableEntry(2L), singleEntryMeta(19, 2L, 1L, 200L, 7, 2L)); - } - - final AtomicLong currentWalVersion = new AtomicLong(0L); - final WALNode walNode = mock(WALNode.class); - when(walNode.getLogDirectory()).thenReturn(dir.toFile()); - when(walNode.getCurrentWALFileVersion()).thenAnswer(ignored -> currentWalVersion.get()); - when(walNode.getCurrentWALMetaDataSnapshot()) - .thenReturn(singleEntryMeta(19, 1L, 1L, 100L, 7, 1L)); - - try (ProgressWALIterator iterator = new ProgressWALIterator(walNode, 1L)) { - assertTrue(iterator.hasNext()); - assertEquals(1L, iterator.next().getSearchIndex()); - assertFalse(iterator.hasNext()); - - // Simulate a listing in which an unread version is absent. Normal rotation leaves the - // name of a WAL with search-index entries unchanged. - Files.move(firstWal.toPath(), temporarilyHiddenWal); - try (WALWriter writer = new WALWriter(successorWal, WALFileVersion.V3)) { - writer.write(searchableEntry(3L), singleEntryMeta(19, 3L, 1L, 300L, 7, 3L)); - } - currentWalVersion.set(1L); - assertFalse(iterator.hasNext()); - assertTrue(iterator.hasIncompleteScan()); - assertFalse(iterator.hasNext()); - - Files.move(temporarilyHiddenWal, firstWal.toPath()); - assertTrue(iterator.hasNext()); - assertEquals(2L, iterator.next().getSearchIndex()); - assertTrue(iterator.hasNext()); - assertEquals(3L, iterator.next().getSearchIndex()); - } - } finally { - Files.deleteIfExists(firstWal.toPath()); - Files.deleteIfExists(temporarilyHiddenWal); - Files.deleteIfExists(successorWal.toPath()); - Files.deleteIfExists(dir); - } - } - @Test public void testAdvanceToKeepsUncoveredFollowerRequestInCurrentFile() throws Exception { final Path dir = Files.createTempDirectory("progress-wal-iterator-current-file-advance"); From fb8599c2bca559472af6887a2588e238fce205c2 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 9 Oct 2026 15:54:00 +0800 Subject: [PATCH 7/8] Remove unproven seek boundary changes from subscription fix --- .../consensus/ConsensusPrefetchingQueue.java | 30 ++------- .../ConsensusPrefetchingQueueTest.java | 61 ------------------- 2 files changed, 6 insertions(+), 85 deletions(-) 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 925c74c82f2fc..8d2f0af7cec88 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 @@ -882,9 +882,7 @@ private ReplayLocateDecision resolveInitReplayStartDecision( case AT_END: final long resolvedReplayStartSearchIndex = replayTarget.getStatus() == ReplayLocateStatus.AT_END - ? Math.min( - nextSearchIndexAfter(replayTarget.getStartSearchIndex()), - fallbackTailSearchIndex) + ? Math.min(replayTarget.getStartSearchIndex(), fallbackTailSearchIndex) : replayTarget.getStartSearchIndex(); return new ReplayLocateDecision( replayTarget.getStatus(), @@ -2037,6 +2035,7 @@ private void advanceWalReplayCursorIfPresent(final IndexedConsensusRequest reque if (!hasLocalSearchIndex(request)) { return; } + final long actualSearchIndex = request.getSearchIndex(); final long expectedSearchIndex = nextExpectedSearchIndex.get(); if (actualSearchIndex > expectedSearchIndex) { @@ -3073,28 +3072,15 @@ public void cleanUp() { * old files have been reclaimed, the earliest available position may be later than 0. */ public void seekToBeginning() { - final RegionProgress beginningProgress = new RegionProgress(Collections.emptyMap()); - if (consensusReqReader instanceof WALNode) { - // The first retained WAL index can be greater than one after older files are reclaimed. - seekToRegionProgress(beginningProgress); - } else { - seekToResolvedPosition(FIRST_CONSENSUS_SEARCH_INDEX, beginningProgress, "beginning"); - } + seekToResolvedPosition(0L, new RegionProgress(Collections.emptyMap()), "beginning"); } /** * Seeks to the current WAL write position. After this, only newly written data will be consumed. */ public void seekToEnd() { - final RegionProgress tailProgress = computeTailRegionProgress(); seekToResolvedPosition( - nextSearchIndexAfter(consensusReqReader.getCurrentSearchIndex()), tailProgress, "end"); - } - - private static long nextSearchIndexAfter(final long searchIndex) { - return searchIndex == Long.MAX_VALUE - ? Long.MAX_VALUE - : Math.max(FIRST_CONSENSUS_SEARCH_INDEX, searchIndex + 1L); + consensusReqReader.getCurrentSearchIndex(), computeTailRegionProgress(), "end"); } public void seekToRegionProgress(final RegionProgress regionProgress) { @@ -3122,9 +3108,7 @@ public void seekToRegionProgress(final RegionProgress regionProgress) { replayTarget.getStatus(), replayTarget.getStartSearchIndex()); seekToResolvedPosition( - replayTarget.getStatus() == ReplayLocateStatus.AT_END - ? nextSearchIndexAfter(replayTarget.getStartSearchIndex()) - : replayTarget.getStartSearchIndex(), + replayTarget.getStartSearchIndex(), replayTarget.getRecoveryRegionProgress(), "regionProgress"); return; @@ -3165,9 +3149,7 @@ public void seekAfterRegionProgress(final RegionProgress regionProgress) { replayTarget.getStatus(), replayTarget.getStartSearchIndex()); seekToResolvedPosition( - replayTarget.getStatus() == ReplayLocateStatus.AT_END - ? nextSearchIndexAfter(replayTarget.getStartSearchIndex()) - : replayTarget.getStartSearchIndex(), + replayTarget.getStartSearchIndex(), replayTarget.getRecoveryRegionProgress(), "regionProgressAfter"); return; 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 9edaef2fb7ee1..0f321341401ed 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 @@ -1210,67 +1210,6 @@ protected ReplayLocateDecision locateReplayStartForRegionProgress( } } - @Test - public void testAtEndInitializationStartsAfterLastWalIndex() throws Exception { - final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); - final File systemDir = temporaryFolder.newFolder("atEndAfterTail"); - final File walDirectory = temporaryFolder.newFolder("atEndAfterTailWal"); - ConsensusPrefetchingQueue queue = null; - try { - final DataRegionId regionId = new DataRegionId(8); - final WALNode walNode = mock(WALNode.class); - when(walNode.getLogDirectory()).thenReturn(walDirectory); - when(walNode.getCurrentSearchIndex()).thenReturn(4L); - final IoTConsensusServerImpl serverImpl = mock(IoTConsensusServerImpl.class); - when(serverImpl.getConsensusReqReader()).thenReturn(walNode); - when(serverImpl.getWriterSafeFrontierTracker()).thenReturn(new WriterSafeFrontierTracker()); - final ConsensusLogToTabletConverter converter = mock(ConsensusLogToTabletConverter.class); - when(converter.convert(any())).thenReturn(Collections.singletonList(createTablet())); - when(converter.getDatabaseName()).thenReturn("db"); - final RegionProgress committedProgress = - new RegionProgress( - Collections.singletonMap( - new WriterId(regionId.toString(), 7), new WriterProgress(1000L, 2L))); - queue = - new ConsensusPrefetchingQueue( - "consumerGroup", - "topic", - TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE, - regionId, - serverImpl, - new SubscriptionWalRetentionPolicy( - "topic", - SubscriptionWalRetentionPolicy.UNBOUNDED, - SubscriptionWalRetentionPolicy.UNBOUNDED), - converter, - newCommitManager(systemDir), - committedProgress, - 5L, - 1L, - true) { - @Override - protected ReplayLocateDecision locateReplayStartForRegionProgress( - final RegionProgress regionProgress, final boolean seekAfter) { - return ReplayLocateDecision.atEnd( - walNode.getCurrentSearchIndex(), regionProgress, "empty sealed WAL"); - } - }; - queue.setSubscriptionMemoryManager(new SubscriptionMemoryManager(16L * 1024 * 1024)); - - assertNull(queue.poll("consumer")); - assertEquals(5L, queue.getCurrentReadSearchIndex()); - assertTrue(pendingEntries(queue).offer(createRequest(5L))); - queue.drivePrefetchOnce(); - assertEquals(6L, queue.getCurrentReadSearchIndex()); - assertEquals(1L, queue.getPendingPathAcceptedEntries()); - } finally { - if (queue != null) { - queue.close(); - } - IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir); - } - } - @Test public void testZeroPhysicalTimeUsesWriterLocalSequenceForReplayLookup() throws Exception { final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); From 7df1d044f8e84b27f988d5cd52f8924f589350b7 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 9 Oct 2026 16:18:23 +0800 Subject: [PATCH 8/8] test: cover replicated WAL index after fragmented local requests --- .../consensus/ProgressWALIteratorTest.java | 76 ++++++++++++++++++- 1 file changed, 73 insertions(+), 3 deletions(-) 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 45bff39d6ba78..c800a2327e900 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 @@ -22,6 +22,7 @@ import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeType; import org.apache.iotdb.consensus.common.request.IndexedConsensusRequest; import org.apache.iotdb.db.conf.IoTDBDescriptor; +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.io.WALFileVersion; @@ -87,7 +88,7 @@ public void testIteratorGroupsByLocalSeqAndCarriesWriterMetadata() throws Except } @Test - public void testIteratorUsesMetadataSearchIndexForStartFiltering() throws Exception { + public void testIteratorUsesBodySearchIndexForStartFiltering() throws Exception { final Path dir = Files.createTempDirectory("progress-wal-iterator-metadata-search-index"); final File firstWal = dir.resolve(WALFileUtils.getLogFileName(0, 0, WALFileStatus.CONTAINS_SEARCH_INDEX)) @@ -98,8 +99,8 @@ public void testIteratorUsesMetadataSearchIndexForStartFiltering() throws Except try { try (WALWriter writer = new WALWriter(firstWal, WALFileVersion.V3)) { - writer.write(searchableEntry(-1L), singleEntryMeta(19, 5L, 1L, 1000L, 7, 105L)); - writer.write(searchableEntry(-1L), singleEntryMeta(19, 6L, 1L, 2000L, 7, 106L)); + writer.write(searchableEntry(5L), singleEntryMeta(19, 5L, 1L, 1000L, 7, 105L)); + writer.write(searchableEntry(6L), singleEntryMeta(19, 6L, 1L, 2000L, 7, 106L)); } try (WALWriter ignored = new WALWriter(lastWal, WALFileVersion.V3)) { // Create a sealed successor so the first WAL becomes historical and readable. @@ -121,6 +122,75 @@ public void testIteratorUsesMetadataSearchIndexForStartFiltering() throws Except } } + @Test + public void testIteratorPreservesAbsentLocalIndexAfterManyFragments() throws Exception { + final Path dir = Files.createTempDirectory("progress-wal-iterator-mixed-fragments"); + final File dataWal = + dir.resolve(WALFileUtils.getLogFileName(0, 0, WALFileStatus.CONTAINS_SEARCH_INDEX)) + .toFile(); + final File successorWal = + dir.resolve(WALFileUtils.getLogFileName(1, 1878, WALFileStatus.CONTAINS_SEARCH_INDEX)) + .toFile(); + + try { + try (WALWriter writer = new WALWriter(dataWal, WALFileVersion.V3)) { + // Physical WAL entries count fragments, not local consensus requests. The old fallback + // assigns the replicated entry 1877 + 550 = 2427, creating a false [1878, 2427) gap. + for (int fragment = 0; fragment < 550; fragment++) { + writer.write(searchableEntry(1877L), singleEntryMeta(19, 1877L, 1L, 1000L, 7, 1877L)); + } + writer.write( + searchableEntry(SearchNode.NO_CONSENSUS_INDEX), + singleEntryMeta(19, SearchNode.NO_CONSENSUS_INDEX, 1L, 2000L, 8, 10000L)); + writer.write(searchableEntry(1878L), singleEntryMeta(19, 1878L, 1L, 3000L, 7, 1878L)); + } + try (WALWriter ignored = new WALWriter(successorWal, WALFileVersion.V3)) { + // Keep both files retained and sealed; there is no deletion or concurrent write. + } + + try (ProgressWALIterator iterator = new ProgressWALIterator(dir.toFile(), Long.MIN_VALUE)) { + assertTrue(iterator.hasNext()); + final IndexedConsensusRequest fragmented = iterator.next(); + assertEquals(1877L, fragmented.getSearchIndex()); + assertEquals(1877L, fragmented.getProgressLocalSeq()); + assertEquals(7, fragmented.getNodeId()); + assertEquals(550, fragmented.getRequests().size()); + + assertTrue(iterator.hasNext()); + final IndexedConsensusRequest replicated = iterator.next(); + assertEquals(SearchNode.NO_CONSENSUS_INDEX, replicated.getSearchIndex()); + assertEquals(10000L, replicated.getProgressLocalSeq()); + assertEquals(2000L, replicated.getPhysicalTime()); + assertEquals(8, replicated.getNodeId()); + assertEquals(1, replicated.getRequests().size()); + + assertTrue(iterator.hasNext()); + final IndexedConsensusRequest nextLocal = iterator.next(); + assertEquals(1878L, nextLocal.getSearchIndex()); + assertEquals(1878L, nextLocal.getProgressLocalSeq()); + assertEquals(7, nextLocal.getNodeId()); + assertEquals(1, nextLocal.getRequests().size()); + assertFalse(iterator.hasNext()); + assertFalse(iterator.hasIncompleteScan()); + assertTrue(dataWal.isFile()); + } + + // A local-index seek must also retain replicated requests with their own writer progress. + try (ProgressWALIterator iterator = new ProgressWALIterator(dir.toFile(), 1878L)) { + assertTrue(iterator.hasNext()); + assertEquals(SearchNode.NO_CONSENSUS_INDEX, iterator.next().getSearchIndex()); + assertTrue(iterator.hasNext()); + assertEquals(1878L, iterator.next().getSearchIndex()); + assertFalse(iterator.hasNext()); + assertFalse(iterator.hasIncompleteScan()); + } + } finally { + Files.deleteIfExists(dataWal.toPath()); + Files.deleteIfExists(successorWal.toPath()); + Files.deleteIfExists(dir); + } + } + @Test public void testIteratorMergesFragmentsWithSameLocalSeq() throws Exception { final Path dir = Files.createTempDirectory("progress-wal-iterator-merge");