You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ji...@apache.org on 2026/10/20 14:25:43 UTC
(iotdb) branch master updated: [Subscription] Fix stalled WAL replay retries across file refresh and rotation (#18919)
This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 6844c476885 [Subscription] Fix stalled WAL replay retries across file refresh and rotation (#18919)
6844c476885 is described below
commit 6844c4768855004dae5313aa435b41a5058cdc2d
Author: Caideyipi <87...@users.noreply.github.com>
AuthorDate: Tue Oct 20 22:25:36 2026 +0800
[Subscription] Fix stalled WAL replay retries across file refresh and rotation (#18919)
---
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 6 +
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 6 +
.../dataregion/wal/buffer/WALBuffer.java | 11 +
.../storageengine/dataregion/wal/node/WALNode.java | 4 +
.../consensus/ConsensusPrefetchingQueue.java | 4 +-
.../broker/consensus/ProgressWALIterator.java | 211 ++++++++++++++++---
.../wal/buffer/WALBufferRollRecoveryTest.java | 23 +++
.../ConsensusPrefetchingQueueDeliveryTest.java | 2 +-
...onsensusPrefetchingQueueInitializationTest.java | 5 +-
.../consensus/ConsensusPrefetchingQueueTest.java | 5 +-
...nsensusPrefetchingQueueWalBackpressureTest.java | 11 +-
...onsensusPrefetchingQueueWriterOrderingTest.java | 2 +-
.../broker/consensus/ProgressWALIteratorTest.java | 230 ++++++++++++++++++++-
13 files changed, 471 insertions(+), 49 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 374a35e1642..bb04beac563 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 @@ public final class 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 d9439de9014..d319d0e3772 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 @@ public final class 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 45432e7ab3d..ca975b397fc 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 class WALBuffer extends AbstractWALBuffer {
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 fd186d695d8..10526aaec51 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 class WALNode implements IWALNode {
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 743beeeb7ad..812fc0f6f7d 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
@@ -4741,8 +4741,8 @@ public class ConsensusPrefetchingQueue {
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 ce33ee974bb..4b1a58a3009 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.EOFException;
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 @@ public class ProgressWALIterator implements Closeable, Iterator<IndexedConsensus
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 @@ public class ProgressWALIterator implements Closeable, Iterator<IndexedConsensus
private final List<String> 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 @@ public class ProgressWALIterator implements Closeable, Iterator<IndexedConsensus
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 @@ public class ProgressWALIterator implements Closeable, Iterator<IndexedConsensus
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 class ProgressWALIterator implements Closeable, Iterator<IndexedConsensus
+ consumedEntryCountInCurrentFile
+ ", bodyReadAttempts="
+ bodyReadAttempts
+ + ", readerOpenAttempts="
+ + readerOpenAttempts
+ ", minimumSearchIndex="
+ minimumSearchIndex
+ + (retryCurrentReader
+ ? ", retryFile=" + currentReaderFile + ", retryReason=" + currentReaderRetryDetail
+ : "")
+ "]";
}
@@ -558,7 +572,9 @@ public class ProgressWALIterator implements Closeable, Iterator<IndexedConsensus
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 @@ public class ProgressWALIterator implements Closeable, Iterator<IndexedConsensus
}
}
+ 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 @@ public class ProgressWALIterator implements Closeable, Iterator<IndexedConsensus
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 @@ public class ProgressWALIterator implements Closeable, Iterator<IndexedConsensus
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 @@ public class ProgressWALIterator implements Closeable, Iterator<IndexedConsensus
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 @@ public class ProgressWALIterator implements Closeable, Iterator<IndexedConsensus
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 @@ public class ProgressWALIterator implements Closeable, Iterator<IndexedConsensus
// 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 @@ public class ProgressWALIterator implements Closeable, Iterator<IndexedConsensus
}
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 @@ public class ProgressWALIterator implements Closeable, Iterator<IndexedConsensus
}
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 84e61d7f00c..b6f6d8f5bd6 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 java.util.concurrent.TimeUnit;
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 class WALBufferRollRecoveryTest {
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 e1ff4aaeb0b..df12f63caa6 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 class ConsensusPrefetchingQueueDeliveryTest {
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 a663437e360..ad7333751d1 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.assertNull;
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 class ConsensusPrefetchingQueueInitializationTest {
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 class ConsensusPrefetchingQueueInitializationTest {
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 e3de8b31279..7ee54b5f830 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.assertNull;
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;
@@ -1813,7 +1814,7 @@ public class ConsensusPrefetchingQueueTest {
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 -> {
@@ -1876,7 +1877,7 @@ public class ConsensusPrefetchingQueueTest {
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 adcf18a50d8..24d76d2d123 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.assertNull;
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 class ConsensusPrefetchingQueueWalBackpressureTest {
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 @@ public class ConsensusPrefetchingQueueWalBackpressureTest {
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 class ConsensusPrefetchingQueueWalBackpressureTest {
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 class ConsensusPrefetchingQueueWalBackpressureTest {
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 @@ public class ConsensusPrefetchingQueueWalBackpressureTest {
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 b34dd71b6f5..d943ca41241 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 @@ public class ConsensusPrefetchingQueueWriterOrderingTest {
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 370fca53796..a4bfb29304d 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 java.util.concurrent.atomic.AtomicLong;
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 @@ public class ProgressWALIteratorTest {
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 class ProgressWALIteratorTest {
}
}
+ @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 class ProgressWALIteratorTest {
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 class ProgressWALIteratorTest {
}
}
+ @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 class ProgressWALIteratorTest {
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 class ProgressWALIteratorTest {
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());