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 0c0116afe22 Bound consensus subscription WAL scans per prefetch round
(#18849)
0c0116afe22 is described below
commit 0c0116afe22532f32312a3cb30352a818fba4e38
Author: Caideyipi <[email protected]>
AuthorDate: Wed Oct 14 16:10:57 2026 +0800
Bound consensus subscription WAL scans per prefetch round (#18849)
---
.../consensus/ConsensusPrefetchingQueue.java | 100 ++++++++++-----
.../broker/consensus/ProgressWALIterator.java | 78 +++++++++++-
...onsensusPrefetchingQueueInitializationTest.java | 5 +-
.../ConsensusPrefetchingQueueSeekTest.java | 3 +-
.../consensus/ConsensusPrefetchingQueueTest.java | 35 ++++--
...nsensusPrefetchingQueueWalBackpressureTest.java | 140 +++++++++++++++++++--
.../broker/consensus/ProgressWALIteratorTest.java | 110 ++++++++++++++++
7 files changed, 414 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 63c06f06960..a678646a735 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
@@ -386,6 +386,7 @@ public class ConsensusPrefetchingQueue {
SUCCESS,
MEMORY_BLOCKED,
WAL_GAP,
+ WAL_SCAN_YIELDED,
STALE
}
@@ -1546,6 +1547,8 @@ public class ConsensusPrefetchingQueue {
private static final long WAL_GAP_RETRY_SLEEP_MS = 10L;
private static final long MEMORY_RETRY_SLEEP_MS = 100L;
+ // Yield long WAL scans to other regions without closing readers or
splitting writer requests.
+ private static final long WAL_SCAN_TIME_SLICE_NANOS =
TimeUnit.MILLISECONDS.toNanos(50L);
private static final long WAL_GAP_WAIT_LOG_INTERVAL_MS = 5_000L;
private static final long PREFETCH_STATS_LOG_INTERVAL_MS = 5_000L;
@@ -1619,6 +1622,7 @@ public class ConsensusPrefetchingQueue {
acquireReadLock();
try {
+ final ProgressWALIterator.ScanBudget walScanBudget = newWalScanBudget();
if (isClosed || closeRequested || !prefetchInitialized) {
return PrefetchRoundResult.dormant();
}
@@ -1642,10 +1646,10 @@ public class ConsensusPrefetchingQueue {
if (!isActive) {
blockRealtimeAdmission(SubscriptionQueueRejectionReason.INACTIVE_OR_CLOSED);
- return computeIdleRoundResult();
+ return computeIdleRoundResult(walScanBudget);
}
if (prefetchingQueue.size() >= MAX_PREFETCHING_QUEUE_SIZE) {
- return computeIdleRoundResult();
+ return computeIdleRoundResult(walScanBudget);
}
final SubscriptionConfig config = SubscriptionConfig.getInstance();
@@ -1666,7 +1670,7 @@ public class ConsensusPrefetchingQueue {
if (!realtimeEntriesByWriter.isEmpty()) {
blockRealtimeAdmission(SubscriptionQueueRejectionReason.WRITER_BACKLOG);
}
- return computeIdleRoundResult();
+ return computeIdleRoundResult(walScanBudget);
}
if (shouldWaitForSubscriptionMemory()) {
blockRealtimeAdmission(memoryBlockReason);
@@ -1699,11 +1703,18 @@ public class ConsensusPrefetchingQueue {
observedSeekGeneration,
maxWalEntries,
maxTablets,
- maxBatchBytes);
+ maxBatchBytes,
+ walScanBudget);
} finally {
pendingEntries.release(batch);
}
if (batchResult != MaterializationResult.SUCCESS) {
+ if (batchResult == MaterializationResult.WAL_SCAN_YIELDED) {
+ if (!lingerBatch.isEmpty() && !flushBatch(lingerBatch,
observedSeekGeneration)) {
+ resetRoundStateForSeek(seekGeneration.get());
+ }
+ return PrefetchRoundResult.rescheduleNow();
+ }
if (batchResult == MaterializationResult.WAL_GAP) {
if (!lingerBatch.isEmpty() && !flushBatch(lingerBatch,
observedSeekGeneration)) {
resetRoundStateForSeek(seekGeneration.get());
@@ -1727,7 +1738,11 @@ public class ConsensusPrefetchingQueue {
}
if (batch.isEmpty() && lingerBatch.isEmpty()) {
- final MaterializationResult walResult =
tryCatchUpFromWAL(observedSeekGeneration);
+ final MaterializationResult walResult =
+ tryCatchUpFromWAL(observedSeekGeneration, walScanBudget);
+ if (walResult == MaterializationResult.WAL_SCAN_YIELDED) {
+ return PrefetchRoundResult.rescheduleNow();
+ }
if (walResult == MaterializationResult.MEMORY_BLOCKED) {
blockRealtimeAdmission(memoryBlockReason);
return PrefetchRoundResult.rescheduleAfter(MEMORY_RETRY_SLEEP_MS);
@@ -1763,7 +1778,7 @@ public class ConsensusPrefetchingQueue {
}
maybeInjectWatermark();
- return computeIdleRoundResult();
+ return computeIdleRoundResult(walScanBudget);
} catch (final Throwable fatal) {
LOGGER.error(
DataNodePipeMessages
@@ -1832,14 +1847,21 @@ public class ConsensusPrefetchingQueue {
return batch;
}
- private PrefetchRoundResult computeIdleRoundResult() {
+ private static ProgressWALIterator.ScanBudget newWalScanBudget() {
+ return new ProgressWALIterator.ScanBudget(
+
SubscriptionConfig.getInstance().getSubscriptionConsensusBatchMaxWalEntries(),
+ WAL_SCAN_TIME_SLICE_NANOS);
+ }
+
+ private PrefetchRoundResult computeIdleRoundResult(
+ final ProgressWALIterator.ScanBudget walScanBudget) {
if (isClosed || !prefetchInitialized || !isActive) {
return PrefetchRoundResult.dormant();
}
if (prefetchingQueue.size() >= MAX_PREFETCHING_QUEUE_SIZE) {
return PrefetchRoundResult.dormant();
}
- if (hasImmediatePrefetchableWork()) {
+ if (hasImmediatePrefetchableWork(walScanBudget)) {
return PrefetchRoundResult.rescheduleNow();
}
long delayMs = Long.MAX_VALUE;
@@ -1887,9 +1909,13 @@ public class ConsensusPrefetchingQueue {
}
private boolean hasImmediatePrefetchableWork() {
+ return hasImmediatePrefetchableWork(newWalScanBudget());
+ }
+
+ private boolean hasImmediatePrefetchableWork(final
ProgressWALIterator.ScanBudget walScanBudget) {
return !pendingEntries.isEmpty()
|| !realtimeEntriesByWriter.isEmpty()
- || hasReadableWalEntries();
+ || hasReadableWalEntries(walScanBudget);
}
private boolean hasHistoricalWalLag() {
@@ -1967,7 +1993,8 @@ public class ConsensusPrefetchingQueue {
final long expectedSeekGeneration,
final int maxWalEntries,
final int maxTablets,
- final long maxBatchBytes) {
+ final long maxBatchBytes,
+ final ProgressWALIterator.ScanBudget walScanBudget) {
int processedCount = 0;
int skippedCount = 0;
@@ -1996,7 +2023,8 @@ public class ConsensusPrefetchingQueue {
expectedSeekGeneration,
maxWalEntries,
maxTablets,
- maxBatchBytes);
+ maxBatchBytes,
+ walScanBudget);
if (gapFillResult != MaterializationResult.SUCCESS) {
return gapFillResult;
}
@@ -2075,7 +2103,8 @@ public class ConsensusPrefetchingQueue {
final long expectedSeekGeneration,
final int maxWalEntries,
final int maxTablets,
- final long maxBatchBytes) {
+ final long maxBatchBytes,
+ final ProgressWALIterator.ScanBudget walScanBudget) {
pendingWalGapRetryRequested = false;
if (seekGeneration.get() != expectedSeekGeneration || isClosed) {
return MaterializationResult.STALE;
@@ -2093,7 +2122,12 @@ public class ConsensusPrefetchingQueue {
// queue read lock indefinitely and block unsubscribe from acquiring the
close write lock.
final MaterializationResult pumpResult =
pumpFromSubscriptionWAL(
- batchState, expectedSeekGeneration, maxWalEntries, maxTablets,
maxBatchBytes);
+ batchState,
+ expectedSeekGeneration,
+ maxWalEntries,
+ maxTablets,
+ maxBatchBytes,
+ walScanBudget);
if (pumpResult != MaterializationResult.SUCCESS) {
return pumpResult;
}
@@ -2132,7 +2166,8 @@ public class ConsensusPrefetchingQueue {
* Try catch-up from WAL when the pending queue was empty. This handles
cold-start or scenarios
* where the subscription started after data was already written.
*/
- private MaterializationResult tryCatchUpFromWAL(final long
expectedSeekGeneration) {
+ private MaterializationResult tryCatchUpFromWAL(
+ final long expectedSeekGeneration, final ProgressWALIterator.ScanBudget
walScanBudget) {
final SubscriptionConfig config = SubscriptionConfig.getInstance();
final int maxTablets =
config.getSubscriptionConsensusBatchMaxTabletCount();
final long maxBatchBytes =
config.getSubscriptionConsensusBatchMaxSizeInBytes();
@@ -2146,8 +2181,14 @@ public class ConsensusPrefetchingQueue {
// segment. Pending-path cursor advances and seek operations request
explicit realignment.
final MaterializationResult materializationResult =
pumpFromSubscriptionWAL(
- batchState, expectedSeekGeneration, maxWalEntries, maxTablets,
maxBatchBytes);
- if (materializationResult != MaterializationResult.SUCCESS) {
+ batchState,
+ expectedSeekGeneration,
+ maxWalEntries,
+ maxTablets,
+ maxBatchBytes,
+ walScanBudget);
+ if (materializationResult != MaterializationResult.SUCCESS
+ && materializationResult != MaterializationResult.WAL_SCAN_YIELDED) {
if (materializationResult == MaterializationResult.MEMORY_BLOCKED &&
!batchState.isEmpty()) {
if (!flushBatch(batchState, expectedSeekGeneration)) {
discardBatch(batchState);
@@ -2163,7 +2204,7 @@ public class ConsensusPrefetchingQueue {
discardBatch(batchState);
return MaterializationResult.STALE;
}
- return MaterializationResult.SUCCESS;
+ return materializationResult;
}
private MaterializationResult pumpFromSubscriptionWAL(
@@ -2171,7 +2212,8 @@ public class ConsensusPrefetchingQueue {
final long expectedSeekGeneration,
final int maxWalEntries,
final int maxTablets,
- final long maxBatchBytes) {
+ final long maxBatchBytes,
+ final ProgressWALIterator.ScanBudget walScanBudget) {
if (Objects.isNull(subscriptionWALIterator)) {
return MaterializationResult.SUCCESS;
}
@@ -2185,13 +2227,13 @@ public class ConsensusPrefetchingQueue {
maxWalEntries,
prefetchingQueue.size());
}
- ensureSubscriptionWalReadable();
+ ensureSubscriptionWalReadable(walScanBudget);
reportUnreadableWalReplayIfNecessary();
int entriesRead = 0;
while (entriesRead < maxWalEntries
- && subscriptionWALIterator.hasNext()
- && prefetchingQueue.size() < MAX_PREFETCHING_QUEUE_SIZE) {
+ && prefetchingQueue.size() < MAX_PREFETCHING_QUEUE_SIZE
+ && subscriptionWALIterator.hasNext(walScanBudget)) {
reportUnreadableWalReplayIfNecessary();
try {
final IndexedConsensusRequest walEntry =
subscriptionWALIterator.next();
@@ -2245,7 +2287,9 @@ public class ConsensusPrefetchingQueue {
entriesRead,
nextExpectedSearchIndex.get());
}
- return MaterializationResult.SUCCESS;
+ return walScanBudget.hasYielded()
+ ? MaterializationResult.WAL_SCAN_YIELDED
+ : MaterializationResult.SUCCESS;
}
private void reportUnreadableWalReplayIfNecessary() {
@@ -2298,11 +2342,11 @@ public class ConsensusPrefetchingQueue {
nextExpectedSearchIndex.set(actualSearchIndex + 1);
}
- private void ensureSubscriptionWalReadable() {
+ private void ensureSubscriptionWalReadable(final
ProgressWALIterator.ScanBudget walScanBudget) {
if (Objects.isNull(subscriptionWALIterator)) {
return;
}
- if (subscriptionWALIterator.hasNext()) {
+ if (subscriptionWALIterator.hasNext(walScanBudget) ||
walScanBudget.hasYielded()) {
reportUnreadableWalReplayIfNecessary();
return;
}
@@ -2311,7 +2355,7 @@ public class ConsensusPrefetchingQueue {
// 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()) {
+ if (subscriptionWALIterator.hasNext(walScanBudget) ||
walScanBudget.hasYielded()) {
reportUnreadableWalReplayIfNecessary();
return;
}
@@ -2355,7 +2399,7 @@ public class ConsensusPrefetchingQueue {
protected void onWalGapRetryScheduled() {}
- private boolean hasReadableWalEntries() {
+ private boolean hasReadableWalEntries(final ProgressWALIterator.ScanBudget
walScanBudget) {
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.
@@ -2364,9 +2408,9 @@ public class ConsensusPrefetchingQueue {
if (Objects.isNull(subscriptionWALIterator)) {
return false;
}
- final boolean readable = subscriptionWALIterator.hasNext();
+ final boolean readable = subscriptionWALIterator.hasNext(walScanBudget);
reportUnreadableWalReplayIfNecessary();
- return readable;
+ return readable || walScanBudget.hasYielded();
}
private void requestSubscriptionWalReset(
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 3636829274f..3f5186893ee 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
@@ -51,6 +51,7 @@ import java.util.List;
import java.util.NoSuchElementException;
import java.util.Objects;
import java.util.Set;
+import java.util.function.LongSupplier;
/**
* Writer-based WAL iterator for the new subscription progress model.
@@ -66,6 +67,41 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
boolean isCovered(long physicalTime, int nodeId, long localSeq);
}
+ /**
+ * One shared budget for all WAL probes and reads in a prefetch round. The
time limit is
+ * cooperative: it is checked between operations and cannot interrupt a
single file read.
+ */
+ static final class ScanBudget {
+
+ private int remainingEntries;
+ private final long startNanos;
+ private final long maxDurationNanos;
+ private final LongSupplier nanoTime;
+ private boolean yielded;
+
+ ScanBudget(final int maxEntries, final long maxDurationNanos) {
+ this(maxEntries, maxDurationNanos, System::nanoTime);
+ }
+
+ ScanBudget(final int maxEntries, final long maxDurationNanos, final
LongSupplier nanoTime) {
+ this.remainingEntries = maxEntries;
+ this.maxDurationNanos = maxDurationNanos;
+ this.nanoTime = nanoTime;
+ this.startNanos = nanoTime.getAsLong();
+ }
+
+ private boolean shouldYield() {
+ if (remainingEntries <= 0 || nanoTime.getAsLong() - startNanos >=
maxDurationNanos) {
+ yielded = true;
+ }
+ return yielded;
+ }
+
+ boolean hasYielded() {
+ return yielded;
+ }
+ }
+
private static final Logger LOGGER =
LoggerFactory.getLogger(ProgressWALIterator.class);
private static final int MAX_SKIPPED_FILE_DETAILS = 8;
@@ -104,6 +140,7 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
private final List<IConsensusRequest> pendingRequests = new ArrayList<>();
private IndexedConsensusRequest nextReady;
+ private boolean scanYielded;
public ProgressWALIterator(final File logDirectory) {
this(logDirectory, Long.MIN_VALUE);
@@ -183,6 +220,15 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
}
public boolean hasNext() {
+ return hasNext(null);
+ }
+
+ /**
+ * A false result can also mean that the round's scan budget was consumed.
In that case the reader
+ * position and incomplete request fragments are preserved for the next
round.
+ */
+ boolean hasNext(final ScanBudget scanBudget) {
+ scanYielded = false;
while (true) {
if (nextReady != null) {
if (!shouldSkip(nextReady)) {
@@ -191,7 +237,7 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
nextReady = null;
}
try {
- nextReady = advance();
+ nextReady = advance(scanBudget);
if (nextReady != null) {
lastError = null;
}
@@ -214,6 +260,10 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
return nextReady != null && !shouldSkip(nextReady);
}
+ boolean hasYieldedScan() {
+ return scanYielded;
+ }
+
/**
* Keeps a request returned by {@link #next()} available for a retry without
reopening the WAL.
* The caller uses this when materialization is temporarily blocked and
replay progress has not
@@ -433,11 +483,16 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
lastError = null;
incompleteScan = false;
incompleteScanDetail = null;
+ scanYielded = false;
resetCurrentFileTracking();
}
- private IndexedConsensusRequest advance() throws IOException {
+ private IndexedConsensusRequest advance(final ScanBudget scanBudget) throws
IOException {
while (true) {
+ if (scanBudget != null && scanBudget.shouldYield()) {
+ scanYielded = true;
+ return null;
+ }
if (retryCurrentReader) {
refresh();
final int retryFileIndex =
findFileIndexByVersion(currentReaderVersionId);
@@ -455,6 +510,11 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
}
if (currentReader != null && currentReader.hasNext()) {
try {
+ if (scanBudget != null) {
+ // Count physical entries, including skipped requests and request
fragments. A single
+ // hasNext() can otherwise read an arbitrary WAL prefix before
returning a request.
+ scanBudget.remainingEntries--;
+ }
final ByteBuffer buffer = currentReader.next();
consumedEntryCountInCurrentFile =
currentReader.getCurrentEntryIndex() + 1;
final WALEntryType type = WALEntryType.valueOf(buffer.get());
@@ -529,8 +589,8 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
continue;
}
- if (!openNextReader()) {
- if (retryCurrentReader) {
+ if (!openNextReader(scanBudget)) {
+ if (scanYielded || retryCurrentReader) {
return null;
}
final IndexedConsensusRequest flushed = flushPending();
@@ -542,8 +602,13 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
}
}
- private boolean openNextReader() throws IOException {
- while (++currentFileIndex < walFiles.length) {
+ private boolean openNextReader(final ScanBudget scanBudget) throws
IOException {
+ while (currentFileIndex + 1 < walFiles.length) {
+ if (scanBudget != null && scanBudget.shouldYield()) {
+ scanYielded = true;
+ return false;
+ }
+ currentFileIndex++;
if (openReaderAtIndex(currentFileIndex, 0)) {
return true;
}
@@ -551,6 +616,7 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
return false;
}
}
+ currentFileIndex = walFiles.length;
return false;
}
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 93b805c3032..a663437e360 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
@@ -210,7 +210,10 @@ public class ConsensusPrefetchingQueueInitializationTest {
assertNull(queue.poll("consumer"));
assertEquals(1L, queue.getInitializedStatus());
assertEquals(2L, queue.getCurrentReadSearchIndex());
- queue.drivePrefetchOnce();
+ // WAL replay may yield when the round's time slice expires before
reaching the tail.
+ for (int round = 0; round < 10 && queue.getCurrentReadSearchIndex() < 4L;
round++) {
+ queue.drivePrefetchOnce();
+ }
final ArgumentCaptor<InsertNode> requests =
ArgumentCaptor.forClass(InsertNode.class);
verify(converter, times(2)).convert(requests.capture());
assertEquals(
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueSeekTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueSeekTest.java
index 54fcb704f00..ab5aff9b0d7 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueSeekTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueSeekTest.java
@@ -164,7 +164,8 @@ public class ConsensusPrefetchingQueueSeekTest {
|| request.getSearchIndex() >=
startSearchIndex)
.iterator();
final ProgressWALIterator iterator =
mock(ProgressWALIterator.class);
- when(iterator.hasNext()).thenAnswer(ignored ->
retainedEntries.hasNext());
+ when(iterator.hasNext(any(ProgressWALIterator.ScanBudget.class)))
+ .thenAnswer(ignored -> retainedEntries.hasNext());
when(iterator.next()).thenAnswer(ignored ->
retainedEntries.next());
return iterator;
}
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 0b9e12eda48..7f9c4139040 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
@@ -1758,7 +1758,8 @@ public class ConsensusPrefetchingQueueTest {
createRequest(5L))
.iterator();
final ProgressWALIterator iterator = mock(ProgressWALIterator.class);
- when(iterator.hasNext()).thenAnswer(ignored -> walEntries.hasNext());
+ when(iterator.hasNext(any(ProgressWALIterator.ScanBudget.class)))
+ .thenAnswer(ignored -> walEntries.hasNext());
when(iterator.next()).thenAnswer(ignored -> walEntries.next());
queue =
@@ -1916,7 +1917,8 @@ public class ConsensusPrefetchingQueueTest {
createRequest(5L))
.iterator();
final ProgressWALIterator iterator =
mock(ProgressWALIterator.class);
- when(iterator.hasNext()).thenAnswer(ignored ->
walEntries.hasNext());
+ when(iterator.hasNext(any(ProgressWALIterator.ScanBudget.class)))
+ .thenAnswer(ignored -> walEntries.hasNext());
when(iterator.next()).thenAnswer(ignored -> walEntries.next());
return iterator;
}
@@ -2264,14 +2266,21 @@ public class ConsensusPrefetchingQueueTest {
true);
final ProgressWALIterator iterator = mock(ProgressWALIterator.class);
- when(iterator.hasNext()).thenReturn(false);
+
when(iterator.hasNext(any(ProgressWALIterator.ScanBudget.class))).thenReturn(false);
setSubscriptionWalIterator(queue, iterator);
final Method tryCatchUpFromWAL =
-
ConsensusPrefetchingQueue.class.getDeclaredMethod("tryCatchUpFromWAL",
long.class);
+ ConsensusPrefetchingQueue.class.getDeclaredMethod(
+ "tryCatchUpFromWAL", long.class,
ProgressWALIterator.ScanBudget.class);
tryCatchUpFromWAL.setAccessible(true);
- tryCatchUpFromWAL.invoke(queue, queue.getCurrentSeekGeneration());
- tryCatchUpFromWAL.invoke(queue, queue.getCurrentSeekGeneration());
+ tryCatchUpFromWAL.invoke(
+ queue,
+ queue.getCurrentSeekGeneration(),
+ new ProgressWALIterator.ScanBudget(128, Long.MAX_VALUE));
+ tryCatchUpFromWAL.invoke(
+ queue,
+ queue.getCurrentSeekGeneration(),
+ new ProgressWALIterator.ScanBudget(128, Long.MAX_VALUE));
assertSame(iterator, subscriptionWalIterator(queue));
verify(iterator, times(2)).refresh();
@@ -2313,12 +2322,12 @@ public class ConsensusPrefetchingQueueTest {
true);
final ProgressWALIterator iterator = mock(ProgressWALIterator.class);
- when(iterator.hasNext()).thenReturn(true);
+
when(iterator.hasNext(any(ProgressWALIterator.ScanBudget.class))).thenReturn(true);
setSubscriptionWalIterator(queue, iterator);
invokeEnsureSubscriptionWalReadable(queue);
- verify(iterator).hasNext();
+ verify(iterator).hasNext(any(ProgressWALIterator.ScanBudget.class));
verify(iterator, never()).refresh();
} finally {
if (queue != null) {
@@ -2369,12 +2378,12 @@ public class ConsensusPrefetchingQueueTest {
};
final ProgressWALIterator exhaustedIterator =
mock(ProgressWALIterator.class);
- when(exhaustedIterator.hasNext()).thenReturn(false);
+
when(exhaustedIterator.hasNext(any(ProgressWALIterator.ScanBudget.class))).thenReturn(false);
setSubscriptionWalIterator(queue, exhaustedIterator);
invokeEnsureSubscriptionWalReadable(queue);
- verify(exhaustedIterator, times(2)).hasNext();
+ verify(exhaustedIterator,
times(2)).hasNext(any(ProgressWALIterator.ScanBudget.class));
verify(exhaustedIterator).refresh();
verify(exhaustedIterator).close();
verify(walNode).rollWALFile();
@@ -2428,6 +2437,7 @@ public class ConsensusPrefetchingQueueTest {
verify(iterator).hasBufferedNext();
verify(iterator, never()).hasNext();
+ verify(iterator,
never()).hasNext(any(ProgressWALIterator.ScanBudget.class));
verify(iterator, never()).next();
verify(iterator, never()).refresh();
assertEquals(1L, queue.getCurrentReadSearchIndex());
@@ -3203,9 +3213,10 @@ public class ConsensusPrefetchingQueueTest {
private static void invokeEnsureSubscriptionWalReadable(final
ConsensusPrefetchingQueue queue)
throws Exception {
final Method method =
-
ConsensusPrefetchingQueue.class.getDeclaredMethod("ensureSubscriptionWalReadable");
+ ConsensusPrefetchingQueue.class.getDeclaredMethod(
+ "ensureSubscriptionWalReadable",
ProgressWALIterator.ScanBudget.class);
method.setAccessible(true);
- method.invoke(queue);
+ method.invoke(queue, new ProgressWALIterator.ScanBudget(128,
Long.MAX_VALUE));
}
private static void invokeReportUnreadableWalReplayIfNecessary(
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 bb7b2e59371..377f20801c4 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
@@ -73,6 +73,8 @@ import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
public class ConsensusPrefetchingQueueWalBackpressureTest {
@@ -82,6 +84,118 @@ public class ConsensusPrefetchingQueueWalBackpressureTest {
@Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder();
+ @Test
+ public void testYieldedGapScanKeepsIteratorAndDeliversEachRequestOnce()
throws Exception {
+ final String originalSystemDir =
IoTDBDescriptor.getInstance().getConfig().getSystemDir();
+ final CommonConfig config = CommonDescriptor.getInstance().getConfig();
+ final int originalBatchMaxWalEntries =
config.getSubscriptionConsensusBatchMaxWalEntries();
+ final int originalBatchMaxTabletCount =
config.getSubscriptionConsensusBatchMaxTabletCount();
+ final int originalBatchMaxDelay =
config.getSubscriptionConsensusBatchMaxDelayInMs();
+ final File systemDir = temporaryFolder.newFolder("system-yielded-wal-gap");
+ final File walDirectory = temporaryFolder.newFolder("yielded-wal-gap");
+ ConsensusPrefetchingQueue queue = null;
+ try {
+ config.setSubscriptionConsensusBatchMaxWalEntries(2);
+ config.setSubscriptionConsensusBatchMaxTabletCount(1);
+ config.setSubscriptionConsensusBatchMaxDelayInMs(0);
+ final File dataWal =
+ new File(
+ walDirectory,
+ WALFileUtils.getLogFileName(0L, 0L,
WALFileStatus.CONTAINS_SEARCH_INDEX));
+ final File liveWal =
+ new File(
+ walDirectory,
+ WALFileUtils.getLogFileName(1L, 3L,
WALFileStatus.CONTAINS_SEARCH_INDEX));
+ try (WALWriter writer = new WALWriter(dataWal, WALFileVersion.V3)) {
+ // The replay cursor is 2, but its WAL file has a fragmented prefix
for local index 1.
+ for (int fragment = 0; fragment < 9; fragment++) {
+ writeWalEntry(writer, 1L);
+ }
+ writeWalEntry(writer, 2L);
+ writeWalEntry(writer, 3L);
+ }
+ try (WALWriter ignored = new WALWriter(liveWal, WALFileVersion.V3)) {
+ // Keep a sealed successor so the fragmented prefix can be read.
+ }
+
+ final WALNode walNode = mock(WALNode.class);
+ when(walNode.getLogDirectory()).thenReturn(walDirectory);
+ when(walNode.getCurrentSearchIndex()).thenReturn(3L);
+ when(walNode.getCurrentWALFileVersion()).thenReturn(1L);
+ when(walNode.getCurrentWALMetaDataSnapshot()).thenReturn(new
WALMetaData());
+ final IoTConsensusServerImpl serverImpl =
mock(IoTConsensusServerImpl.class);
+ when(serverImpl.getConsensusReqReader()).thenReturn(walNode);
+ when(serverImpl.getWriterSafeFrontierTracker()).thenReturn(new
WriterSafeFrontierTracker());
+ final List<Long> conversions = new ArrayList<>();
+ final ConsensusLogToTabletConverter converter =
mock(ConsensusLogToTabletConverter.class);
+ when(converter.convert(any()))
+ .thenAnswer(
+ invocation -> {
+ conversions.add(((InsertNode)
invocation.getArgument(0)).getSearchIndex());
+ return Collections.singletonList(createTablet());
+ });
+ when(converter.getDatabaseName()).thenReturn("db");
+ queue =
+ new ConsensusPrefetchingQueue(
+ "consumerGroup",
+ "topic",
+ TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE,
+ new DataRegionId(1),
+ serverImpl,
+ new SubscriptionWalRetentionPolicy(
+ "topic",
+ SubscriptionWalRetentionPolicy.UNBOUNDED,
+ SubscriptionWalRetentionPolicy.UNBOUNDED),
+ converter,
+ newCommitManager(systemDir),
+ new RegionProgress(Collections.emptyMap()),
+ 2L,
+ 2L,
+ true);
+ final SubscriptionMemoryManager memoryManager =
+ new SubscriptionMemoryManager(createTablet().ramBytesUsed() * 8L);
+ queue.setSubscriptionMemoryManager(memoryManager);
+ assertNull(queue.poll("consumer"));
+ assertTrue(pendingEntries(queue).offer(createRequest(3L)));
+
+ assertEquals(PrefetchRoundResult.Type.RESCHEDULE_NOW,
queue.drivePrefetchOnce().getType());
+ final ProgressWALIterator iterator = subscriptionWalIterator(queue);
+ assertTrue(iterator.hasYieldedScan());
+ assertTrue(conversions.isEmpty());
+ assertEquals(2L, queue.getCurrentReadSearchIndex());
+ verify(walNode, never()).rollWALFile();
+
+ // Repeated realtime jumps must continue the yielded scan rather than
reset it to index 2.
+ for (int round = 0; round < 20 && queue.getCurrentReadSearchIndex() <
4L; round++) {
+ if (queue.getCurrentReadSearchIndex() < 3L) {
+ assertTrue(pendingEntries(queue).offer(createRequest(3L)));
+ }
+ queue.drivePrefetchOnce();
+ assertSame(iterator, subscriptionWalIterator(queue));
+ }
+ assertEquals(4L, queue.getCurrentReadSearchIndex());
+ assertEquals(Arrays.asList(2L, 3L), conversions);
+ for (long expected = 2L; expected <= 3L; expected++) {
+ final SubscriptionEvent event = queue.poll("consumer");
+ assertNotNull(event);
+ assertEquals(expected,
event.getCommitContext().getWriterProgress().getLocalSeq());
+ assertTrue(queue.ack("consumer", event.getCommitContext()));
+ }
+ queue.drivePrefetchOnce();
+ assertNull(queue.poll("consumer"));
+ assertEquals(0L, memoryManager.getUsedMemorySizeInBytes());
+ verify(walNode, never()).rollWALFile();
+ } finally {
+ if (queue != null) {
+ queue.close();
+ }
+
config.setSubscriptionConsensusBatchMaxWalEntries(originalBatchMaxWalEntries);
+
config.setSubscriptionConsensusBatchMaxTabletCount(originalBatchMaxTabletCount);
+ config.setSubscriptionConsensusBatchMaxDelayInMs(originalBatchMaxDelay);
+
IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir);
+ }
+ }
+
@Test
public void testAckRecoversDrainedSuffixFromWalWithoutReoffer() throws
Exception {
final String originalSystemDir =
IoTDBDescriptor.getInstance().getConfig().getSystemDir();
@@ -369,8 +483,8 @@ public class ConsensusPrefetchingQueueWalBackpressureTest {
config.setSubscriptionConsensusBatchMaxDelayInMs(60_000);
final int replayedRequests = materializationFailure ? 7 : 6;
- final int rounds = memoryBackpressure ? replayedRequests :
(replayedRequests + 1) / 2;
- final int requestCount = replayedRequests + 2 * rounds;
+ final int maxRounds = 20;
+ final int requestCount = replayedRequests + 2 * maxRounds;
writeSealedWal(walDirectory, requestCount);
final WALNode walNode = mock(WALNode.class);
@@ -444,15 +558,20 @@ public class ConsensusPrefetchingQueueWalBackpressureTest
{
}
long deliveredLocalSeq = 0L;
- for (int round = 0; round < rounds; round++) {
+ for (int round = 0; round < maxRounds && deliveredLocalSeq <
replayedRequests; round++) {
+ // Physical scan limits include lookahead and skipped prefixes; a
deadline can also end a
+ // round early. Bound the final logical batch so replay stops at the
asserted frontier.
+ final int maxWalEntries =
+ memoryBackpressure ? 2 : (int) Math.min(2L, replayedRequests -
deliveredLocalSeq);
+ config.setSubscriptionConsensusBatchMaxWalEntries(maxWalEntries);
+ final long memoryRejections =
queue.getSubscriptionMemoryRejectionCount();
queue.drivePrefetchOnce();
- final long expectedLocalSeq =
- materializationFailure
- ? round * 2L + 1L
- : memoryBackpressure ? round + 1L : (round + 1L) * 2L;
- assertEquals(expectedLocalSeq + 1L, queue.getCurrentReadSearchIndex());
+ final long expectedLocalSeq = queue.getCurrentReadSearchIndex() - 1L;
+ assertTrue(expectedLocalSeq >= deliveredLocalSeq);
+ assertTrue(expectedLocalSeq <= replayedRequests);
+ assertTrue(expectedLocalSeq - deliveredLocalSeq <= maxWalEntries);
assertMemoryBounded(queue, memoryManager);
- if (memoryBackpressure) {
+ if (queue.getSubscriptionMemoryRejectionCount() > memoryRejections) {
assertTrue(subscriptionWalIterator(queue).hasBufferedNext());
}
while (deliveredLocalSeq < expectedLocalSeq) {
@@ -475,6 +594,9 @@ public class ConsensusPrefetchingQueueWalBackpressureTest {
assertEquals(0L, queue.getWalGapSkippedEntries());
assertEquals(0L, queue.getSubscriptionUncommittedEventCount());
assertEquals(0L, queue.getRetainedTabletBytes());
+ if (memoryBackpressure) {
+ assertTrue(queue.getSubscriptionMemoryRejectionCount() > 0L);
+ }
// Grouping reads one request ahead; a blocked request can require one
further lookahead.
// Count actual body reads, including discarded iterators, rather than
timing disk caches.
// An exceptional conversion failure needs one reset, rereading its
three consumed bodies.
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 d6159e6c692..cd136e4cc72 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
@@ -47,6 +47,7 @@ import java.nio.file.Files;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.List;
+import java.util.concurrent.atomic.AtomicLong;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
@@ -58,6 +59,115 @@ import static org.mockito.Mockito.when;
public class ProgressWALIteratorTest {
+ @Test
+ public void
testBoundedScanKeepsSkippedPrefixAndPendingFragmentsAcrossRounds() throws
Exception {
+ final Path dir = Files.createTempDirectory("progress-wal-bounded-scan");
+ 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)) {
+ for (int fragment = 0; fragment < 5; fragment++) {
+ writer.write(searchableEntry(1L), singleEntryMeta(19, 1L, 1L, 1000L,
7, 1L));
+ }
+ for (int fragment = 0; fragment < 5; fragment++) {
+ writer.write(
+ searchableEntry(SearchNode.NO_CONSENSUS_INDEX),
+ singleEntryMeta(19, SearchNode.NO_CONSENSUS_INDEX, 1L, 2000L, 8,
50L));
+ }
+ writer.write(searchableEntry(2L), singleEntryMeta(19, 2L, 1L, 3000L,
7, 2L));
+ }
+ try (WALWriter ignored = new WALWriter(successorWal, WALFileVersion.V3))
{
+ // Seal a successor so replay reads the historical file.
+ }
+
+ try (ProgressWALIterator iterator = new
ProgressWALIterator(dir.toFile(), 2L)) {
+ for (int round = 0; round < 5; round++) {
+ final ProgressWALIterator.ScanBudget budget =
+ new ProgressWALIterator.ScanBudget(2, Long.MAX_VALUE);
+ assertFalse(iterator.hasNext(budget));
+ assertTrue(budget.hasYielded());
+ assertTrue(iterator.hasYieldedScan());
+ assertFalse(iterator.hasBufferedNext());
+ assertFalse(iterator.hasIncompleteScan());
+ // Neither a skipped prefix nor an unfinished follower request is a
completed delivery.
+ assertFalse(iterator.hasNext(budget));
+ }
+
+ final ProgressWALIterator.ScanBudget budget =
+ new ProgressWALIterator.ScanBudget(2, Long.MAX_VALUE);
+ assertTrue(iterator.hasNext(budget));
+ final IndexedConsensusRequest follower = iterator.next();
+ assertEquals(SearchNode.NO_CONSENSUS_INDEX, follower.getSearchIndex());
+ assertEquals(8, follower.getNodeId());
+ assertEquals(50L, follower.getProgressLocalSeq());
+ assertEquals(5, follower.getRequests().size());
+ // Repeated probes share the same budget and retain the final pending
local request.
+ assertTrue(iterator.hasNext(budget));
+ assertEquals(2L, iterator.next().getSearchIndex());
+ assertFalse(iterator.hasNext(budget));
+ assertFalse(budget.hasYielded());
+ assertFalse(iterator.hasIncompleteScan());
+ }
+ } finally {
+ Files.deleteIfExists(dataWal.toPath());
+ Files.deleteIfExists(successorWal.toPath());
+ Files.deleteIfExists(dir);
+ }
+ }
+
+ @Test
+ public void testElapsedScanBudgetYieldsWithoutFlushingIncompleteRequest()
throws Exception {
+ final Path dir = Files.createTempDirectory("progress-wal-scan-time-slice");
+ final File dataWal =
+ 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();
+ final AtomicLong clock = new AtomicLong();
+ try {
+ try (WALWriter writer = new WALWriter(dataWal, WALFileVersion.V3)) {
+ writer.write(searchableEntry(1L), singleEntryMeta(19, 1L, 1L, 1000L,
7, 1L));
+ writer.write(searchableEntry(1L), singleEntryMeta(19, 1L, 1L, 1000L,
7, 1L));
+ }
+ try (WALWriter ignored = new WALWriter(successorWal, WALFileVersion.V3))
{
+ // Seal a successor so replay reads the historical file.
+ }
+ try (ProgressWALIterator iterator = new
ProgressWALIterator(dir.toFile(), 1L)) {
+ final ProgressWALIterator.ScanBudget first =
+ new ProgressWALIterator.ScanBudget(1, Long.MAX_VALUE);
+ assertFalse(iterator.hasNext(first));
+ final ProgressWALIterator.ScanBudget expired =
+ new ProgressWALIterator.ScanBudget(100, 50L, clock::get);
+ clock.set(50L);
+ assertFalse(iterator.hasNext(expired));
+ assertTrue(expired.hasYielded());
+ assertFalse(iterator.hasBufferedNext());
+ assertFalse(iterator.hasIncompleteScan());
+
+ final ProgressWALIterator.ScanBudget resumed =
+ new ProgressWALIterator.ScanBudget(100, 50L, clock::get);
+ assertTrue(iterator.hasNext(resumed));
+ final IndexedConsensusRequest request = iterator.next();
+ assertEquals(1L, request.getSearchIndex());
+ assertEquals(2, request.getRequests().size());
+ iterator.retry(request);
+ // Already buffered requests remain available even when no further I/O
is permitted.
+ assertTrue(iterator.hasNext(expired));
+ assertEquals(request, iterator.next());
+ assertFalse(iterator.hasNext(resumed));
+ }
+ } finally {
+ Files.deleteIfExists(dataWal.toPath());
+ Files.deleteIfExists(successorWal.toPath());
+ Files.deleteIfExists(dir);
+ }
+ }
+
@Test
public void testIteratorGroupsByLocalSeqAndCarriesWriterMetadata() throws
Exception {
final Path dir = Files.createTempDirectory("progress-wal-iterator");
