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 84c4bafa38d [Subscription] Reduce consensus WAL replay and ACK
contention (#18402)
84c4bafa38d is described below
commit 84c4bafa38d7aa8fc61f57fa2b9f29107c843699
Author: Caideyipi <[email protected]>
AuthorDate: Thu Aug 6 12:23:21 2026 +0800
[Subscription] Reduce consensus WAL replay and ACK contention (#18402)
* [Subscription] Reduce consensus WAL replay and ACK contention
* [Subscription] Avoid redundant consensus WAL refresh work
* [Subscription] Fast-forward WAL iterator across pending delivery
---
.../consensus/ConsensusPrefetchingQueue.java | 74 +++--
.../broker/consensus/ProgressWALIterator.java | 251 ++++++++++++---
.../consensus/ConsensusPrefetchingQueueTest.java | 340 +++++++++++++++++++++
.../broker/consensus/ProgressWALIteratorTest.java | 285 +++++++++++++++++
4 files changed, 893 insertions(+), 57 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java
index dd02836cac9..879346dad65 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
@@ -195,9 +195,9 @@ public class ConsensusPrefetchingQueue {
private volatile ProgressWALIterator subscriptionWALIterator;
/**
- * Seek requests must not close/reset the WAL iterator from RPC threads
because the prefetch
- * worker may be reading it concurrently. Instead, seek only records the
latest desired reset and
- * the queue's next prefetch round applies it after observing the new seek
generation.
+ * WAL cursor changes outside the iterator must not close/reset it from RPC
threads because the
+ * prefetch worker may be reading it concurrently. Instead, the latest
desired reset is recorded
+ * and applied by the next prefetch round after observing the expected seek
generation.
*/
private volatile long pendingSubscriptionWalResetSearchIndex =
Long.MIN_VALUE;
@@ -1597,12 +1597,41 @@ public class ConsensusPrefetchingQueue {
return hasLocalSearchIndex(request) && request.getSearchIndex() <
nextExpectedSearchIndex.get();
}
- private void advanceLocalCursorIfPresent(final IndexedConsensusRequest
request) {
+ private boolean advanceLocalCursorIfPresent(final IndexedConsensusRequest
request) {
if (hasLocalSearchIndex(request)) {
nextExpectedSearchIndex.set(request.getSearchIndex() + 1);
+ return true;
+ }
+ return false;
+ }
+
+ private void advanceLocalCursorFromPendingIfPresent(
+ final IndexedConsensusRequest request, final long
expectedSeekGeneration) {
+ if (advanceLocalCursorIfPresent(request)) {
+ // Pending delivery advances independently of the WAL reader. Raise its
local lower bound in
+ // place so stale local requests are filtered without rebuilding and
rescanning retained WAL.
+ final ProgressWALIterator iterator = subscriptionWALIterator;
+ if (Objects.nonNull(iterator) && seekGeneration.get() ==
expectedSeekGeneration) {
+ iterator.advanceTo(
+ nextExpectedSearchIndex.get(),
this::isWriterProgressCoveredForWalFastForward);
+ }
}
}
+ private boolean isWriterProgressCoveredForWalFastForward(
+ final long physicalTime, final int nodeId, final long localSeq) {
+ final WriterProgress candidate = new WriterProgress(physicalTime,
localSeq);
+ final WriterProgress recoveryProgress =
+ recoveryWriterProgressByWriter.get(new
WriterId(consensusGroupId.toString(), nodeId));
+ if (Objects.nonNull(recoveryProgress)
+ && compareWriterProgress(candidate, recoveryProgress) <= 0) {
+ return true;
+ }
+ final WriterProgress materializedProgress =
materializedProgressByWriter.get(nodeId);
+ return Objects.nonNull(materializedProgress)
+ && compareWriterProgress(candidate, materializedProgress) <= 0;
+ }
+
private MaterializationResult appendRealtimeRequest(
final IndexedConsensusRequest request,
final DeliveryBatchState batchState,
@@ -1678,12 +1707,12 @@ public class ConsensusPrefetchingQueue {
if (shouldSkipForRecoveryProgress(request)) {
skippedCount++;
- advanceLocalCursorIfPresent(request);
+ advanceLocalCursorFromPendingIfPresent(request,
expectedSeekGeneration);
continue;
}
if (shouldSkipForMaterializedProgress(request)) {
skippedCount++;
- advanceLocalCursorIfPresent(request);
+ advanceLocalCursorFromPendingIfPresent(request,
expectedSeekGeneration);
continue;
}
@@ -1695,7 +1724,7 @@ public class ConsensusPrefetchingQueue {
}
markMaterializedProgress(request);
processedCount++;
- advanceLocalCursorIfPresent(request);
+ advanceLocalCursorFromPendingIfPresent(request, expectedSeekGeneration);
if (prefetchingQueue.size() >= MAX_PREFETCHING_QUEUE_SIZE) {
break;
}
@@ -1787,7 +1816,9 @@ public class ConsensusPrefetchingQueue {
// Use the persistent linger batch so an unexpected runtime failure cannot
orphan already
// reserved Tablets or advance replay progress past data that has become
unreachable.
final DeliveryBatchState batchState = lingerBatch;
- resetSubscriptionWALPosition(nextExpectedSearchIndex.get());
+ // Keep the iterator and its buffered next request across rounds.
Reopening it here discards the
+ // request prepared by hasNext() and repeatedly re-reads, skips, and
decompresses the same WAL
+ // segment. Pending-path cursor advances and seek operations request
explicit realignment.
final MaterializationResult materializationResult =
pumpFromSubscriptionWAL(
batchState, expectedSeekGeneration, maxWalEntries, maxTablets,
maxBatchBytes);
@@ -1820,7 +1851,6 @@ public class ConsensusPrefetchingQueue {
return MaterializationResult.SUCCESS;
}
- subscriptionWALIterator.refresh();
ensureSubscriptionWalReadable();
int entriesRead = 0;
@@ -1876,9 +1906,14 @@ public class ConsensusPrefetchingQueue {
}
private void ensureSubscriptionWalReadable() {
- if (Objects.isNull(subscriptionWALIterator)
- || subscriptionWALIterator.hasNext()
- || !(consensusReqReader instanceof WALNode)) {
+ if (Objects.isNull(subscriptionWALIterator) ||
subscriptionWALIterator.hasNext()) {
+ return;
+ }
+
+ // Listing and sorting all retained WAL files is only necessary after the
iterator is
+ // exhausted. While it still has a readable request, refreshing cannot
affect the next result.
+ subscriptionWALIterator.refresh();
+ if (subscriptionWALIterator.hasNext() || !(consensusReqReader instanceof
WALNode)) {
return;
}
@@ -1895,9 +1930,6 @@ public class ConsensusPrefetchingQueue {
currentWalIndex);
((WALNode) consensusReqReader).rollWALFile();
resetSubscriptionWALPosition(nextExpectedSearchIndex.get());
- if (Objects.nonNull(subscriptionWALIterator)) {
- subscriptionWALIterator.refresh();
- }
}
private void resetSubscriptionWALPosition(final long startSearchIndex) {
@@ -1915,6 +1947,11 @@ public class ConsensusPrefetchingQueue {
protected void onWalGapRetryScheduled() {}
private boolean hasReadableWalEntries() {
+ if (pendingSubscriptionWalResetSearchIndex != Long.MIN_VALUE) {
+ // Do not advance the stale iterator only to discard its buffered
request when the next round
+ // applies the pending realignment. Returning true keeps the worker
scheduled for that round.
+ return true;
+ }
return Objects.nonNull(subscriptionWALIterator) &&
subscriptionWALIterator.hasNext();
}
@@ -2209,7 +2246,10 @@ public class ConsensusPrefetchingQueue {
private boolean ackMissingInFlightEvent(
final SubscriptionCommitContext commitContext, final boolean silent) {
- acquireWriteLock();
+ // Late or duplicate ACKs touch the same concurrent lifecycle indexes and
commit manager as the
+ // regular in-flight ACK path. A read lock is sufficient to fence
seek/close transitions while
+ // allowing ACKs to proceed concurrently with a long-running WAL prefetch
round.
+ acquireReadLock();
try {
if (!canAcceptCommitContext(commitContext, "ack", silent)) {
return false;
@@ -2256,7 +2296,7 @@ public class ConsensusPrefetchingQueue {
}
return true;
} finally {
- releaseWriteLock();
+ releaseReadLock();
}
}
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 56847c8903c..56320a08a82 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
@@ -42,6 +42,7 @@ import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
@@ -56,6 +57,12 @@ import java.util.Set;
*/
public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensusRequest> {
+ @FunctionalInterface
+ public interface WriterProgressCoverage {
+
+ boolean isCovered(long physicalTime, int nodeId, long localSeq);
+ }
+
private static final Logger LOGGER =
LoggerFactory.getLogger(ProgressWALIterator.class);
private static final int SEARCH_INDEX_OFFSET =
@@ -65,9 +72,10 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
WALFileVersion.V2.getVersionBytes().length,
WALFileVersion.V3.getVersionBytes().length);
private final File logDirectory;
- private final long startSearchIndex;
+ private long minimumSearchIndex;
private final WALNode liveWalNode;
private File[] walFiles;
+ private long[] walFileVersionIds;
private int currentFileIndex = -1;
private ProgressWALReader currentReader;
private long currentReaderVersionId = -1L;
@@ -105,7 +113,7 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
private ProgressWALIterator(
final File logDirectory, final long startSearchIndex, final WALNode
liveWalNode) {
this.logDirectory = logDirectory;
- this.startSearchIndex = startSearchIndex;
+ this.minimumSearchIndex = startSearchIndex;
this.liveWalNode = liveWalNode;
refreshFileList();
}
@@ -114,23 +122,29 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
final File[] discoveredWalFiles =
WALFileUtils.listAllWALFiles(logDirectory);
if (discoveredWalFiles == null) {
walFiles = new File[0];
+ walFileVersionIds = new long[0];
return;
}
WALFileUtils.ascSortByVersionId(discoveredWalFiles);
- final List<File> filteredWalFiles = new
ArrayList<>(discoveredWalFiles.length);
+ final File[] filteredWalFiles = new File[discoveredWalFiles.length];
+ final long[] filteredWalFileVersionIds = new
long[discoveredWalFiles.length];
+ int filteredWalFileCount = 0;
for (int i = 0; i < discoveredWalFiles.length; i++) {
final File walFile = discoveredWalFiles[i];
+ final long versionId = WALFileUtils.parseVersionId(walFile.getName());
final boolean isLastWalFile = i == discoveredWalFiles.length - 1;
- if (!isLastWalFile && shouldSkipWalFile(walFile)) {
+ if (!isLastWalFile && shouldSkipWalFile(walFile, versionId)) {
continue;
}
- filteredWalFiles.add(walFile);
+ filteredWalFiles[filteredWalFileCount] = walFile;
+ filteredWalFileVersionIds[filteredWalFileCount] = versionId;
+ filteredWalFileCount++;
}
- walFiles = filteredWalFiles.toArray(new File[0]);
+ walFiles = Arrays.copyOf(filteredWalFiles, filteredWalFileCount);
+ walFileVersionIds = Arrays.copyOf(filteredWalFileVersionIds,
filteredWalFileCount);
}
- private boolean shouldSkipWalFile(final File walFile) {
- final long versionId = WALFileUtils.parseVersionId(walFile.getName());
+ private boolean shouldSkipWalFile(final File walFile, final long versionId) {
return skippedBrokenWalVersionIds.contains(versionId) ||
isHeaderOnlyWalFile(walFile);
}
@@ -139,42 +153,188 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
}
public void refresh() {
+ final boolean exhaustedKnownFiles = currentFileIndex >= walFiles.length;
final long currentVersionId =
(currentFileIndex >= 0 && currentFileIndex < walFiles.length)
- ? WALFileUtils.parseVersionId(walFiles[currentFileIndex].getName())
- : -1;
+ ? walFileVersionIds[currentFileIndex]
+ : exhaustedKnownFiles && walFileVersionIds.length > 0
+ ? walFileVersionIds[walFileVersionIds.length - 1]
+ : -1;
refreshFileList();
if (currentVersionId >= 0) {
+ final int refreshedIndex = Arrays.binarySearch(walFileVersionIds,
currentVersionId);
+ currentFileIndex = refreshedIndex >= 0 ? refreshedIndex :
-refreshedIndex - 2;
+ } else if (exhaustedKnownFiles) {
currentFileIndex = -1;
- for (int i = 0; i < walFiles.length; i++) {
- if (WALFileUtils.parseVersionId(walFiles[i].getName()) >=
currentVersionId) {
- currentFileIndex = i;
- break;
+ }
+ }
+
+ public boolean hasNext() {
+ while (true) {
+ if (nextReady != null) {
+ if (!shouldSkip(nextReady)) {
+ return true;
}
+ nextReady = null;
}
- if (currentFileIndex < 0) {
- currentFileIndex = walFiles.length;
+ try {
+ nextReady = advance();
+ if (nextReady != null) {
+ lastError = null;
+ }
+ } catch (IOException e) {
+ lastError = e;
+ LOGGER.warn(
+
DataNodePipeMessages.PIPE_LOG_PROGRESSWALITERATOR_ERROR_READING_WAL_2DB46D41,
e);
+ return false;
+ }
+ if (nextReady == null) {
+ return false;
}
}
}
- public boolean hasNext() {
- if (nextReady != null) {
- return true;
+ /**
+ * Advances the local search-index lower bound without rebuilding the
iterator. Whole WAL files
+ * are skipped only when every writer progress tuple in them is already
covered by queue state.
+ */
+ public void advanceTo(
+ final long targetSearchIndex, final WriterProgressCoverage
writerProgressCoverage) {
+ if (targetSearchIndex <= minimumSearchIndex) {
+ return;
}
+ minimumSearchIndex = targetSearchIndex;
+
+ if (writerProgressCoverage == null ||
!canDiscardBufferedRequests(writerProgressCoverage)) {
+ return;
+ }
+
+ refreshForNewerLiveWalFile();
+ int targetFileIndex = locateTargetFile(targetSearchIndex);
+ if (targetFileIndex < 0
+ && liveWalNode != null
+ && targetSearchIndex <= liveWalNode.getCurrentSearchIndex()) {
+ refresh();
+ targetFileIndex = locateTargetFile(targetSearchIndex);
+ }
+ if (targetFileIndex < 0 || targetFileIndex <= currentFileIndex) {
+ return;
+ }
+
+ final int firstFileToSkip = Math.max(0, currentFileIndex);
try {
- nextReady = advance();
- if (nextReady != null) {
- lastError = null;
+ for (int fileIndex = firstFileToSkip; fileIndex < targetFileIndex;
fileIndex++) {
+ if (!isWalFileCovered(fileIndex, writerProgressCoverage)) {
+ return;
+ }
}
- } catch (IOException e) {
- lastError = e;
-
LOGGER.warn(DataNodePipeMessages.PIPE_LOG_PROGRESSWALITERATOR_ERROR_READING_WAL_2DB46D41,
e);
+
+ closeCurrentReader();
+ nextReady = null;
+ pendingRequests.clear();
+ pendingSearchIndex = Long.MIN_VALUE;
+ pendingLocalSeq = Long.MIN_VALUE;
+ currentFileIndex = targetFileIndex - 1;
+ resetCurrentFileTracking();
+ } catch (final IOException ignored) {
+ // Fast-forward is opportunistic. Sequential replay remains the
correctness fallback.
+ }
+ }
+
+ private void refreshForNewerLiveWalFile() {
+ if (walFileVersionIds.length == 0
+ || (liveWalNode != null
+ && walFileVersionIds[walFileVersionIds.length - 1]
+ < liveWalNode.getCurrentWALFileVersion())) {
+ refresh();
+ }
+ }
+
+ private int locateTargetFile(final long targetSearchIndex) {
+ if (walFiles.length == 0
+ || (liveWalNode != null && targetSearchIndex >
liveWalNode.getCurrentSearchIndex())) {
+ return -1;
+ }
+ return WALFileUtils.binarySearchFileBySearchIndex(walFiles,
targetSearchIndex);
+ }
+
+ private boolean canDiscardBufferedRequests(final WriterProgressCoverage
writerProgressCoverage) {
+ return (nextReady == null || isCovered(nextReady, writerProgressCoverage))
+ && (pendingRequests.isEmpty()
+ || isCovered(
+ pendingSearchIndex,
+ pendingPhysicalTime,
+ pendingNodeId,
+ pendingLocalSeq,
+ writerProgressCoverage));
+ }
+
+ private boolean isCovered(
+ final IndexedConsensusRequest request, final WriterProgressCoverage
writerProgressCoverage) {
+ return isCovered(
+ request.getSearchIndex(),
+ request.getPhysicalTime(),
+ request.getNodeId(),
+ request.getProgressLocalSeq(),
+ writerProgressCoverage);
+ }
+
+ private boolean isCovered(
+ final long searchIndex,
+ final long physicalTime,
+ final int nodeId,
+ final long localSeq,
+ final WriterProgressCoverage writerProgressCoverage) {
+ if (searchIndex >= 0 && searchIndex < minimumSearchIndex) {
+ return true;
+ }
+ return nodeId >= 0
+ && physicalTime >= 0
+ && localSeq >= 0
+ && writerProgressCoverage.isCovered(physicalTime, nodeId, localSeq);
+ }
+
+ private boolean isWalFileCovered(
+ final int fileIndex, final WriterProgressCoverage
writerProgressCoverage) throws IOException {
+ final File walFile = walFiles[fileIndex];
+ if (WALFileVersion.getVersion(walFile) != WALFileVersion.V3) {
return false;
}
- return nextReady != null;
+
+ final WALMetaData metadata;
+ final long versionId = walFileVersionIds[fileIndex];
+ if (liveWalNode != null && versionId ==
liveWalNode.getCurrentWALFileVersion()) {
+ metadata = liveWalNode.getCurrentWALMetaDataSnapshot();
+ } else {
+ try (final ProgressWALReader reader = new ProgressWALReader(walFile)) {
+ metadata = reader.getMetaData();
+ }
+ }
+
+ final List<Integer> bufferSizes = metadata.getBuffersSize();
+ final List<Long> physicalTimes = metadata.getPhysicalTimes();
+ final List<Short> nodeIds = metadata.getNodeIds();
+ final List<Long> localSeqs = metadata.getLocalSeqs();
+ if (physicalTimes.size() != bufferSizes.size()
+ || nodeIds.size() != bufferSizes.size()
+ || localSeqs.size() != bufferSizes.size()) {
+ return false;
+ }
+
+ for (int entryIndex = 0; entryIndex < bufferSizes.size(); entryIndex++) {
+ final long physicalTime = physicalTimes.get(entryIndex);
+ final int nodeId = nodeIds.get(entryIndex);
+ final long localSeq = localSeqs.get(entryIndex);
+ if (nodeId < 0
+ || physicalTime < 0
+ || localSeq < 0
+ || !writerProgressCoverage.isCovered(physicalTime, nodeId,
localSeq)) {
+ return false;
+ }
+ }
+ return true;
}
public IndexedConsensusRequest next() {
@@ -337,7 +497,7 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
if (fileIndex < 0) {
return false;
}
- return openReaderAtIndex(fileIndex, consumedEntryCountInCurrentFile);
+ return openReaderAtIndex(fileIndex, consumedEntryCountInCurrentFile,
true, snapshot);
}
final int previousFileIndex =
findFileIndexByVersion(currentReaderVersionId);
@@ -351,21 +511,34 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
}
private boolean openReaderAtIndex(final int fileIndex, final int
skipEntries) throws IOException {
- return openReaderAtIndex(fileIndex, skipEntries, true);
+ return openReaderAtIndex(fileIndex, skipEntries, true, null);
}
private boolean openReaderAtIndex(
final int fileIndex, final int skipEntries, final boolean
allowNearLiveRetry)
throws IOException {
+ return openReaderAtIndex(fileIndex, skipEntries, allowNearLiveRetry, null);
+ }
+
+ private boolean openReaderAtIndex(
+ final int fileIndex,
+ final int skipEntries,
+ final boolean allowNearLiveRetry,
+ final WALMetaData liveMetaDataSnapshot)
+ throws IOException {
final File walFile = walFiles[fileIndex];
- final long versionId = WALFileUtils.parseVersionId(walFile.getName());
+ final long versionId = walFileVersionIds[fileIndex];
final boolean useLiveSnapshot =
liveWalNode != null && versionId ==
liveWalNode.getCurrentWALFileVersion();
try {
final ProgressWALReader reader =
useLiveSnapshot
- ? new ProgressWALReader(walFile,
liveWalNode.getCurrentWALMetaDataSnapshot())
+ ? new ProgressWALReader(
+ walFile,
+ liveMetaDataSnapshot != null
+ ? liveMetaDataSnapshot
+ : liveWalNode.getCurrentWALMetaDataSnapshot())
: new ProgressWALReader(walFile);
if (!skipEntries(reader, skipEntries)) {
reader.close();
@@ -422,18 +595,16 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
}
private int findFileIndexByVersion(final long versionId) {
- for (int i = 0; i < walFiles.length; i++) {
- if (WALFileUtils.parseVersionId(walFiles[i].getName()) == versionId) {
- return i;
- }
- }
- return -1;
+ final int fileIndex = Arrays.binarySearch(walFileVersionIds, versionId);
+ return fileIndex >= 0 ? fileIndex : -1;
}
private boolean openFirstReaderAfterVersion(final long versionId) throws
IOException {
- for (int i = 0; i < walFiles.length; i++) {
- if (WALFileUtils.parseVersionId(walFiles[i].getName()) > versionId
- && openReaderAtIndex(i, 0)) {
+ final int matchedFileIndex = Arrays.binarySearch(walFileVersionIds,
versionId);
+ final int firstFileIndexAfterVersion =
+ matchedFileIndex >= 0 ? matchedFileIndex + 1 : -matchedFileIndex - 1;
+ for (int i = firstFileIndexAfterVersion; i < walFiles.length; i++) {
+ if (openReaderAtIndex(i, 0)) {
return true;
}
}
@@ -495,7 +666,7 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
}
private boolean shouldSkip(final IndexedConsensusRequest request) {
- return request.getSearchIndex() >= 0 && request.getSearchIndex() <
startSearchIndex;
+ return request.getSearchIndex() >= 0 && request.getSearchIndex() <
minimumSearchIndex;
}
private void closeCurrentReader() throws IOException {
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 2656c662e96..845a797f149 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
@@ -65,16 +65,21 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.function.BooleanSupplier;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
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.doAnswer;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -925,6 +930,312 @@ public class ConsensusPrefetchingQueueTest {
}
}
+ @Test
+ public void testLateAckDoesNotWaitForPrefetchReadLock() throws Exception {
+ final String originalSystemDir =
IoTDBDescriptor.getInstance().getConfig().getSystemDir();
+ final File systemDir = temporaryFolder.newFolder("late-ack-read-lock");
+ ConsensusPrefetchingQueue queue = null;
+ try {
+ final DataRegionId regionId = new DataRegionId(1);
+ final FakeConsensusReqReader reader = new FakeConsensusReqReader();
+ 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()),
Collections.emptyList());
+ when(converter.getDatabaseName()).thenReturn("db");
+
+ 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);
+
+ final IndexedConsensusRequest dataRequest =
+ new IndexedConsensusRequest(
+ 1L,
Collections.singletonList(StatementTestUtils.genInsertRowNode(1)))
+ .setPhysicalTime(1000L)
+ .setNodeId(7);
+ final IndexedConsensusRequest requestWithEmptyConversionResult =
+ new IndexedConsensusRequest(
+ 2L,
Collections.singletonList(StatementTestUtils.genInsertRowNode(2)))
+ .setPhysicalTime(1001L)
+ .setNodeId(7);
+ reader.currentSearchIndex = 2L;
+ assertTrue(pendingEntries(queue).offer(dataRequest));
+
assertTrue(pendingEntries(queue).offer(requestWithEmptyConversionResult));
+ assertNull(queue.poll("consumer"));
+ queue.drivePrefetchOnce();
+
+ final SubscriptionEvent event = queue.poll("consumer");
+ assertNotNull(event);
+ assertTrue(queue.ackSilent("consumer", event.getCommitContext()));
+
+ final ConsensusPrefetchingQueue activeQueue = queue;
+ final ReentrantReadWriteLock queueLock = queueLock(queue);
+ final CountDownLatch ackStarted = new CountDownLatch(1);
+ final CountDownLatch ackCompleted = new CountDownLatch(1);
+ final AtomicReference<Boolean> lateAckResult = new AtomicReference<>();
+ final AtomicReference<Throwable> asyncFailure = new AtomicReference<>();
+ final Thread lateAckThread =
+ new Thread(
+ () -> {
+ ackStarted.countDown();
+ try {
+ lateAckResult.set(activeQueue.ackSilent("consumer",
event.getCommitContext()));
+ } catch (final Throwable t) {
+ asyncFailure.set(t);
+ } finally {
+ ackCompleted.countDown();
+ }
+ });
+ lateAckThread.setDaemon(true);
+
+ final boolean completedWhileReadLocked;
+ queueLock.readLock().lock();
+ try {
+ lateAckThread.start();
+ assertTrue(ackStarted.await(5, TimeUnit.SECONDS));
+ completedWhileReadLocked = ackCompleted.await(5, TimeUnit.SECONDS);
+ } finally {
+ queueLock.readLock().unlock();
+ }
+ lateAckThread.join(TimeUnit.SECONDS.toMillis(5));
+
+ assertTrue(completedWhileReadLocked);
+ assertFalse(lateAckThread.isAlive());
+ if (asyncFailure.get() != null) {
+ throw new AssertionError(asyncFailure.get());
+ }
+ assertTrue(Boolean.TRUE.equals(lateAckResult.get()));
+ } finally {
+ if (queue != null) {
+ queue.close();
+ }
+
IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir);
+ }
+ }
+
+ @Test
+ public void testWalCatchUpReusesIteratorAcrossRounds() throws Exception {
+ final String originalSystemDir =
IoTDBDescriptor.getInstance().getConfig().getSystemDir();
+ final File systemDir = temporaryFolder.newFolder("reuse-wal-iterator");
+ ConsensusPrefetchingQueue queue = null;
+ try {
+ final FakeConsensusReqReader reader = new FakeConsensusReqReader();
+ final IoTConsensusServerImpl serverImpl =
mock(IoTConsensusServerImpl.class);
+ when(serverImpl.getConsensusReqReader()).thenReturn(reader);
+ when(serverImpl.getWriterSafeFrontierTracker()).thenReturn(new
WriterSafeFrontierTracker());
+ queue =
+ new ConsensusPrefetchingQueue(
+ "consumerGroup",
+ "topic",
+ TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE,
+ new DataRegionId(1),
+ serverImpl,
+ new SubscriptionWalRetentionPolicy(
+ "topic",
+ SubscriptionWalRetentionPolicy.UNBOUNDED,
+ SubscriptionWalRetentionPolicy.UNBOUNDED),
+ mock(ConsensusLogToTabletConverter.class),
+ newCommitManager(systemDir),
+ new RegionProgress(Collections.emptyMap()),
+ 1L,
+ 1L,
+ true);
+
+ final ProgressWALIterator iterator = mock(ProgressWALIterator.class);
+ when(iterator.hasNext()).thenReturn(false);
+ setSubscriptionWalIterator(queue, iterator);
+
+ final Method tryCatchUpFromWAL =
+
ConsensusPrefetchingQueue.class.getDeclaredMethod("tryCatchUpFromWAL",
long.class);
+ tryCatchUpFromWAL.setAccessible(true);
+ tryCatchUpFromWAL.invoke(queue, queue.getCurrentSeekGeneration());
+ tryCatchUpFromWAL.invoke(queue, queue.getCurrentSeekGeneration());
+
+ assertSame(iterator, subscriptionWalIterator(queue));
+ verify(iterator, times(2)).refresh();
+ verify(iterator, never()).close();
+ } finally {
+ if (queue != null) {
+ queue.close();
+ }
+
IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir);
+ }
+ }
+
+ @Test
+ public void testReadableWalIteratorSkipsFileListRefresh() throws Exception {
+ final String originalSystemDir =
IoTDBDescriptor.getInstance().getConfig().getSystemDir();
+ final File systemDir =
temporaryFolder.newFolder("skip-readable-wal-refresh");
+ ConsensusPrefetchingQueue queue = null;
+ try {
+ final FakeConsensusReqReader reader = new FakeConsensusReqReader();
+ final IoTConsensusServerImpl serverImpl =
mock(IoTConsensusServerImpl.class);
+ when(serverImpl.getConsensusReqReader()).thenReturn(reader);
+ when(serverImpl.getWriterSafeFrontierTracker()).thenReturn(new
WriterSafeFrontierTracker());
+ queue =
+ new ConsensusPrefetchingQueue(
+ "consumerGroup",
+ "topic",
+ TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE,
+ new DataRegionId(1),
+ serverImpl,
+ new SubscriptionWalRetentionPolicy(
+ "topic",
+ SubscriptionWalRetentionPolicy.UNBOUNDED,
+ SubscriptionWalRetentionPolicy.UNBOUNDED),
+ mock(ConsensusLogToTabletConverter.class),
+ newCommitManager(systemDir),
+ new RegionProgress(Collections.emptyMap()),
+ 1L,
+ 1L,
+ true);
+
+ final ProgressWALIterator iterator = mock(ProgressWALIterator.class);
+ when(iterator.hasNext()).thenReturn(true);
+ setSubscriptionWalIterator(queue, iterator);
+
+ invokeEnsureSubscriptionWalReadable(queue);
+
+ verify(iterator).hasNext();
+ verify(iterator, never()).refresh();
+ } finally {
+ if (queue != null) {
+ queue.close();
+ }
+
IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir);
+ }
+ }
+
+ @Test
+ public void testWalRollDoesNotRefreshNewIteratorTwice() throws Exception {
+ final String originalSystemDir =
IoTDBDescriptor.getInstance().getConfig().getSystemDir();
+ final File systemDir =
temporaryFolder.newFolder("skip-new-iterator-refresh");
+ final File walDirectory =
temporaryFolder.newFolder("skip-new-iterator-refresh-wal");
+ ConsensusPrefetchingQueue queue = null;
+ try {
+ final WALNode walNode = mock(WALNode.class);
+ when(walNode.getLogDirectory()).thenReturn(walDirectory);
+ when(walNode.getCurrentSearchIndex()).thenReturn(1L);
+ final IoTConsensusServerImpl serverImpl =
mock(IoTConsensusServerImpl.class);
+ when(serverImpl.getConsensusReqReader()).thenReturn(walNode);
+ when(serverImpl.getWriterSafeFrontierTracker()).thenReturn(new
WriterSafeFrontierTracker());
+
+ final ProgressWALIterator replacementIterator =
mock(ProgressWALIterator.class);
+ queue =
+ new ConsensusPrefetchingQueue(
+ "consumerGroup",
+ "topic",
+ TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE,
+ new DataRegionId(1),
+ serverImpl,
+ new SubscriptionWalRetentionPolicy(
+ "topic",
+ SubscriptionWalRetentionPolicy.UNBOUNDED,
+ SubscriptionWalRetentionPolicy.UNBOUNDED),
+ mock(ConsensusLogToTabletConverter.class),
+ newCommitManager(systemDir),
+ new RegionProgress(Collections.emptyMap()),
+ 1L,
+ 1L,
+ true) {
+ @Override
+ protected ProgressWALIterator createSubscriptionWALIterator(
+ final long startSearchIndex) {
+ assertEquals(1L, startSearchIndex);
+ return replacementIterator;
+ }
+ };
+
+ final ProgressWALIterator exhaustedIterator =
mock(ProgressWALIterator.class);
+ when(exhaustedIterator.hasNext()).thenReturn(false);
+ setSubscriptionWalIterator(queue, exhaustedIterator);
+
+ invokeEnsureSubscriptionWalReadable(queue);
+
+ verify(exhaustedIterator, times(2)).hasNext();
+ verify(exhaustedIterator).refresh();
+ verify(exhaustedIterator).close();
+ verify(walNode).rollWALFile();
+ verify(replacementIterator, never()).refresh();
+ assertSame(replacementIterator, subscriptionWalIterator(queue));
+ } finally {
+ if (queue != null) {
+ queue.close();
+ }
+
IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir);
+ }
+ }
+
+ @Test
+ public void testPendingCursorAdvanceFastForwardsWalIteratorInPlace() throws
Exception {
+ final String originalSystemDir =
IoTDBDescriptor.getInstance().getConfig().getSystemDir();
+ final File systemDir =
temporaryFolder.newFolder("deferred-wal-realignment");
+ ConsensusPrefetchingQueue queue = null;
+ try {
+ final FakeConsensusReqReader reader = new FakeConsensusReqReader();
+ final IoTConsensusServerImpl serverImpl =
mock(IoTConsensusServerImpl.class);
+ when(serverImpl.getConsensusReqReader()).thenReturn(reader);
+ when(serverImpl.getWriterSafeFrontierTracker()).thenReturn(new
WriterSafeFrontierTracker());
+ queue =
+ new ConsensusPrefetchingQueue(
+ "consumerGroup",
+ "topic",
+ TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE,
+ new DataRegionId(1),
+ serverImpl,
+ new SubscriptionWalRetentionPolicy(
+ "topic",
+ SubscriptionWalRetentionPolicy.UNBOUNDED,
+ SubscriptionWalRetentionPolicy.UNBOUNDED),
+ mock(ConsensusLogToTabletConverter.class),
+ newCommitManager(systemDir),
+ new RegionProgress(Collections.emptyMap()),
+ 1L,
+ 1L,
+ true);
+
+ final ProgressWALIterator staleIterator =
mock(ProgressWALIterator.class);
+ setSubscriptionWalIterator(queue, staleIterator);
+
+ final Method advancePendingCursor =
+ ConsensusPrefetchingQueue.class.getDeclaredMethod(
+ "advanceLocalCursorFromPendingIfPresent",
IndexedConsensusRequest.class, long.class);
+ advancePendingCursor.setAccessible(true);
+ advancePendingCursor.invoke(
+ queue,
+ new IndexedConsensusRequest(7L, Collections.emptyList()),
+ queue.getCurrentSeekGeneration());
+
+ assertEquals(8L, queue.getCurrentReadSearchIndex());
+ verify(staleIterator)
+ .advanceTo(anyLong(),
any(ProgressWALIterator.WriterProgressCoverage.class));
+ verify(staleIterator, never()).close();
+ assertSame(staleIterator, subscriptionWalIterator(queue));
+ } finally {
+ if (queue != null) {
+ queue.close();
+ }
+
IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir);
+ }
+ }
+
@Test
public void testCleanupReconcilesUnindexedTabletReservation() throws
Exception {
final String originalSystemDir =
IoTDBDescriptor.getInstance().getConfig().getSystemDir();
@@ -1412,6 +1723,35 @@ public class ConsensusPrefetchingQueueTest {
return (BlockingQueue<IndexedConsensusRequest>) field.get(queue);
}
+ private static ReentrantReadWriteLock queueLock(final
ConsensusPrefetchingQueue queue)
+ throws Exception {
+ final Field field =
ConsensusPrefetchingQueue.class.getDeclaredField("lock");
+ field.setAccessible(true);
+ return (ReentrantReadWriteLock) field.get(queue);
+ }
+
+ private static ProgressWALIterator subscriptionWalIterator(final
ConsensusPrefetchingQueue queue)
+ throws Exception {
+ final Field field =
ConsensusPrefetchingQueue.class.getDeclaredField("subscriptionWALIterator");
+ field.setAccessible(true);
+ return (ProgressWALIterator) field.get(queue);
+ }
+
+ private static void setSubscriptionWalIterator(
+ final ConsensusPrefetchingQueue queue, final ProgressWALIterator
iterator) throws Exception {
+ final Field field =
ConsensusPrefetchingQueue.class.getDeclaredField("subscriptionWALIterator");
+ field.setAccessible(true);
+ field.set(queue, iterator);
+ }
+
+ private static void invokeEnsureSubscriptionWalReadable(final
ConsensusPrefetchingQueue queue)
+ throws Exception {
+ final Method method =
+
ConsensusPrefetchingQueue.class.getDeclaredMethod("ensureSubscriptionWalReadable");
+ method.setAccessible(true);
+ method.invoke(queue);
+ }
+
private static Tablet createTablet() {
final List<String> columnNames = Arrays.asList("device", "temperature");
final List<TSDataType> dataTypes = Arrays.asList(TSDataType.STRING,
TSDataType.DOUBLE);
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 b37542e936d..92928663126 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
@@ -30,6 +30,7 @@ import
org.apache.iotdb.db.storageengine.dataregion.wal.node.WALNode;
import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALFileStatus;
import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALFileUtils;
+import org.junit.Assume;
import org.junit.Test;
import java.io.File;
@@ -41,6 +42,8 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
public class ProgressWALIteratorTest {
@@ -272,6 +275,41 @@ public class ProgressWALIteratorTest {
}
}
+ @Test
+ public void testLiveWalReopenReusesMetadataSnapshot() throws Exception {
+ final Path dir =
Files.createTempDirectory("progress-wal-iterator-live-snapshot");
+ final File liveWal =
+ dir.resolve(WALFileUtils.getLogFileName(0, 0,
WALFileStatus.CONTAINS_SEARCH_INDEX))
+ .toFile();
+
+ try {
+ try (WALWriter writer = new WALWriter(liveWal, WALFileVersion.V3)) {
+ writer.write(searchableEntry(1L), singleEntryMeta(19, 1L, 1L, 1000L,
7, 1L));
+ writer.write(searchableEntry(2L), singleEntryMeta(19, 2L, 1L, 1001L,
7, 2L));
+ }
+
+ final WALMetaData firstSnapshot = singleEntryMeta(19, 1L, 1L, 1000L, 7,
1L);
+ final WALMetaData secondSnapshot = firstSnapshot.copy();
+ secondSnapshot.add(19, 2L, 1L, 1001L, 7, 2L);
+
+ final WALNode walNode = mock(WALNode.class);
+ when(walNode.getLogDirectory()).thenReturn(dir.toFile());
+ when(walNode.getCurrentWALFileVersion()).thenReturn(0L);
+ when(walNode.getCurrentWALMetaDataSnapshot()).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();
+ }
+ } finally {
+ Files.deleteIfExists(liveWal.toPath());
+ Files.deleteIfExists(dir);
+ }
+ }
+
@Test
public void testIteratorMarksIncompleteScanWhenNearLiveWalCannotBeOpened()
throws Exception {
final Path dir =
Files.createTempDirectory("progress-wal-iterator-incomplete-scan");
@@ -299,6 +337,253 @@ public class ProgressWALIteratorTest {
}
}
+ @Test
+ public void testAdvanceToKeepsUncoveredFollowerRequestInCurrentFile() throws
Exception {
+ final Path dir =
Files.createTempDirectory("progress-wal-iterator-current-file-advance");
+ final File dataWal =
+ dir.resolve(WALFileUtils.getLogFileName(0, 0,
WALFileStatus.CONTAINS_SEARCH_INDEX))
+ .toFile();
+ final File successorWal =
+ dir.resolve(WALFileUtils.getLogFileName(1, 3,
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));
+ }
+ try (WALWriter writer = new WALWriter(successorWal, WALFileVersion.V3)) {
+ writer.write(searchableEntry(-1L), singleEntryMeta(19, -1L, 1L, 200L,
8, 20L));
+ 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());
+
+ iterator.advanceTo(3L, (physicalTime, nodeId, localSeq) -> false);
+
+ assertTrue(iterator.hasNext());
+ final IndexedConsensusRequest followerRequest = iterator.next();
+ assertEquals(-1L, followerRequest.getSearchIndex());
+ assertEquals(8, followerRequest.getNodeId());
+ assertTrue(iterator.hasNext());
+ assertEquals(3L, iterator.next().getSearchIndex());
+ }
+ } finally {
+ Files.deleteIfExists(dataWal.toPath());
+ Files.deleteIfExists(successorWal.toPath());
+ Files.deleteIfExists(dir);
+ }
+ }
+
+ @Test
+ public void testAdvanceToSkipsOnlyFullyCoveredWalFiles() throws Exception {
+ final Path dir =
Files.createTempDirectory("progress-wal-iterator-covered-file-advance");
+ final File localWal =
+ dir.resolve(WALFileUtils.getLogFileName(0, 0,
WALFileStatus.CONTAINS_SEARCH_INDEX))
+ .toFile();
+ final File followerWal =
+ dir.resolve(WALFileUtils.getLogFileName(1, 2,
WALFileStatus.CONTAINS_NONE_SEARCH_INDEX))
+ .toFile();
+ final File targetWal =
+ dir.resolve(WALFileUtils.getLogFileName(2, 2,
WALFileStatus.CONTAINS_SEARCH_INDEX))
+ .toFile();
+
+ try {
+ try (WALWriter writer = new WALWriter(localWal, WALFileVersion.V3)) {
+ writer.write(searchableEntry(1L), singleEntryMeta(19, 1L, 1L, 100L, 7,
1L));
+ writer.write(searchableEntry(2L), singleEntryMeta(19, 2L, 1L, 200L, 7,
2L));
+ }
+ try (WALWriter writer = new WALWriter(followerWal, WALFileVersion.V3)) {
+ writer.write(searchableEntry(-1L), singleEntryMeta(19, -1L, 1L, 300L,
8, 30L));
+ }
+ try (WALWriter writer = new WALWriter(targetWal, WALFileVersion.V3)) {
+ writer.write(searchableEntry(6L), singleEntryMeta(19, 6L, 1L, 600L, 7,
6L));
+ writer.write(searchableEntry(7L), singleEntryMeta(19, 7L, 1L, 700L, 7,
7L));
+ }
+
+ try (ProgressWALIterator iterator = new
ProgressWALIterator(dir.toFile(), 1L)) {
+ assertTrue(iterator.hasNext());
+ assertEquals(1L, iterator.next().getSearchIndex());
+
+ iterator.advanceTo(6L, (physicalTime, nodeId, localSeq) -> nodeId ==
7);
+
+ assertTrue(iterator.hasNext());
+ final IndexedConsensusRequest uncoveredFollower = iterator.next();
+ assertEquals(-1L, uncoveredFollower.getSearchIndex());
+ assertEquals(8, uncoveredFollower.getNodeId());
+ }
+
+ try (ProgressWALIterator iterator = new
ProgressWALIterator(dir.toFile(), 1L)) {
+ assertTrue(iterator.hasNext());
+ assertEquals(1L, iterator.next().getSearchIndex());
+
+ iterator.advanceTo(6L, (physicalTime, nodeId, localSeq) -> true);
+
+ assertTrue(iterator.hasNext());
+ assertEquals(6L, iterator.next().getSearchIndex());
+ }
+ } finally {
+ Files.deleteIfExists(localWal.toPath());
+ Files.deleteIfExists(followerWal.toPath());
+ Files.deleteIfExists(targetWal.toPath());
+ Files.deleteIfExists(dir);
+ }
+ }
+
+ @Test
+ public void testRefreshAfterExhaustionDiscoversNextWalFile() throws
Exception {
+ final Path dir =
Files.createTempDirectory("progress-wal-iterator-refresh-after-exhaustion");
+ final File firstWal =
+ dir.resolve(WALFileUtils.getLogFileName(0, 0,
WALFileStatus.CONTAINS_SEARCH_INDEX))
+ .toFile();
+ final File secondWal =
+ dir.resolve(WALFileUtils.getLogFileName(1, 1,
WALFileStatus.CONTAINS_SEARCH_INDEX))
+ .toFile();
+
+ try {
+ try (WALWriter writer = new WALWriter(firstWal, WALFileVersion.V3)) {
+ writer.write(searchableEntry(1L), singleEntryMeta(19, 1L, 1L, 100L, 7,
1L));
+ }
+
+ try (ProgressWALIterator iterator = new
ProgressWALIterator(dir.toFile(), 1L)) {
+ assertTrue(iterator.hasNext());
+ assertEquals(1L, iterator.next().getSearchIndex());
+ assertFalse(iterator.hasNext());
+
+ try (WALWriter writer = new WALWriter(secondWal, WALFileVersion.V3)) {
+ writer.write(searchableEntry(2L), singleEntryMeta(19, 2L, 1L, 200L,
7, 2L));
+ }
+ iterator.refresh();
+
+ assertTrue(iterator.hasNext());
+ assertEquals(2L, iterator.next().getSearchIndex());
+ }
+ } finally {
+ Files.deleteIfExists(firstWal.toPath());
+ Files.deleteIfExists(secondWal.toPath());
+ Files.deleteIfExists(dir);
+ }
+ }
+
+ @Test
+ public void testIteratorReusePerformance() throws Exception {
+ Assume.assumeTrue(
+ "Enable with -Diotdb.test.subscription.performance=true",
+ Boolean.getBoolean("iotdb.test.subscription.performance"));
+
+ final int entryCount =
Integer.getInteger("iotdb.test.subscription.performance.entries", 4096);
+ final int batchSize =
Integer.getInteger("iotdb.test.subscription.performance.batch-size", 64);
+ assertTrue("entry count must be positive", entryCount > 0);
+ assertTrue("batch size must be positive", batchSize > 0);
+ final Path dir =
Files.createTempDirectory("progress-wal-iterator-performance");
+ final File dataWal =
+ dir.resolve(WALFileUtils.getLogFileName(0, 0,
WALFileStatus.CONTAINS_SEARCH_INDEX))
+ .toFile();
+ final File successorWal =
+ dir.resolve(
+ WALFileUtils.getLogFileName(
+ 1, entryCount + 1L, WALFileStatus.CONTAINS_SEARCH_INDEX))
+ .toFile();
+
+ try {
+ try (WALWriter writer = new WALWriter(dataWal, WALFileVersion.V3)) {
+ for (long index = 1; index <= entryCount; index++) {
+ writer.write(searchableEntry(index), singleEntryMeta(19, index, 1L,
index, 7, index));
+ }
+ }
+ try (WALWriter ignored = new WALWriter(successorWal, WALFileVersion.V3))
{
+ // Seal the data WAL so both benchmark variants read the same
historical file.
+ }
+
+ assertEquals(entryCount, consumeWithReusedIterator(dir.toFile(),
entryCount));
+
+ final long reopenStartNanos = System.nanoTime();
+ assertEquals(entryCount, consumeWithReopenedIterator(dir.toFile(),
entryCount, batchSize));
+ final long reopenNanos = System.nanoTime() - reopenStartNanos;
+
+ final long refreshStartNanos = System.nanoTime();
+ assertEquals(entryCount, consumeWithRefreshedIterator(dir.toFile(),
entryCount, batchSize));
+ final long refreshNanos = System.nanoTime() - refreshStartNanos;
+
+ final long reuseStartNanos = System.nanoTime();
+ assertEquals(entryCount, consumeWithReusedIterator(dir.toFile(),
entryCount));
+ final long reuseNanos = System.nanoTime() - reuseStartNanos;
+
+ final double reopenSpeedup = (double) reopenNanos / Math.max(1L,
reuseNanos);
+ final double refreshSpeedup = (double) refreshNanos / Math.max(1L,
reuseNanos);
+ System.out.printf(
+ "Subscription WAL iterator benchmark: entries=%d, batchSize=%d, "
+ + "reopen=%.3f ms, refreshEachBatch=%.3f ms, reuse=%.3f ms, "
+ + "reopenSpeedup=%.2fx, refreshSpeedup=%.2fx%n",
+ entryCount,
+ batchSize,
+ reopenNanos / 1_000_000.0,
+ refreshNanos / 1_000_000.0,
+ reuseNanos / 1_000_000.0,
+ reopenSpeedup,
+ refreshSpeedup);
+ assertTrue(
+ "Reusing the iterator should be faster than reopening each batch",
reopenSpeedup > 1.0);
+ } finally {
+ Files.deleteIfExists(dataWal.toPath());
+ Files.deleteIfExists(successorWal.toPath());
+ Files.deleteIfExists(dir);
+ }
+ }
+
+ private static int consumeWithReopenedIterator(
+ final File walDirectory, final int entryCount, final int batchSize)
throws Exception {
+ int consumed = 0;
+ long nextSearchIndex = 1L;
+ while (consumed < entryCount) {
+ int batchCount = 0;
+ try (ProgressWALIterator iterator = new
ProgressWALIterator(walDirectory, nextSearchIndex)) {
+ while (batchCount < batchSize && iterator.hasNext()) {
+ nextSearchIndex = iterator.next().getSearchIndex() + 1L;
+ batchCount++;
+ consumed++;
+ }
+ }
+ if (batchCount == 0) {
+ break;
+ }
+ }
+ return consumed;
+ }
+
+ private static int consumeWithReusedIterator(final File walDirectory, final
int entryCount)
+ throws Exception {
+ int consumed = 0;
+ try (ProgressWALIterator iterator = new ProgressWALIterator(walDirectory,
1L)) {
+ while (consumed < entryCount && iterator.hasNext()) {
+ iterator.next();
+ consumed++;
+ }
+ }
+ return consumed;
+ }
+
+ private static int consumeWithRefreshedIterator(
+ final File walDirectory, final int entryCount, final int batchSize)
throws Exception {
+ int consumed = 0;
+ try (ProgressWALIterator iterator = new ProgressWALIterator(walDirectory,
1L)) {
+ while (consumed < entryCount) {
+ iterator.refresh();
+ int batchCount = 0;
+ while (batchCount < batchSize && iterator.hasNext()) {
+ iterator.next();
+ batchCount++;
+ consumed++;
+ }
+ if (batchCount == 0) {
+ break;
+ }
+ }
+ }
+ return consumed;
+ }
+
private static ByteBuffer searchableEntry(final long bodySearchIndex) {
final ByteBuffer buffer =
ByteBuffer.allocate(WALInfoEntry.FIXED_SERIALIZED_SIZE +
PlanNodeType.BYTES + Long.BYTES);
