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 c520debda0b Subscription: improve WAL replay gap observability (#18377)
c520debda0b is described below
commit c520debda0b9d49b2d0c76bbbe33c9e83af10163
Author: Caideyipi <[email protected]>
AuthorDate: Tue Aug 25 18:57:22 2026 +0800
Subscription: improve WAL replay gap observability (#18377)
* Subscription: improve WAL replay gap observability
* [Subscription] Fix WAL iterator initialization
---
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 16 ++++--
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 13 +++--
.../consensus/ConsensusPrefetchingQueue.java | 40 ++++++++++----
.../broker/consensus/ProgressWALIterator.java | 60 ++++++++++++++++++---
.../consensus/ConsensusPrefetchingQueueTest.java | 62 ++++++++++++++++++++++
.../broker/consensus/ProgressWALIteratorTest.java | 38 +++++++++++++
6 files changed, 206 insertions(+), 23 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 4df5d7b4147..a61028e25da 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
@@ -2100,11 +2100,11 @@ public final class DataNodePipeMessages {
public static final String
PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_PREFETCH_INITIALIZED_STARTSEARCHINDEX_69B53EE6
=
"ConsensusPrefetchingQueue {}: prefetch initialized,
startSearchIndex={}, progressSource={}, "
+ "recoveryWriterCount={}";
- public static final String
PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_PERIODIC_STATS_LAG_PENDINGDELTA_D75375D0 =
+ public static final String
PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_PERIODIC_STATS_LAG_PENDINGDELTA_WALGAPSKIPPEDENTRIES_9A4E6608
=
"ConsensusPrefetchingQueue {}: periodic stats, lag={}, pendingDelta={},
walDelta={}, "
- + "pendingTotal={}, walTotal={}, pendingQueueSize={},
prefetchingQueueSize={}, "
- + "inFlightEventsSize={}, realtimeWriterCount={}, walHasNext={},
isActive={}, "
- + "subtaskScheduled={}";
+ + "pendingTotal={}, walTotal={}, walGapSkippedEntries={},
pendingQueueSize={}, "
+ + "prefetchingQueueSize={}, inFlightEventsSize={},
realtimeWriterCount={}, "
+ + "walHasNext={}, isActive={}, subtaskScheduled={}";
public static final String
PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_WAITING_MS_FOR_WAL_GAP_TO_BECOME_7D91C6C5 =
"ConsensusPrefetchingQueue {}: waiting {}ms for WAL gap [{}, {}) to
become visible, "
+ "currentNextExpected={}, currentWalIndex={}, seekGeneration={}";
@@ -2184,6 +2184,14 @@ public final class DataNodePipeMessages {
"ProgressWALIterator: error reading WAL";
public static final String
PIPE_LOG_PROGRESSWALITERATOR_FAILED_TO_OPEN_WAL_FILE_SKIPPING_29CA1092 =
"ProgressWALIterator: failed to open WAL file {}, skipping";
+ public static final String
PIPE_LOG_PROGRESSWALITERATOR_SKIPPED_UNREADABLE_RETAINED_WAL_FILES_FFC8455E =
+ "ProgressWALIterator: skipped {} unreadable retained WAL files in
directory {}, "
+ + "firstFile={}, lastFile={}, firstError={}; historical subscription
data in these "
+ + "files cannot be replayed";
+ public static final String
PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_WAL_REPLAY_SKIPPED_UNAVAILABLE_SEARCH_INDEXES_B8023B64
=
+ "ConsensusPrefetchingQueue {}: WAL replay skipped unavailable search
indexes [{}, {}), "
+ + "skippedEntries={}, totalWalGapSkippedEntries={}; the missing WAL
data may have been "
+ + "reclaimed before subscription consumption";
public static final String
PIPE_LOG_PIPE_TERMINATE_EVENT_COMMITTED_FOR_HISTORICAL_TRANSFER_CREATIONTIME_9B807B28
=
"Pipe {}@{}: terminate event committed for historical transfer.
creationTime: {}, "
+ "shouldMark: {}. {}";
diff --git
a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
index 027e6b4c34c..4de5ee9aa99 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
@@ -1949,10 +1949,11 @@ public final class DataNodePipeMessages {
public static final String
PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_PREFETCH_INITIALIZED_STARTSEARCHINDEX_69B53EE6
=
"ConsensusPrefetchingQueue {}：prefetch
已初始化，startSearchIndex={}，progressSource={}，"
+ "recoveryWriterCount={}";
- public static final String
PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_PERIODIC_STATS_LAG_PENDINGDELTA_D75375D0 =
+ public static final String
PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_PERIODIC_STATS_LAG_PENDINGDELTA_WALGAPSKIPPEDENTRIES_9A4E6608
=
"ConsensusPrefetchingQueue
{}：周期统计，lag={}，pendingDelta={}，walDelta={}，pendingTotal={}，"
- +
"walTotal={}，pendingQueueSize={}，prefetchingQueueSize={}，inFlightEventsSize={}，"
- +
"realtimeWriterCount={}，walHasNext={}，isActive={}，subtaskScheduled={}";
+ +
"walTotal={}，walGapSkippedEntries={}，pendingQueueSize={}，prefetchingQueueSize={}，"
+ +
"inFlightEventsSize={}，realtimeWriterCount={}，walHasNext={}，isActive={}，"
+ + "subtaskScheduled={}";
public static final String
PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_WAITING_MS_FOR_WAL_GAP_TO_BECOME_7D91C6C5 =
"ConsensusPrefetchingQueue {}：等待 {}ms，使 WAL 缺口 [{}, {})
可见，currentNextExpected={}，"
+ "currentWalIndex={}，seekGeneration={}";
@@ -2028,6 +2029,12 @@ public final class DataNodePipeMessages {
"ProgressWALIterator：读取 WAL 出错";
public static final String
PIPE_LOG_PROGRESSWALITERATOR_FAILED_TO_OPEN_WAL_FILE_SKIPPING_29CA1092 =
"ProgressWALIterator：打开 WAL 文件 {} 失败，跳过该文件";
+ public static final String
PIPE_LOG_PROGRESSWALITERATOR_SKIPPED_UNREADABLE_RETAINED_WAL_FILES_FFC8455E =
+ "ProgressWALIterator：跳过了 {} 个无法读取的保留 WAL 文件，directory={}，firstFile={}，"
+ + "lastFile={}，firstError={}；这些文件中的历史订阅数据无法重放";
+ public static final String
PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_WAL_REPLAY_SKIPPED_UNAVAILABLE_SEARCH_INDEXES_B8023B64
=
+ "ConsensusPrefetchingQueue {}：WAL 重放跳过了不可用的 searchIndex 区间 [{}, {})，"
+ + "skippedEntries={}，totalWalGapSkippedEntries={}；缺失的 WAL
数据可能已在订阅消费前被回收";
public static final String
PIPE_LOG_PIPE_TERMINATE_EVENT_COMMITTED_FOR_HISTORICAL_TRANSFER_CREATIONTIME_9B807B28
=
"Pipe {}@{}：历史传输的终止事件已提交。creationTime：{}，shouldMark：{}。{}";
public static final String
PIPE_LOG_PIPE_HISTORICAL_SOURCE_HAS_SUPPLIED_ALL_EVENTS_EMITTING_8B58DE19 =
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java
index a856ac7b942..3a47be814e5 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
@@ -797,11 +797,10 @@ public class ConsensusPrefetchingQueue {
}
this.nextExpectedSearchIndex.set(resolvedStart.getStartSearchIndex());
- if (consensusReqReader instanceof WALNode) {
- this.subscriptionWALIterator =
- new ProgressWALIterator(
- (WALNode) consensusReqReader,
resolvedStart.getStartSearchIndex());
- }
+ // Use the same factory as cursor resets; the default implementation still
returns null for
+ // readers without WAL support.
+ this.subscriptionWALIterator =
+ createSubscriptionWALIterator(resolvedStart.getStartSearchIndex());
this.prefetchInitialized = true;
this.observedSeekGeneration = seekGeneration.get();
discardBatch(this.lingerBatch);
@@ -1529,13 +1528,14 @@ public class ConsensusPrefetchingQueue {
final long currentWalAcceptedEntries = walPathAcceptedEntries.get();
LOGGER.info(
DataNodePipeMessages
-
.PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_PERIODIC_STATS_LAG_PENDINGDELTA_D75375D0,
+
.PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_PERIODIC_STATS_LAG_PENDINGDELTA_WALGAPSKIPPEDENTRIES_9A4E6608,
this,
getLag(),
currentPendingAcceptedEntries - lastPendingAcceptedEntries,
currentWalAcceptedEntries - lastWalAcceptedEntries,
currentPendingAcceptedEntries,
currentWalAcceptedEntries,
+ walGapSkippedEntries.get(),
pendingEntries.size(),
prefetchingQueue.size(),
inFlightEvents.size(),
@@ -1914,11 +1914,11 @@ public class ConsensusPrefetchingQueue {
continue;
}
if (shouldSkipForRecoveryProgress(walEntry)) {
- advanceLocalCursorIfPresent(walEntry);
+ advanceWalReplayCursorIfPresent(walEntry);
continue;
}
if (shouldSkipForMaterializedProgress(walEntry)) {
- advanceLocalCursorIfPresent(walEntry);
+ advanceWalReplayCursorIfPresent(walEntry);
continue;
}
@@ -1932,7 +1932,7 @@ public class ConsensusPrefetchingQueue {
return appendResult;
}
markMaterializedProgress(walEntry);
- advanceLocalCursorIfPresent(walEntry);
+ advanceWalReplayCursorIfPresent(walEntry);
} catch (final Exception e) {
LOGGER.warn(
DataNodePipeMessages
@@ -1954,6 +1954,28 @@ public class ConsensusPrefetchingQueue {
return MaterializationResult.SUCCESS;
}
+ private void advanceWalReplayCursorIfPresent(final IndexedConsensusRequest
request) {
+ if (!hasLocalSearchIndex(request)) {
+ return;
+ }
+
+ final long actualSearchIndex = request.getSearchIndex();
+ final long expectedSearchIndex = nextExpectedSearchIndex.get();
+ if (actualSearchIndex > expectedSearchIndex) {
+ final long skippedEntries = actualSearchIndex - expectedSearchIndex;
+ final long totalSkippedEntries =
walGapSkippedEntries.addAndGet(skippedEntries);
+ LOGGER.warn(
+ DataNodePipeMessages
+
.PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_WAL_REPLAY_SKIPPED_UNAVAILABLE_SEARCH_INDEXES_B8023B64,
+ this,
+ expectedSearchIndex,
+ actualSearchIndex,
+ skippedEntries,
+ totalSkippedEntries);
+ }
+ nextExpectedSearchIndex.set(actualSearchIndex + 1);
+ }
+
private void ensureSubscriptionWalReadable() {
if (Objects.isNull(subscriptionWALIterator) ||
subscriptionWALIterator.hasNext()) {
return;
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 56320a08a82..d9f8ca5e94e 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
@@ -82,6 +82,10 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
private boolean currentReaderUsesLiveSnapshot = false;
private int consumedEntryCountInCurrentFile = 0;
private final Set<Long> skippedBrokenWalVersionIds = new HashSet<>();
+ private int unreportedSkippedBrokenWalFileCount = 0;
+ private String firstUnreportedSkippedBrokenWalFile;
+ private String lastUnreportedSkippedBrokenWalFile;
+ private String firstUnreportedSkippedBrokenWalError;
private IOException lastError;
private boolean incompleteScan = false;
private String incompleteScanDetail;
@@ -188,9 +192,9 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
lastError = e;
LOGGER.warn(
DataNodePipeMessages.PIPE_LOG_PROGRESSWALITERATOR_ERROR_READING_WAL_2DB46D41,
e);
- return false;
}
if (nextReady == null) {
+ logSkippedBrokenWalFilesIfNecessary();
return false;
}
}
@@ -358,6 +362,10 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
return !skippedBrokenWalVersionIds.isEmpty();
}
+ int getSkippedBrokenWalFileCount() {
+ return skippedBrokenWalVersionIds.size();
+ }
+
public boolean hasIncompleteScan() {
return incompleteScan || hasReadError() || hasSkippedBrokenWalFiles();
}
@@ -579,16 +587,54 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
e);
return false;
}
- skippedBrokenWalVersionIds.add(versionId);
- LOGGER.warn(
- DataNodePipeMessages
-
.PIPE_LOG_PROGRESSWALITERATOR_FAILED_TO_OPEN_WAL_FILE_SKIPPING_29CA1092,
- walFile.getName(),
- e);
+ recordSkippedBrokenWalFile(versionId, walFile, e);
return false;
}
}
+ private void recordSkippedBrokenWalFile(
+ final long versionId, final File walFile, final IOException error) {
+ if (!skippedBrokenWalVersionIds.add(versionId)) {
+ return;
+ }
+
+ if (unreportedSkippedBrokenWalFileCount == 0) {
+ firstUnreportedSkippedBrokenWalFile = walFile.getName();
+ firstUnreportedSkippedBrokenWalError = summarizeException(error);
+ }
+ lastUnreportedSkippedBrokenWalFile = walFile.getName();
+ unreportedSkippedBrokenWalFileCount++;
+ LOGGER.debug(
+
DataNodePipeMessages.PIPE_LOG_PROGRESSWALITERATOR_FAILED_TO_OPEN_WAL_FILE_SKIPPING_29CA1092,
+ walFile.getName(),
+ error);
+ }
+
+ private void logSkippedBrokenWalFilesIfNecessary() {
+ if (unreportedSkippedBrokenWalFileCount == 0) {
+ return;
+ }
+
+ LOGGER.warn(
+ DataNodePipeMessages
+
.PIPE_LOG_PROGRESSWALITERATOR_SKIPPED_UNREADABLE_RETAINED_WAL_FILES_FFC8455E,
+ unreportedSkippedBrokenWalFileCount,
+ logDirectory,
+ firstUnreportedSkippedBrokenWalFile,
+ lastUnreportedSkippedBrokenWalFile,
+ firstUnreportedSkippedBrokenWalError);
+ unreportedSkippedBrokenWalFileCount = 0;
+ firstUnreportedSkippedBrokenWalFile = null;
+ lastUnreportedSkippedBrokenWalFile = null;
+ firstUnreportedSkippedBrokenWalError = null;
+ }
+
+ private static String summarizeException(final IOException error) {
+ return error.getMessage() == null
+ ? error.getClass().getSimpleName()
+ : error.getClass().getSimpleName() + ": " + error.getMessage();
+ }
+
private boolean skipEntries(final ProgressWALReader reader, final int
skipEntries)
throws IOException {
return reader.skipToEntryIndex(skipEntries);
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 a2907745179..a6d76a8d98a 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
@@ -63,6 +63,7 @@ import java.lang.reflect.Modifier;
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.Collections;
+import java.util.Iterator;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
@@ -923,6 +924,67 @@ public class ConsensusPrefetchingQueueTest {
}
}
+ @Test
+ public void testWalReplayCountsOnlyUnavailableSearchIndexes() throws
Exception {
+ final String originalSystemDir =
IoTDBDescriptor.getInstance().getConfig().getSystemDir();
+ final File systemDir = temporaryFolder.newFolder("wal-replay-gap-counter");
+ ConsensusPrefetchingQueue queue = null;
+ try {
+ final DataRegionId regionId = new DataRegionId(9);
+ final FakeConsensusReqReader reader = new FakeConsensusReqReader();
+ reader.currentSearchIndex = 4L;
+ final IoTConsensusServerImpl serverImpl =
mock(IoTConsensusServerImpl.class);
+ when(serverImpl.getConsensusReqReader()).thenReturn(reader);
+ when(serverImpl.getWriterSafeFrontierTracker()).thenReturn(new
WriterSafeFrontierTracker());
+
+ final ConsensusLogToTabletConverter converter =
mock(ConsensusLogToTabletConverter.class);
+
when(converter.convert(any())).thenReturn(Collections.singletonList(createTablet()));
+ when(converter.getDatabaseName()).thenReturn("db");
+
+ final Iterator<IndexedConsensusRequest> retainedWalEntries =
+ Arrays.asList(createRequest(1L), createRequest(4L)).iterator();
+ final ProgressWALIterator walIterator = mock(ProgressWALIterator.class);
+ when(walIterator.hasNext()).thenAnswer(ignored ->
retainedWalEntries.hasNext());
+ when(walIterator.next()).thenAnswer(ignored ->
retainedWalEntries.next());
+
+ queue =
+ new ConsensusPrefetchingQueue(
+ "consumerGroup",
+ "topic",
+ TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE,
+ regionId,
+ serverImpl,
+ new SubscriptionWalRetentionPolicy(
+ "topic",
+ SubscriptionWalRetentionPolicy.UNBOUNDED,
+ SubscriptionWalRetentionPolicy.UNBOUNDED),
+ converter,
+ newCommitManager(systemDir),
+ new RegionProgress(Collections.emptyMap()),
+ 1L,
+ 1L,
+ true) {
+ @Override
+ protected ProgressWALIterator createSubscriptionWALIterator(
+ final long startSearchIndex) {
+ return walIterator;
+ }
+ };
+
+ assertNull(queue.poll("consumer"));
+ queue.drivePrefetchOnce();
+
+ assertEquals(2L, queue.getWalPathAcceptedEntries());
+ assertEquals(2L, queue.getWalGapSkippedEntries());
+ assertEquals(5L, queue.getCurrentReadSearchIndex());
+ } finally {
+ if (queue != null) {
+ queue.close();
+ }
+
IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir);
+ }
+ }
+
@Test
public void
testActivationRetriesUntilConfigNodeProgressIsExplicitlyAvailable() throws
Exception {
final String originalSystemDir =
IoTDBDescriptor.getInstance().getConfig().getSystemDir();
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 fed8febea52..7f816937091 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
@@ -278,6 +278,44 @@ public class ProgressWALIteratorTest {
}
}
+ @Test
+ public void testIteratorAggregatesUnreadableRetainedWalFiles() throws
Exception {
+ final Path dir =
Files.createTempDirectory("progress-wal-iterator-unreadable-files");
+ final File firstBrokenWal =
+ dir.resolve(WALFileUtils.getLogFileName(0, 0,
WALFileStatus.CONTAINS_SEARCH_INDEX))
+ .toFile();
+ final File secondBrokenWal =
+ dir.resolve(WALFileUtils.getLogFileName(1, 1,
WALFileStatus.CONTAINS_SEARCH_INDEX))
+ .toFile();
+ final File lastWal =
+ dir.resolve(WALFileUtils.getLogFileName(2, 2,
WALFileStatus.CONTAINS_SEARCH_INDEX))
+ .toFile();
+
+ try {
+ Files.write(firstBrokenWal.toPath(), new byte[128]);
+ Files.write(secondBrokenWal.toPath(), new byte[128]);
+ try (WALWriter writer = new WALWriter(lastWal, WALFileVersion.V3)) {
+ // Create a readable successor so both malformed WAL files are treated
as retained history.
+ writer.write(searchableEntry(2L), singleEntryMeta(19, 2L, 1L, 200L, 7,
2L));
+ }
+
+ try (ProgressWALIterator iterator = new
ProgressWALIterator(dir.toFile(), Long.MIN_VALUE)) {
+ assertTrue(iterator.hasNext());
+ assertEquals(2L, iterator.next().getSearchIndex());
+ assertFalse(iterator.hasNext());
+ assertTrue(iterator.hasSkippedBrokenWalFiles());
+ assertEquals(2, iterator.getSkippedBrokenWalFileCount());
+ assertTrue(iterator.hasIncompleteScan());
+ assertFalse(iterator.hasReadError());
+ }
+ } finally {
+ Files.deleteIfExists(firstBrokenWal.toPath());
+ Files.deleteIfExists(secondBrokenWal.toPath());
+ Files.deleteIfExists(lastWal.toPath());
+ Files.deleteIfExists(dir);
+ }
+ }
+
@Test
public void testLiveWalReopenReusesMetadataSnapshot() throws Exception {
final Path dir =
Files.createTempDirectory("progress-wal-iterator-live-snapshot");
