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 e676d5c93e94..97a835e139e2 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 @@ -2270,6 +2270,12 @@ private DataNodePipeMessages() {} MESSAGE_SUBSCRIPTION_WAL_REPLAY_SKIPPED_ARG_UNREADABLE_RETAINED_WAL_FILE_S_IN_QUEUE_ARG_REGION_ARG_HISTORICAL_DATA_MAY_BE_LOST_DETAILS_RECENT_FILES_FULL_DETAILS_IN_SERVER_LOG_ARG_FEDEFF7E = "Subscription WAL replay skipped %d unreadable retained WAL file(s) in queue %s, region %s. Historical data may be lost. Details (recent files; full details in server log): %s"; public static final String MESSAGE_UNKNOWN_AD921D60 = "unknown"; + public static final String + MESSAGE_WAL_METADATA_SNAPSHOT_FOR_FILE_ARG_VERSIONID_ARG_IS_UNAVAILABLE_AFTER_ROTATION_REPLAY_WILL_RETRY_WITHOUT_SKIPPING_D99C12A0 = + "WAL metadata snapshot for file %s (versionId=%d) is unavailable after rotation; replay will retry without skipping"; + public static final String + EXCEPTION_WAL_FILE_ARG_VERSIONID_ARG_IS_MISSING_FROM_TWO_SUCCESSIVE_DIRECTORY_SCANS_AFTER_A_NEWER_WAL_FILE_BECAME_VISIBLE_1E170B4B = + "WAL file %s (versionId=%d) is missing from two successive directory scans after a newer WAL file became visible"; public static final String MESSAGE_WAL_FILE_ARG_VERSIONID_ARG_ENTRYOFFSET_ARG_IS_TEMPORARILY_UNREADABLE_REPLAY_WILL_RETRY_WITHOUT_SKIPPING_ARG_EA11FBDD = "WAL file %s (versionId=%d, entryOffset=%d) is temporarily unreadable; replay will retry without skipping: %s"; 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 a3949cd283e3..03e819fc635c 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 @@ -2111,6 +2111,12 @@ private DataNodePipeMessages() {} MESSAGE_SUBSCRIPTION_WAL_REPLAY_SKIPPED_ARG_UNREADABLE_RETAINED_WAL_FILE_S_IN_QUEUE_ARG_REGION_ARG_HISTORICAL_DATA_MAY_BE_LOST_DETAILS_RECENT_FILES_FULL_DETAILS_IN_SERVER_LOG_ARG_FEDEFF7E = "订阅 WAL 重放累计跳过了 %d 个无法读取的保留 WAL 文件，队列 %s，region %s。历史数据可能丢失。详情（近期文件；完整信息见服务端日志）：%s"; public static final String MESSAGE_UNKNOWN_AD921D60 = "未知"; + public static final String + MESSAGE_WAL_METADATA_SNAPSHOT_FOR_FILE_ARG_VERSIONID_ARG_IS_UNAVAILABLE_AFTER_ROTATION_REPLAY_WILL_RETRY_WITHOUT_SKIPPING_D99C12A0 = + "WAL 文件 %s（versionId=%d）在轮转后无法获取对应的元数据快照；回放将重试，不跳过该文件"; + public static final String + EXCEPTION_WAL_FILE_ARG_VERSIONID_ARG_IS_MISSING_FROM_TWO_SUCCESSIVE_DIRECTORY_SCANS_AFTER_A_NEWER_WAL_FILE_BECAME_VISIBLE_1E170B4B = + "后继 WAL 文件可见后，WAL 文件 %s（versionId=%d）在连续两次目录扫描中均不存在"; public static final String MESSAGE_WAL_FILE_ARG_VERSIONID_ARG_ENTRYOFFSET_ARG_IS_TEMPORARILY_UNREADABLE_REPLAY_WILL_RETRY_WITHOUT_SKIPPING_ARG_EA11FBDD = "WAL 文件 %s（versionId=%d，entryOffset=%d）暂不可读；重放将重试而不跳过：%s"; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java index 45432e7ab3df..ca975b397fc9 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java @@ -893,6 +893,17 @@ public WALMetaData getCurrentWALMetaDataSnapshot() { return writer == null ? new WALMetaData() : writer.snapshotMetaData(); } + /** Returns null if rotation has already installed a writer for another file version. */ + public WALMetaData getWALMetaDataSnapshot(final long fileVersionId) { + final WALWriter writer = currentWALFileWriter; + // The writer and currentWALFileVersion are published separately during rotation. Bind the + // snapshot to the captured writer itself, including while the version field is still old. + return writer != null + && WALFileUtils.parseVersionId(writer.getLogFile().getName()) == fileVersionId + ? writer.snapshotMetaData() + : null; + } + public CheckpointManager getCheckpointManager() { return checkpointManager; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNode.java index fd186d695d87..10526aaec51b 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNode.java @@ -1090,6 +1090,10 @@ public WALMetaData getCurrentWALMetaDataSnapshot() { return buffer.getCurrentWALMetaDataSnapshot(); } + public WALMetaData getWALMetaDataSnapshot(final long fileVersionId) { + return buffer.getWALMetaDataSnapshot(fileVersionId); + } + @Override public long getTotalSize() { return WALManager.getInstance().getTotalDiskUsage(); 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 5d0443c69960..5d6c60559aec 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 @@ -4736,8 +4736,8 @@ ProgressWALReader openProgressWALReader(final File walFile) throws IOException { private static LiveWALMetaDataSnapshot captureLiveWALMetaDataSnapshot(final WALNode walNode) { while (true) { final long versionId = walNode.getCurrentWALFileVersion(); - final WALMetaData metadata = walNode.getCurrentWALMetaDataSnapshot(); - if (versionId == walNode.getCurrentWALFileVersion()) { + final WALMetaData metadata = walNode.getWALMetaDataSnapshot(versionId); + if (metadata != null && versionId == walNode.getCurrentWALFileVersion()) { return new LiveWALMetaDataSnapshot(versionId, metadata); } } 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 ce33ee974bb3..4b1a58a30094 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 @@ -43,6 +43,8 @@ import java.io.File; import java.io.IOException; import java.nio.ByteBuffer; +import java.nio.channels.FileChannel; +import java.nio.file.Files; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -127,6 +129,7 @@ boolean hasYielded() { private long[] walFileVersionIds; private int currentFileIndex = -1; private ProgressWALReader currentReader; + private File currentReaderFile; private long currentReaderVersionId = -1L; private boolean currentReaderUsesLiveSnapshot = false; private int consumedEntryCountInCurrentFile = 0; @@ -135,6 +138,8 @@ boolean hasYielded() { private final List skippedBrokenWalFileDetails = new ArrayList<>(); private long skippedBrokenWalEntryCount; private boolean retryCurrentReader; + private boolean missingRetryFileObserved; + private String currentReaderRetryDetail; private long eofRetryVersionId = -1L; private int eofRetryEntryOffset = -1; private long eofRetryFileLength = -1L; @@ -152,6 +157,7 @@ boolean hasYielded() { private IndexedConsensusRequest nextReady; private boolean scanYielded; private long bodyReadAttempts; + private long readerOpenAttempts; public ProgressWALIterator(final File logDirectory) { this(logDirectory, Long.MIN_VALUE); @@ -398,7 +404,10 @@ private boolean isWalFileCovered( final WALMetaData metadata; final long versionId = walFileVersionIds[fileIndex]; if (liveWalNode != null && versionId == liveWalNode.getCurrentWALFileVersion()) { - metadata = liveWalNode.getCurrentWALMetaDataSnapshot(); + metadata = liveWalNode.getWALMetaDataSnapshot(versionId); + if (metadata == null) { + return false; + } } else { try (final ProgressWALReader reader = new ProgressWALReader(walFile)) { metadata = reader.getMetaData(); @@ -503,8 +512,13 @@ public String toString() { + consumedEntryCountInCurrentFile + ", bodyReadAttempts=" + bodyReadAttempts + + ", readerOpenAttempts=" + + readerOpenAttempts + ", minimumSearchIndex=" + minimumSearchIndex + + (retryCurrentReader + ? ", retryFile=" + currentReaderFile + ", retryReason=" + currentReaderRetryDetail + : "") + "]"; } @@ -558,7 +572,9 @@ private IndexedConsensusRequest advance(final ScanBudget scanBudget) throws IOEx refresh(); final int retryFileIndex = findFileIndexByVersion(currentReaderVersionId); if (retryFileIndex < 0) { - // Do not let a temporarily invisible file make replay silently move to its successor. + if (resumeAfterFilteredRetryFile()) { + continue; + } return null; } if (!openReaderAtIndex(retryFileIndex, consumedEntryCountInCurrentFile)) { @@ -664,6 +680,97 @@ private IndexedConsensusRequest advance(final ScanBudget scanBudget) throws IOEx } } + private boolean resumeAfterFilteredRetryFile() throws IOException { + final long versionId = currentReaderVersionId; + final boolean sealed = + liveWalNode != null + ? versionId < liveWalNode.getCurrentWALFileVersion() + : walFileVersionIds.length > 0 + && versionId < walFileVersionIds[walFileVersionIds.length - 1]; + final File[] discoveredFiles = WALFileUtils.listAllWALFiles(logDirectory); + if (discoveredFiles == null) { + missingRetryFileObserved = false; + return false; + } + File retryFile = null; + for (final File file : discoveredFiles) { + if (WALFileUtils.parseVersionId(file.getName()) == versionId) { + if (retryFile != null) { + missingRetryFileObserved = false; + return false; + } + retryFile = file; + } + } + if (retryFile != null) { + // Refresh deliberately omits sealed header-only files. A retry of such a file must finish + // instead of waiting forever for its version to reappear in the filtered list. + missingRetryFileObserved = false; + if (sealed && isHeaderOnlyWalFile(retryFile)) { + try (FileChannel channel = FileChannel.open(retryFile.toPath())) { + if (WALFileVersion.isEmptyOrHeaderOnly(channel)) { + if (consumedEntryCountInCurrentFile > 0) { + // This version used to contain entries. Confirm truncation before discarding its + // incomplete request, rather than treating its remaining header as a normal empty + // WAL. + final EOFException truncation = + new EOFException( + String.format( + DataNodePipeMessages + .MESSAGE_FAILED_TO_REOPEN_WAL_FILE_ARG_AT_ENTRY_OFFSET_ARG_ITERATOR_COULD_NOT_SKIP_TO_THE_REQUESTED_POSITION_332B3AD9, + retryFile.getName(), + consumedEntryCountInCurrentFile)); + if (!isConfirmedRetainedWalTruncation( + retryFile, versionId, consumedEntryCountInCurrentFile, false)) { + markIncompleteScan(truncation.getMessage(), truncation); + return false; + } + recordSkippedBrokenWalFile( + versionId, retryFile, truncation, null, consumedEntryCountInCurrentFile); + pendingRequests.clear(); + pendingSearchIndex = Long.MIN_VALUE; + pendingLocalSeq = Long.MIN_VALUE; + } + resetCurrentFileTracking(); + return true; + } + } + } + return false; + } else { + // Do not skip a temporarily invisible active file, an inaccessible directory, or a rename + // for which another filename with the same version is still present. Require the retry + // version to be sealed, a visible successor, and two listings confirming the path is absent. + if (!sealed + || currentReaderFile == null + || walFileVersionIds.length == 0 + || walFileVersionIds[walFileVersionIds.length - 1] <= versionId + || !Files.notExists(currentReaderFile.toPath())) { + missingRetryFileObserved = false; + return false; + } + if (!missingRetryFileObserved) { + missingRetryFileObserved = true; + return false; + } + final IOException missingFile = + new IOException( + String.format( + DataNodePipeMessages + .EXCEPTION_WAL_FILE_ARG_VERSIONID_ARG_IS_MISSING_FROM_TWO_SUCCESSIVE_DIRECTORY_SCANS_AFTER_A_NEWER_WAL_FILE_BECAME_VISIBLE_1E170B4B, + currentReaderFile.getAbsolutePath(), + versionId)); + recordSkippedBrokenWalFile( + versionId, currentReaderFile, missingFile, null, consumedEntryCountInCurrentFile); + // The missing file may have continued the incomplete request. Never publish its prefix. + pendingRequests.clear(); + pendingSearchIndex = Long.MIN_VALUE; + pendingLocalSeq = Long.MIN_VALUE; + resetCurrentFileTracking(); + return true; + } + } + private boolean openNextReader(final ScanBudget scanBudget) throws IOException { while (currentFileIndex + 1 < walFiles.length) { if (scanBudget != null && scanBudget.shouldYield()) { @@ -692,7 +799,15 @@ private boolean reopenLiveSnapshotReader() throws IOException { final long currentLiveVersionId = liveWalNode.getCurrentWALFileVersion(); if (currentLiveVersionId == currentReaderVersionId) { - final WALMetaData snapshot = liveWalNode.getCurrentWALMetaDataSnapshot(); + final WALMetaData snapshot = liveWalNode.getWALMetaDataSnapshot(currentReaderVersionId); + if (snapshot == null) { + final int fileIndex = findFileIndexByVersion(currentReaderVersionId); + if (fileIndex < 0) { + retryCurrentReader = true; + return false; + } + return openReaderAtIndex(fileIndex, consumedEntryCountInCurrentFile); + } if (snapshot.getBuffersSize().size() <= consumedEntryCountInCurrentFile) { return false; } @@ -736,39 +851,52 @@ private boolean openReaderAtIndex( final File walFile = walFiles[fileIndex]; final long versionId = walFileVersionIds[fileIndex]; currentFileIndex = fileIndex; + currentReaderFile = walFile; + readerOpenAttempts++; final boolean useLiveSnapshot = liveWalNode != null && versionId == liveWalNode.getCurrentWALFileVersion(); ProgressWALReader reader = null; try { - reader = - useLiveSnapshot - ? new ProgressWALReader( - walFile, - liveMetaDataSnapshot != null - ? liveMetaDataSnapshot - : liveWalNode.getCurrentWALMetaDataSnapshot()) - : new ProgressWALReader(walFile); - if (!skipEntries(reader, skipEntries)) { - reader.close(); + final WALMetaData snapshot = + !useLiveSnapshot + ? null + : liveMetaDataSnapshot != null + ? liveMetaDataSnapshot + : liveWalNode.getWALMetaDataSnapshot(versionId); + if (useLiveSnapshot && snapshot == null) { markIncompleteScan( String.format( DataNodePipeMessages - .MESSAGE_FAILED_TO_REOPEN_WAL_FILE_ARG_AT_ENTRY_OFFSET_ARG_ITERATOR_COULD_NOT_SKIP_TO_THE_REQUESTED_POSITION_332B3AD9, - walFile.getName(), - skipEntries), + .MESSAGE_WAL_METADATA_SNAPSHOT_FOR_FILE_ARG_VERSIONID_ARG_IS_UNAVAILABLE_AFTER_ROTATION_REPLAY_WILL_RETRY_WITHOUT_SKIPPING_D99C12A0, + walFile.getAbsolutePath(), + versionId), null); currentReaderVersionId = versionId; consumedEntryCountInCurrentFile = skipEntries; retryCurrentReader = true; return false; } + reader = + useLiveSnapshot + ? new ProgressWALReader(walFile, snapshot) + : new ProgressWALReader(walFile); + if (!skipEntries(reader, skipEntries)) { + throw new EOFException( + String.format( + DataNodePipeMessages + .MESSAGE_FAILED_TO_REOPEN_WAL_FILE_ARG_AT_ENTRY_OFFSET_ARG_ITERATOR_COULD_NOT_SKIP_TO_THE_REQUESTED_POSITION_332B3AD9, + walFile.getName(), + skipEntries)); + } currentReader = reader; currentFileIndex = fileIndex; currentReaderVersionId = versionId; currentReaderUsesLiveSnapshot = useLiveSnapshot; consumedEntryCountInCurrentFile = skipEntries; retryCurrentReader = false; + missingRetryFileObserved = false; + currentReaderRetryDetail = null; return true; } catch (final IOException e) { if (reader != null) { @@ -778,15 +906,19 @@ private boolean openReaderAtIndex( e.addSuppressed(closeException); } } - if (isNearLiveWalVersion(versionId) - || !(e instanceof BrokenWALFileException) - || (liveWalNode == null && fileIndex == walFiles.length - 1)) { + final boolean confirmedTruncation = + e instanceof EOFException + && isConfirmedRetainedWalTruncation(walFile, versionId, skipEntries, useLiveSnapshot); + if (!confirmedTruncation + && (isNearLiveWalVersion(versionId) + || !(e instanceof BrokenWALFileException) + || (liveWalNode == null && fileIndex == walFiles.length - 1))) { LOGGER.debug( DataNodePipeMessages .PIPE_LOG_PROGRESSWALITERATOR_FAILED_TO_OPEN_NEAR_LIVE_WAL_FILE_RETRYING_5AEB94AC, walFile.getName(), e); - if (allowNearLiveRetry) { + if (allowNearLiveRetry && !(e instanceof EOFException)) { refresh(); final int refreshedIndex = findFileIndexByVersion(versionId); if (refreshedIndex >= 0) { @@ -809,7 +941,8 @@ private boolean openReaderAtIndex( retryCurrentReader = true; return false; } - recordSkippedBrokenWalFile(versionId, walFile, e, null, skipEntries); + recordSkippedBrokenWalFile( + versionId, walFile, e, reader == null ? null : reader.getMetaData(), skipEntries); pendingRequests.clear(); pendingSearchIndex = Long.MIN_VALUE; pendingLocalSeq = Long.MIN_VALUE; @@ -838,10 +971,11 @@ private void recordSkippedBrokenWalFile( // Filename boundaries are file-level search-index bounds, not an entry count. Footer entry // ordinals also include request fragments and entries from other writers. final long firstSearchIndex = WALFileUtils.parseStartSearchIndex(walFile.getName()); - final int fileIndex = findFileIndexByVersion(versionId); + final int fileIndex = Arrays.binarySearch(walFileVersionIds, versionId); + final int successorIndex = fileIndex >= 0 ? fileIndex + 1 : -fileIndex - 1; final long lastSearchIndex = - fileIndex >= 0 && fileIndex + 1 < walFiles.length - ? WALFileUtils.parseStartSearchIndex(walFiles[fileIndex + 1].getName()) + successorIndex < walFiles.length + ? WALFileUtils.parseStartSearchIndex(walFiles[successorIndex].getName()) : -1L; final String detail = String.format( @@ -871,20 +1005,31 @@ private static String formatKnownValue(final long value) { } private boolean isConfirmedRetainedWalTruncation() { - final File walFile = walFiles[currentFileIndex]; + return isConfirmedRetainedWalTruncation( + walFiles[currentFileIndex], + currentReaderVersionId, + consumedEntryCountInCurrentFile, + currentReaderUsesLiveSnapshot); + } + + private boolean isConfirmedRetainedWalTruncation( + final File walFile, + final long versionId, + final int entryOffset, + final boolean usesLiveSnapshot) { final boolean sealed = liveWalNode != null - ? !isNearLiveWalVersion(currentReaderVersionId) + ? !isNearLiveWalVersion(versionId) : currentFileIndex + 1 < walFiles.length; final boolean confirmed = - !currentReaderUsesLiveSnapshot + !usesLiveSnapshot && sealed - && eofRetryVersionId == currentReaderVersionId - && eofRetryEntryOffset == consumedEntryCountInCurrentFile + && eofRetryVersionId == versionId + && eofRetryEntryOffset == entryOffset && eofRetryFileLength == walFile.length() && eofRetryFileModified == walFile.lastModified(); - eofRetryVersionId = currentReaderVersionId; - eofRetryEntryOffset = consumedEntryCountInCurrentFile; + eofRetryVersionId = versionId; + eofRetryEntryOffset = entryOffset; eofRetryFileLength = walFile.length(); eofRetryFileModified = walFile.lastModified(); return confirmed; @@ -1068,15 +1213,19 @@ private void closeCurrentReader() throws IOException { } private void resetCurrentFileTracking() { + currentReaderFile = null; currentReaderVersionId = -1L; currentReaderUsesLiveSnapshot = false; consumedEntryCountInCurrentFile = 0; retryCurrentReader = false; + missingRetryFileObserved = false; + currentReaderRetryDetail = null; eofRetryVersionId = -1L; eofRetryEntryOffset = -1; } private void markIncompleteScan(final String detail, final IOException cause) { + currentReaderRetryDetail = detail; incompleteScan = true; if (incompleteScanDetail == null) { incompleteScanDetail = detail; diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBufferRollRecoveryTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBufferRollRecoveryTest.java index 84e61d7f00c5..b6f6d8f5bd61 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBufferRollRecoveryTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBufferRollRecoveryTest.java @@ -57,6 +57,7 @@ import static org.junit.Assert.assertArrayEquals; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.doCallRealMethod; @@ -140,6 +141,28 @@ public void testResumeBeforeWritingAfterRepeatedOpenFailures() throws Exception assertEquals(Arrays.asList(2L), readTimes(successor)); } + @Test + public void testMetadataSnapshotIsBoundToWriterDuringVersionPublication() throws Exception { + writeAndAwait(entry(1, 1, "first"), Status.SUCCESS); + assertEquals(1, buffer.getWALMetaDataSnapshot(0L).getBuffersSize().size()); + roll(Status.SUCCESS); + writeAndAwait(entry(2, 2, "second"), Status.SUCCESS); + assertNull(buffer.getWALMetaDataSnapshot(0L)); + final WALMetaData successorMetadata = buffer.getWALMetaDataSnapshot(1L); + assertEquals(1, successorMetadata.getBuffersSize().size()); + + // Reproduce the interval in rollLogWriter between publishing the new writer and publishing + // its version. Checking the version field before and after snapshot capture is insufficient. + buffer.currentWALFileVersion = 0L; + try { + assertNull(buffer.getWALMetaDataSnapshot(0L)); + assertEquals( + successorMetadata.getBuffersSize(), buffer.getWALMetaDataSnapshot(1L).getBuffersSize()); + } finally { + buffer.currentWALFileVersion = 1L; + } + } + /** A retry containing only the roll signal should open exactly one successor. */ @Test public void testRollSignalResumesWithoutRollingTwice() throws Exception { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueDeliveryTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueDeliveryTest.java index e1ff4aaeb0b2..df12f63caa68 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueDeliveryTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueDeliveryTest.java @@ -134,7 +134,7 @@ public void setUp() throws Exception { when(walNode.getLogDirectory()).thenReturn(walDirectory); when(walNode.getCurrentSearchIndex()).thenAnswer(ignored -> currentSearchIndex.get()); when(walNode.getCurrentWALFileVersion()).thenReturn(1L); - when(walNode.getCurrentWALMetaDataSnapshot()).thenReturn(new WALMetaData()); + when(walNode.getWALMetaDataSnapshot(1L)).thenReturn(new WALMetaData()); when(walNode.getSortedWalFilesSnapshot()) .thenAnswer( ignored -> { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueInitializationTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueInitializationTest.java index a663437e360b..ad7333751d1d 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueInitializationTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueInitializationTest.java @@ -59,6 +59,7 @@ import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.times; @@ -90,7 +91,7 @@ public void setUp() throws Exception { when(walNode.getLogDirectory()).thenReturn(walDirectory); when(walNode.getCurrentSearchIndex()).thenReturn(3L); when(walNode.getCurrentWALFileVersion()).thenReturn(3L); - when(walNode.getCurrentWALMetaDataSnapshot()).thenReturn(new WALMetaData()); + when(walNode.getWALMetaDataSnapshot(anyLong())).thenReturn(new WALMetaData()); when(converter.convert(any())).thenReturn(Collections.emptyList()); when(converter.isTableModel()).thenReturn(true); @@ -168,7 +169,7 @@ public void testRepeatedUnreadablePollsKeepInitializationDormant() throws Except Files.delete(liveWal.toPath()); final WALMetaData metadata = writeRequests(liveWal); - when(walNode.getCurrentWALMetaDataSnapshot()).thenReturn(metadata); + when(walNode.getWALMetaDataSnapshot(anyLong())).thenReturn(metadata); assertResumesAfterCommittedProgress(); } 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 614934751159..fd6f1a855a84 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 @@ -104,6 +104,7 @@ import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; @@ -1810,7 +1811,7 @@ public void testTailProgressFlushesPendingWalMetadataBeforeSnapshot() throws Exc flushedMetadata.add(Integer.BYTES, 1L, 0L, 0L, 7, 1L); when(walNode.getLogDirectory()).thenReturn(walDirectory); when(walNode.getCurrentWALFileVersion()).thenReturn(0L); - when(walNode.getCurrentWALMetaDataSnapshot()) + when(walNode.getWALMetaDataSnapshot(anyLong())) .thenAnswer(ignored -> rolled.get() ? flushedMetadata.copy() : new WALMetaData()); doAnswer( ignored -> { @@ -1873,7 +1874,7 @@ public void testTailProgressSkipsHeaderOnlyWalFile() throws Exception { final WALNode walNode = mock(WALNode.class); when(walNode.getLogDirectory()).thenReturn(walDirectory); when(walNode.getCurrentWALFileVersion()).thenReturn(1L); - when(walNode.getCurrentWALMetaDataSnapshot()).thenReturn(new WALMetaData()); + when(walNode.getWALMetaDataSnapshot(anyLong())).thenReturn(new WALMetaData()); final IoTConsensusServerImpl serverImpl = mock(IoTConsensusServerImpl.class); when(serverImpl.getConsensusReqReader()).thenReturn(walNode); queue = diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueWalBackpressureTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueWalBackpressureTest.java index adcf18a50d82..24d76d2d1236 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueWalBackpressureTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueWalBackpressureTest.java @@ -79,6 +79,7 @@ import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; @@ -129,7 +130,7 @@ public void testYieldedGapScanKeepsIteratorAndDeliversEachRequestOnce() throws E when(walNode.getLogDirectory()).thenReturn(walDirectory); when(walNode.getCurrentSearchIndex()).thenReturn(3L); when(walNode.getCurrentWALFileVersion()).thenReturn(1L); - when(walNode.getCurrentWALMetaDataSnapshot()).thenReturn(new WALMetaData()); + when(walNode.getWALMetaDataSnapshot(anyLong())).thenReturn(new WALMetaData()); final IoTConsensusServerImpl serverImpl = mock(IoTConsensusServerImpl.class); when(serverImpl.getConsensusReqReader()).thenReturn(walNode); when(serverImpl.getWriterSafeFrontierTracker()).thenReturn(new WriterSafeFrontierTracker()); @@ -304,7 +305,7 @@ private void assertWalGapRecoveryResumesRealtime( when(walNode.getLogDirectory()).thenReturn(walDirectory); when(walNode.getCurrentSearchIndex()).thenReturn((long) requestCount); when(walNode.getCurrentWALFileVersion()).thenReturn(1L); - when(walNode.getCurrentWALMetaDataSnapshot()).thenReturn(new WALMetaData()); + when(walNode.getWALMetaDataSnapshot(anyLong())).thenReturn(new WALMetaData()); final IoTConsensusServerImpl serverImpl = mock(IoTConsensusServerImpl.class); when(serverImpl.getConsensusReqReader()).thenReturn(walNode); when(serverImpl.getWriterSafeFrontierTracker()).thenReturn(new WriterSafeFrontierTracker()); @@ -453,7 +454,7 @@ public void testAckResumesPendingSuffixWithoutWalReplay() throws Exception { when(walNode.getLogDirectory()).thenReturn(walDirectory); when(walNode.getCurrentSearchIndex()).thenReturn((long) REQUEST_COUNT); when(walNode.getCurrentWALFileVersion()).thenReturn(1L); - when(walNode.getCurrentWALMetaDataSnapshot()).thenReturn(new WALMetaData()); + when(walNode.getWALMetaDataSnapshot(anyLong())).thenReturn(new WALMetaData()); final IoTConsensusServerImpl serverImpl = mock(IoTConsensusServerImpl.class); when(serverImpl.getConsensusReqReader()).thenReturn(walNode); @@ -588,7 +589,7 @@ public void testWalMemoryBlockRetriesWithoutRecreatingIterator() throws Exceptio when(walNode.getLogDirectory()).thenReturn(walDirectory); when(walNode.getCurrentSearchIndex()).thenReturn((long) REQUEST_COUNT); when(walNode.getCurrentWALFileVersion()).thenReturn(1L); - when(walNode.getCurrentWALMetaDataSnapshot()).thenReturn(new WALMetaData()); + when(walNode.getWALMetaDataSnapshot(anyLong())).thenReturn(new WALMetaData()); final IoTConsensusServerImpl serverImpl = mock(IoTConsensusServerImpl.class); when(serverImpl.getConsensusReqReader()).thenReturn(walNode); @@ -726,7 +727,7 @@ private void assertPendingGapReplay( when(walNode.getLogDirectory()).thenReturn(walDirectory); when(walNode.getCurrentSearchIndex()).thenReturn((long) requestCount); when(walNode.getCurrentWALFileVersion()).thenReturn(1L); - when(walNode.getCurrentWALMetaDataSnapshot()).thenReturn(new WALMetaData()); + when(walNode.getWALMetaDataSnapshot(anyLong())).thenReturn(new WALMetaData()); final IoTConsensusServerImpl serverImpl = mock(IoTConsensusServerImpl.class); when(serverImpl.getConsensusReqReader()).thenReturn(walNode); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueWriterOrderingTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueWriterOrderingTest.java index b34dd71b6f5d..d943ca412410 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueWriterOrderingTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueWriterOrderingTest.java @@ -476,7 +476,7 @@ private static WALNode walNode(final File directory, final long currentIndex) { when(walNode.getLogDirectory()).thenReturn(directory); when(walNode.getCurrentSearchIndex()).thenReturn(currentIndex); when(walNode.getCurrentWALFileVersion()).thenReturn(1L); - when(walNode.getCurrentWALMetaDataSnapshot()).thenReturn(new WALMetaData()); + when(walNode.getWALMetaDataSnapshot(1L)).thenReturn(new WALMetaData()); return walNode; } 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 370fca537965..a4bfb29304d7 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 @@ -52,6 +52,7 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; @@ -666,7 +667,7 @@ private void verifyEofRetryPreservesPendingFragments(final boolean nearLive) thr final WALNode walNode = mock(WALNode.class); when(walNode.getLogDirectory()).thenReturn(dir.toFile()); when(walNode.getCurrentWALFileVersion()).thenReturn(1L); - when(walNode.getCurrentWALMetaDataSnapshot()).thenReturn(successorMetadata); + when(walNode.getWALMetaDataSnapshot(anyLong())).thenReturn(successorMetadata); try (ProgressWALIterator iterator = nearLive ? new ProgressWALIterator(walNode, 1L) @@ -732,6 +733,61 @@ public void testRepeatedEofInStableRetainedFileReportsTruncatedRequest() throws } } + @Test + public void testTruncatedWalReopenDoesNotRetryUnreachableOffsetForever() throws Exception { + assertTruncatedWalReplayResumes(false); + } + + @Test + public void testHeaderOnlyWalTruncationDoesNotPublishIncompleteRequest() throws Exception { + assertTruncatedWalReplayResumes(true); + } + + private void assertTruncatedWalReplayResumes(final boolean headerOnly) throws Exception { + final Path dir = Files.createTempDirectory("progress-wal-retry-truncated-offset"); + final File dataWal = + dir.resolve(WALFileUtils.getLogFileName(0, 0, WALFileStatus.CONTAINS_SEARCH_INDEX)) + .toFile(); + final File successorWal = + dir.resolve(WALFileUtils.getLogFileName(1, 2, WALFileStatus.CONTAINS_SEARCH_INDEX)) + .toFile(); + try { + try (WALWriter writer = new WALWriter(dataWal, WALFileVersion.V3)) { + writer.write(searchableEntry(1L), singleEntryMeta(19, 1L, 1L, 100L, 7, 1L)); + writer.write(searchableEntry(2L), singleEntryMeta(19, 2L, 1L, 200L, 7, 2L)); + writer.write(searchableEntry(2L), singleEntryMeta(38, 2L, 1L, 200L, 7, 2L)); + } + try (WALWriter writer = new WALWriter(successorWal, WALFileVersion.V3)) { + writer.write(searchableEntry(3L), singleEntryMeta(19, 3L, 1L, 300L, 7, 3L)); + } + try (ProgressWALIterator iterator = new ProgressWALIterator(dir.toFile(), 1L)) { + assertTrue(iterator.hasNext()); + assertEquals(1L, iterator.next().getSearchIndex()); + assertFalse(iterator.hasNext()); + assertTrue(iterator.isWaitingForReadableWal()); + + // A repaired/truncated retained file now has fewer entries than the saved retry offset. + Files.delete(dataWal.toPath()); + try (WALWriter writer = new WALWriter(dataWal, WALFileVersion.V3)) { + if (!headerOnly) { + writer.write(searchableEntry(1L), singleEntryMeta(19, 1L, 1L, 100L, 7, 1L)); + } + } + assertFalse(iterator.hasNext()); + assertEquals(0, iterator.getSkippedBrokenWalFileCount()); + assertTrue(iterator.hasNext()); + assertEquals(3L, iterator.next().getSearchIndex()); + assertEquals(1, iterator.getSkippedBrokenWalFileCount()); + assertTrue(iterator.getSkippedBrokenWalFileDetails(0).contains("EOFException")); + assertFalse(iterator.hasNext()); + } + } finally { + Files.deleteIfExists(dataWal.toPath()); + Files.deleteIfExists(successorWal.toPath()); + Files.deleteIfExists(dir); + } + } + @Test public void testLiveSnapshotEofPreservesIncompleteRequestUntilBytesAreVisible() throws Exception { final Path dir = Files.createTempDirectory("progress-wal-iterator-live-eof-retry"); @@ -744,7 +800,7 @@ public void testLiveSnapshotEofPreservesIncompleteRequestUntilBytesAreVisible() final WALNode walNode = mock(WALNode.class); when(walNode.getLogDirectory()).thenReturn(dir.toFile()); when(walNode.getCurrentWALFileVersion()).thenReturn(0L); - when(walNode.getCurrentWALMetaDataSnapshot()).thenReturn(metadata); + when(walNode.getWALMetaDataSnapshot(anyLong())).thenReturn(metadata); try (WALWriter writer = new WALWriter(liveWal, WALFileVersion.V3)) { writer.write(searchableEntry(1L), singleEntryMeta(19, 1L, 1L, 100L, 7, 1L)); try (ProgressWALIterator iterator = new ProgressWALIterator(walNode, 1L)) { @@ -811,6 +867,170 @@ public void testSkippedFileDetailsRemainBoundedDuringContinuousReplay() throws E } } + @Test + public void testRetryContinuesAfterSealedEmptyWalIsFiltered() throws Exception { + final Path dir = Files.createTempDirectory("progress-wal-retry-empty-file"); + final File emptyWal = + dir.resolve(WALFileUtils.getLogFileName(0, 0, WALFileStatus.CONTAINS_SEARCH_INDEX)) + .toFile(); + final File successorWal = + dir.resolve(WALFileUtils.getLogFileName(1, 0, WALFileStatus.CONTAINS_SEARCH_INDEX)) + .toFile(); + try { + // Simulate an open failure while this is the active WAL. The same version will become a + // valid sealed header-only file, which refreshFileList deliberately filters out. + assertTrue(emptyWal.mkdir()); + final WALNode walNode = mock(WALNode.class); + when(walNode.getLogDirectory()).thenReturn(dir.toFile()); + when(walNode.getCurrentWALFileVersion()).thenReturn(0L); + when(walNode.getWALMetaDataSnapshot(anyLong())).thenReturn(new WALMetaData()); + try (ProgressWALIterator iterator = new ProgressWALIterator(walNode, 1L)) { + assertFalse(iterator.hasNext()); + assertTrue(iterator.isWaitingForReadableWal()); + Files.delete(emptyWal.toPath()); + try (WALWriter ignored = new WALWriter(emptyWal, WALFileVersion.V3)) { + // A sealed empty WAL has only its version header. + } + final WALMetaData metadata = singleEntryMeta(19, 1L, 1L, 100L, 7, 1L); + try (WALWriter writer = new WALWriter(successorWal, WALFileVersion.V3)) { + writer.write(searchableEntry(1L), metadata); + } + when(walNode.getCurrentWALFileVersion()).thenReturn(1L); + when(walNode.getWALMetaDataSnapshot(anyLong())).thenReturn(metadata); + + assertTrue(iterator.toString(), iterator.hasNext()); + assertEquals(1L, iterator.next().getSearchIndex()); + assertFalse(iterator.hasNext()); + assertFalse(iterator.isWaitingForReadableWal()); + assertEquals(0, iterator.getSkippedBrokenWalFileCount()); + } + } finally { + Files.deleteIfExists(emptyWal.toPath()); + Files.deleteIfExists(successorWal.toPath()); + Files.deleteIfExists(dir); + } + } + + @Test + public void testMissingActiveWalRetriesUntilItReappears() throws Exception { + final Path dir = Files.createTempDirectory("progress-wal-retry-invisible-file"); + final File liveWal = + dir.resolve(WALFileUtils.getLogFileName(0, 0, WALFileStatus.CONTAINS_SEARCH_INDEX)) + .toFile(); + try { + assertTrue(liveWal.mkdir()); + final WALMetaData metadata = singleEntryMeta(19, 1L, 1L, 100L, 7, 1L); + final WALNode walNode = mock(WALNode.class); + when(walNode.getLogDirectory()).thenReturn(dir.toFile()); + when(walNode.getCurrentWALFileVersion()).thenReturn(0L); + when(walNode.getWALMetaDataSnapshot(0L)).thenReturn(metadata); + try (ProgressWALIterator iterator = new ProgressWALIterator(walNode, 1L)) { + assertFalse(iterator.hasNext()); + Files.delete(liveWal.toPath()); + for (int attempt = 0; attempt < 3; attempt++) { + assertFalse(iterator.hasNext()); + assertTrue(iterator.isWaitingForReadableWal()); + assertEquals(0, iterator.getSkippedBrokenWalFileCount()); + } + assertTrue(iterator.toString().contains("retryFile=" + liveWal)); + assertTrue(iterator.toString().contains("retryReason=")); + try (WALWriter writer = new WALWriter(liveWal, WALFileVersion.V3)) { + writer.write(searchableEntry(1L), metadata); + } + assertTrue(iterator.hasNext()); + assertEquals(1L, iterator.next().getSearchIndex()); + assertEquals(0, iterator.getSkippedBrokenWalFileCount()); + } + } finally { + Files.deleteIfExists(liveWal.toPath()); + Files.deleteIfExists(dir); + } + } + + @Test + public void testMissingRetainedRetryFileIsReportedBeforeReplayContinues() throws Exception { + final Path dir = Files.createTempDirectory("progress-wal-retry-missing-file"); + final File missingWal = + dir.resolve(WALFileUtils.getLogFileName(0, 0, WALFileStatus.CONTAINS_SEARCH_INDEX)) + .toFile(); + final File successorWal = + dir.resolve(WALFileUtils.getLogFileName(1, 1, WALFileStatus.CONTAINS_SEARCH_INDEX)) + .toFile(); + try { + assertTrue(missingWal.mkdir()); + final WALNode walNode = mock(WALNode.class); + when(walNode.getLogDirectory()).thenReturn(dir.toFile()); + when(walNode.getCurrentWALFileVersion()).thenReturn(0L); + when(walNode.getWALMetaDataSnapshot(0L)).thenReturn(new WALMetaData()); + try (ProgressWALIterator iterator = new ProgressWALIterator(walNode, 1L)) { + assertFalse(iterator.hasNext()); + Files.delete(missingWal.toPath()); + final WALMetaData metadata = singleEntryMeta(19, 2L, 1L, 200L, 7, 2L); + try (WALWriter writer = new WALWriter(successorWal, WALFileVersion.V3)) { + writer.write(searchableEntry(2L), metadata); + } + when(walNode.getCurrentWALFileVersion()).thenReturn(1L); + when(walNode.getWALMetaDataSnapshot(1L)).thenReturn(metadata); + assertFalse(iterator.hasNext()); + assertEquals(0, iterator.getSkippedBrokenWalFileCount()); + assertTrue(iterator.hasNext()); + assertEquals(2L, iterator.next().getSearchIndex()); + assertEquals(1, iterator.getSkippedBrokenWalFileCount()); + final String detail = iterator.getSkippedBrokenWalFileDetails(0); + assertTrue(detail.contains(missingWal.getName())); + assertTrue(detail.contains("fileSearchIndexRange=(0, 1]")); + assertFalse(iterator.hasNext()); + } + } finally { + Files.deleteIfExists(missingWal.toPath()); + Files.deleteIfExists(successorWal.toPath()); + Files.deleteIfExists(dir); + } + } + + @Test + public void testRotationDuringSnapshotCaptureDoesNotUseSuccessorMetadata() throws Exception { + final Path dir = Files.createTempDirectory("progress-wal-snapshot-rotation"); + final File emptyWal = + dir.resolve(WALFileUtils.getLogFileName(0, 0, WALFileStatus.CONTAINS_SEARCH_INDEX)) + .toFile(); + final File successorWal = + dir.resolve(WALFileUtils.getLogFileName(1, 0, WALFileStatus.CONTAINS_SEARCH_INDEX)) + .toFile(); + try { + try (WALWriter ignored = new WALWriter(emptyWal, WALFileVersion.V3)) { + // Empty WAL that is current when opening starts. + } + final WALMetaData metadata = singleEntryMeta(19, 1L, 1L, 100L, 7, 1L); + final AtomicLong currentVersion = new AtomicLong(0L); + final WALNode walNode = mock(WALNode.class); + when(walNode.getLogDirectory()).thenReturn(dir.toFile()); + when(walNode.getCurrentWALFileVersion()).thenAnswer(ignored -> currentVersion.get()); + when(walNode.getWALMetaDataSnapshot(0L)) + .thenAnswer( + ignored -> { + try (WALWriter writer = new WALWriter(successorWal, WALFileVersion.V3)) { + writer.write(searchableEntry(1L), metadata); + } + currentVersion.set(1L); + return null; + }); + when(walNode.getWALMetaDataSnapshot(1L)).thenReturn(metadata); + try (ProgressWALIterator iterator = new ProgressWALIterator(walNode, 1L)) { + assertFalse(iterator.hasNext()); + assertTrue(iterator.isWaitingForReadableWal()); + assertTrue(iterator.hasNext()); + assertEquals(1L, iterator.next().getSearchIndex()); + assertFalse(iterator.hasNext()); + assertEquals(0, iterator.getSkippedBrokenWalFileCount()); + } + } finally { + Files.deleteIfExists(emptyWal.toPath()); + Files.deleteIfExists(successorWal.toPath()); + Files.deleteIfExists(dir); + } + } + @Test public void testLiveWalReopenReusesMetadataSnapshot() throws Exception { final Path dir = Files.createTempDirectory("progress-wal-iterator-live-snapshot"); @@ -831,14 +1051,14 @@ public void testLiveWalReopenReusesMetadataSnapshot() throws Exception { final WALNode walNode = mock(WALNode.class); when(walNode.getLogDirectory()).thenReturn(dir.toFile()); when(walNode.getCurrentWALFileVersion()).thenReturn(0L); - when(walNode.getCurrentWALMetaDataSnapshot()).thenReturn(firstSnapshot, secondSnapshot); + when(walNode.getWALMetaDataSnapshot(anyLong())).thenReturn(firstSnapshot, secondSnapshot); try (ProgressWALIterator iterator = new ProgressWALIterator(walNode, 1L)) { assertTrue(iterator.hasNext()); assertEquals(1L, iterator.next().getSearchIndex()); assertTrue(iterator.hasNext()); assertEquals(2L, iterator.next().getSearchIndex()); - verify(walNode, times(2)).getCurrentWALMetaDataSnapshot(); + verify(walNode, times(2)).getWALMetaDataSnapshot(anyLong()); } } finally { Files.deleteIfExists(liveWal.toPath()); @@ -859,7 +1079,7 @@ public void testIteratorMarksIncompleteScanWhenNearLiveWalCannotBeOpened() throw final WALNode walNode = mock(WALNode.class); when(walNode.getLogDirectory()).thenReturn(dir.toFile()); when(walNode.getCurrentWALFileVersion()).thenReturn(7L); - when(walNode.getCurrentWALMetaDataSnapshot()).thenReturn(new WALMetaData()); + when(walNode.getWALMetaDataSnapshot(anyLong())).thenReturn(new WALMetaData()); try (ProgressWALIterator iterator = new ProgressWALIterator(walNode, Long.MIN_VALUE)) { assertFalse(iterator.hasNext());