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 9a078cc105e [Subscription] Fix WAL replay progress after realtime gaps
(#18895)
9a078cc105e is described below
commit 9a078cc105e50533004a659ab7d0a00dd8320c39
Author: Caideyipi <[email protected]>
AuthorDate: Fri Oct 16 10:56:44 2026 +0800
[Subscription] Fix WAL replay progress after realtime gaps (#18895)
* [Subscription] Fix WAL replay progress after realtime gaps
* Remove subscription WAL replay documentation
* [Subscription] Initialize WAL position inside iterator factory
---
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 2 +
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 2 +
.../consensus/ConsensusPrefetchingQueue.java | 61 ++++++++++++-
.../broker/consensus/ProgressWALIterator.java | 42 +++++++--
.../consensus/ConsensusPrefetchingQueueTest.java | 99 ++++++++++++++++++++++
...nsensusPrefetchingQueueWalBackpressureTest.java | 51 +++++++++++
.../broker/consensus/ProgressWALIteratorTest.java | 52 +++++++++++-
7 files changed, 299 insertions(+), 10 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 9d400f9d40b..d03146c60b7 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
@@ -20,6 +20,8 @@
package org.apache.iotdb.db.i18n;
public final class DataNodePipeMessages {
+ public static final String
LOG_SUBSCRIPTION_REPLAY_HAS_MADE_NO_CURSOR_PROGRESS_FOR_ARG_MS_ARG_INSPECT_PREFETCH_WORKER_STACKS_AND_WAL_I_O_FAE1F64B
=
+ "Subscription replay has made no cursor progress for {} ms: {}. Inspect
prefetch worker stacks and WAL I/O.";
public static final String
LOG_SUBSCRIPTION_PREFETCH_ROUND_STARTING_FOR_ARG_ACTIVE_ARG_INITIALIZED_ARG_CLOSED_ARG_PENDINGENTRIES_ARG_PREFETCHEDEVENTS_ARG_INFLIGHTEVENTS_ARG_NEXTREADSEARCHINDEX_ARG_F75FF2CD
=
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 4891920e079..0c2282616d8 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
@@ -20,6 +20,8 @@
package org.apache.iotdb.db.i18n;
public final class DataNodePipeMessages {
+ public static final String
LOG_SUBSCRIPTION_REPLAY_HAS_MADE_NO_CURSOR_PROGRESS_FOR_ARG_MS_ARG_INSPECT_PREFETCH_WORKER_STACKS_AND_WAL_I_O_FAE1F64B
=
+ "订阅回放游标已连续 {} ms 未推进：{}。请检查预取工作线程栈和 WAL I/O。";
public static final String
LOG_SUBSCRIPTION_PREFETCH_ROUND_STARTING_FOR_ARG_ACTIVE_ARG_INITIALIZED_ARG_CLOSED_ARG_PENDINGENTRIES_ARG_PREFETCHEDEVENTS_ARG_INFLIGHTEVENTS_ARG_NEXTREADSEARCHINDEX_ARG_F75FF2CD
=
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 9143e40796c..fc9f54f2901 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
@@ -321,6 +321,17 @@ public class ConsensusPrefetchingQueue {
private final SubscriptionQueueTimeTracker timeTracker = new
SubscriptionQueueTimeTracker();
+ private volatile long lastPrefetchDurationMs;
+ private volatile PrefetchRoundResult.Type lastPrefetchResult;
+ private volatile String walReplayState = "NOT_SCANNED";
+
+ // Accessed only by the serialized prefetch worker while holding the queue
read lock.
+ private boolean replayProgressObserved;
+ private long lastObservedReplayCursor;
+ private long lastReplayProgressNanos;
+ private boolean replayStallWarningLogged;
+ private long lastReplayStallWarningNanos;
+
private volatile String lastConsumerId = "";
private volatile long lastPendingAcceptedEntries = 0L;
@@ -1035,8 +1046,7 @@ public class ConsensusPrefetchingQueue {
this.nextExpectedSearchIndex.set(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());
+ resetSubscriptionWALPosition(resolvedStart.getStartSearchIndex());
this.prefetchInitialized = true;
timeTracker.start();
this.observedSeekGeneration = seekGeneration.get();
@@ -1621,6 +1631,8 @@ public class ConsensusPrefetchingQueue {
private static final long WAL_GAP_WAIT_LOG_INTERVAL_MS = 5_000L;
private static final long PREFETCH_STATS_LOG_INTERVAL_MS = 5_000L;
+ private static final long REPLAY_STALL_WARNING_THRESHOLD_NANOS =
TimeUnit.SECONDS.toNanos(60L);
+ private static final long REPLAY_STALL_WARNING_INTERVAL_NANOS =
TimeUnit.SECONDS.toNanos(30L);
private static final long SLOW_PREFETCH_ROUND_THRESHOLD_MS = 10_000L;
@@ -1653,6 +1665,8 @@ public class ConsensusPrefetchingQueue {
return result;
} finally {
final long durationMs = timeTracker.endPrefetch();
+ lastPrefetchDurationMs = durationMs;
+ lastPrefetchResult = Objects.nonNull(result) ? result.getType() : null;
if (debugEnabled) {
LOGGER.debug(
DataNodePipeMessages
@@ -1863,6 +1877,8 @@ public class ConsensusPrefetchingQueue {
}
return PrefetchRoundResult.rescheduleAfter(100L);
} finally {
+ final ProgressWALIterator iterator = subscriptionWALIterator;
+ walReplayState = Objects.nonNull(iterator) ? iterator.toString() :
"UNAVAILABLE";
releaseReadLock();
}
}
@@ -1873,6 +1889,8 @@ public class ConsensusPrefetchingQueue {
return;
}
+ reportReplayStallIfNecessary(System.nanoTime());
+
final long currentPendingAcceptedEntries =
pendingPathAcceptedEntries.get();
final long currentWalAcceptedEntries = walPathAcceptedEntries.get();
LOGGER.info(
@@ -1897,7 +1915,36 @@ public class ConsensusPrefetchingQueue {
lastWalAcceptedEntries = currentWalAcceptedEntries;
}
+ private void reportReplayStallIfNecessary(final long nowNanos) {
+ final long cursor = nextExpectedSearchIndex.get();
+ if (!prefetchInitialized || !isActive || isClosed || closeRequested ||
getLag() <= 0L) {
+ replayProgressObserved = false;
+ replayStallWarningLogged = false;
+ return;
+ }
+ if (!replayProgressObserved || cursor != lastObservedReplayCursor) {
+ replayProgressObserved = true;
+ lastObservedReplayCursor = cursor;
+ lastReplayProgressNanos = nowNanos;
+ replayStallWarningLogged = false;
+ return;
+ }
+ final long stalledNanos = nowNanos - lastReplayProgressNanos;
+ if (stalledNanos >= REPLAY_STALL_WARNING_THRESHOLD_NANOS
+ && (!replayStallWarningLogged
+ || nowNanos - lastReplayStallWarningNanos >=
REPLAY_STALL_WARNING_INTERVAL_NANOS)) {
+ LOGGER.warn(
+ DataNodePipeMessages
+
.LOG_SUBSCRIPTION_REPLAY_HAS_MADE_NO_CURSOR_PROGRESS_FOR_ARG_MS_ARG_INSPECT_PREFETCH_WORKER_STACKS_AND_WAL_I_O_FAE1F64B,
+ TimeUnit.NANOSECONDS.toMillis(stalledNanos),
+ this);
+ lastReplayStallWarningNanos = nowNanos;
+ replayStallWarningLogged = true;
+ }
+ }
+
private void resetRoundStateForSeek(final long newSeekGeneration) {
+ replayProgressObserved = false;
restorePendingSubscriptionWalCursor(newSeekGeneration);
discardRealtimeEntries();
discardBatch(lingerBatch);
@@ -2477,7 +2524,12 @@ public class ConsensusPrefetchingQueue {
protected ProgressWALIterator createSubscriptionWALIterator(final long
startSearchIndex) {
if (consensusReqReader instanceof WALNode) {
- return new ProgressWALIterator((WALNode) consensusReqReader,
startSearchIndex);
+ final ProgressWALIterator iterator =
+ new ProgressWALIterator((WALNode) consensusReqReader,
startSearchIndex);
+ // The constructor sets the local filtering lower bound. Complete
initialization here by
+ // positioning past historical files only when every writer's progress
is already covered.
+ iterator.advanceTo(startSearchIndex,
this::isWriterProgressCoveredForWalFastForward);
+ return iterator;
}
return null;
}
@@ -5105,6 +5157,9 @@ public class ConsensusPrefetchingQueue {
result.put("maxDeliveryIntervalMs",
String.valueOf(getMaxDeliveryIntervalMs()));
result.put("deliveryIdleTimeMs", String.valueOf(getDeliveryIdleTimeMs()));
result.put("prefetchDurationMs", String.valueOf(getPrefetchDurationMs()));
+ result.put("lastPrefetchDurationMs",
String.valueOf(lastPrefetchDurationMs));
+ result.put("lastPrefetchResult", String.valueOf(lastPrefetchResult));
+ result.put("walReplayState", walReplayState);
result.put("maxPrefetchDurationMs",
String.valueOf(getMaxPrefetchDurationMs()));
result.put("prefetchIdleTimeMs", String.valueOf(getPrefetchIdleTimeMs()));
result.put("realtimeAdmissionBlocked",
String.valueOf(realtimeAdmissionBlocked.get()));
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 3f5186893ee..13568fd00be 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
@@ -69,12 +69,14 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
/**
* 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.
+ * cooperative: it starts with the first WAL probe, is checked between
operations, and cannot
+ * interrupt a single file read. Queue housekeeping must not consume the
scan's time slice.
*/
static final class ScanBudget {
private int remainingEntries;
- private final long startNanos;
+ private long startNanos;
+ private boolean started;
private final long maxDurationNanos;
private final LongSupplier nanoTime;
private boolean yielded;
@@ -87,11 +89,15 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
this.remainingEntries = maxEntries;
this.maxDurationNanos = maxDurationNanos;
this.nanoTime = nanoTime;
- this.startNanos = nanoTime.getAsLong();
}
private boolean shouldYield() {
- if (remainingEntries <= 0 || nanoTime.getAsLong() - startNanos >=
maxDurationNanos) {
+ final long nowNanos = nanoTime.getAsLong();
+ if (!started) {
+ startNanos = nowNanos;
+ started = true;
+ }
+ if (remainingEntries <= 0 || nowNanos - startNanos >= maxDurationNanos) {
yielded = true;
}
return yielded;
@@ -141,6 +147,7 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
private IndexedConsensusRequest nextReady;
private boolean scanYielded;
+ private long bodyReadAttempts;
public ProgressWALIterator(final File logDirectory) {
this(logDirectory, Long.MIN_VALUE);
@@ -279,9 +286,13 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
*/
public void advanceTo(
final long targetSearchIndex, final WriterProgressCoverage
writerProgressCoverage) {
- if (targetSearchIndex <= minimumSearchIndex) {
+ if (targetSearchIndex < minimumSearchIndex
+ || (targetSearchIndex == minimumSearchIndex && currentFileIndex >= 0))
{
return;
}
+ // A lower bound filters requests; it does not position the reader. A
fresh/rebuilt iterator
+ // already has this bound but must still be allowed to skip fully covered
historical files.
+ // Once scanning starts, equal-bound probes must not recheck the same
uncovered file footers.
minimumSearchIndex = targetSearchIndex;
if (writerProgressCoverage == null ||
!canDiscardBufferedRequests(writerProgressCoverage)) {
@@ -447,6 +458,26 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
return retryCurrentReader;
}
+ @Override
+ public String toString() {
+ final String status =
+ retryCurrentReader
+ ? "WAITING_FOR_READABLE_WAL"
+ : scanYielded
+ ? "SCAN_YIELDED"
+ : hasBufferedNext() ? "REQUEST_BUFFERED" :
"NO_BUFFERED_REQUEST";
+ return status
+ + "[fileVersion="
+ + currentReaderVersionId
+ + ", entryOffset="
+ + consumedEntryCountInCurrentFile
+ + ", bodyReadAttempts="
+ + bodyReadAttempts
+ + ", minimumSearchIndex="
+ + minimumSearchIndex
+ + "]";
+ }
+
String getSkippedBrokenWalFileDetails(final int firstFileIndex) {
final int firstAvailableIndex =
getSkippedBrokenWalFileCount() - skippedBrokenWalFileDetails.size();
@@ -515,6 +546,7 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
// hasNext() can otherwise read an arbitrary WAL prefix before
returning a request.
scanBudget.remainingEntries--;
}
+ bodyReadAttempts++;
final ByteBuffer buffer = currentReader.next();
consumedEntryCountInCurrentFile =
currentReader.getCurrentEntryIndex() + 1;
final WALEntryType type = WALEntryType.valueOf(buffer.get());
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 a0696342501..61493475115 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
@@ -36,6 +36,7 @@ import
org.apache.iotdb.consensus.iot.subscription.SubscriptionQueueAdmission;
import org.apache.iotdb.consensus.iot.subscription.SubscriptionQueueRegistry;
import
org.apache.iotdb.consensus.iot.subscription.SubscriptionQueueRejectionReason;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.i18n.DataNodePipeMessages;
import org.apache.iotdb.db.queryengine.plan.statement.StatementTestUtils;
import org.apache.iotdb.db.storageengine.dataregion.wal.io.ProgressWALReader;
import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALFileVersion;
@@ -57,6 +58,10 @@ import
org.apache.iotdb.rpc.subscription.payload.poll.SubscriptionPollResponseTy
import org.apache.iotdb.rpc.subscription.payload.poll.WriterId;
import org.apache.iotdb.rpc.subscription.payload.poll.WriterProgress;
+import ch.qos.logback.classic.Level;
+import ch.qos.logback.classic.Logger;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.core.read.ListAppender;
import org.apache.tsfile.enums.ColumnCategory;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.utils.Pair;
@@ -64,6 +69,7 @@ import org.apache.tsfile.write.record.Tablet;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
+import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.IOException;
@@ -113,6 +119,99 @@ public class ConsensusPrefetchingQueueTest {
@Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder();
+ @Test
+ public void testReplayStallWarningIsThrottledAndResetsAfterProgress() throws
Exception {
+ final String originalSystemDir =
IoTDBDescriptor.getInstance().getConfig().getSystemDir();
+ final Logger logger = (Logger)
LoggerFactory.getLogger(ConsensusPrefetchingQueue.class);
+ final Level originalLevel = logger.getLevel();
+ logger.setLevel(Level.WARN);
+ final ListAppender<ILoggingEvent> appender = new ListAppender<>();
+ appender.start();
+ logger.addAppender(appender);
+ ConsensusPrefetchingQueue queue = null;
+ try {
+ final FakeConsensusReqReader reader = new FakeConsensusReqReader();
+ reader.currentSearchIndex = 10L;
+ final IoTConsensusServerImpl server = mock(IoTConsensusServerImpl.class);
+ when(server.getConsensusReqReader()).thenReturn(reader);
+ when(server.getWriterSafeFrontierTracker()).thenReturn(new
WriterSafeFrontierTracker());
+ queue =
+ new ConsensusPrefetchingQueue(
+ "consumerGroup",
+ "topic",
+ TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE,
+ new DataRegionId(1),
+ server,
+ new SubscriptionWalRetentionPolicy(
+ "topic",
+ SubscriptionWalRetentionPolicy.UNBOUNDED,
+ SubscriptionWalRetentionPolicy.UNBOUNDED),
+ mock(ConsensusLogToTabletConverter.class),
+
newCommitManager(temporaryFolder.newFolder("replay-stall-warning")),
+ new RegionProgress(Collections.emptyMap()),
+ 1L,
+ 1L,
+ true);
+ final Field initialized =
+
ConsensusPrefetchingQueue.class.getDeclaredField("prefetchInitialized");
+ initialized.setAccessible(true);
+ initialized.setBoolean(queue, true);
+ final Method report =
+ ConsensusPrefetchingQueue.class.getDeclaredMethod(
+ "reportReplayStallIfNecessary", long.class);
+ report.setAccessible(true);
+ final Supplier<Long> warningCount =
+ () ->
+ appender.list.stream()
+ .filter(
+ event ->
+ event
+ .getMessage()
+ .equals(
+ DataNodePipeMessages
+
.LOG_SUBSCRIPTION_REPLAY_HAS_MADE_NO_CURSOR_PROGRESS_FOR_ARG_MS_ARG_INSPECT_PREFETCH_WORKER_STACKS_AND_WAL_I_O_FAE1F64B))
+ .count();
+ report.invoke(queue, 0L);
+ report.invoke(queue, TimeUnit.SECONDS.toNanos(59L));
+ assertEquals(Long.valueOf(0L), warningCount.get());
+ report.invoke(queue, TimeUnit.SECONDS.toNanos(60L));
+ report.invoke(queue, TimeUnit.SECONDS.toNanos(89L));
+ assertEquals(Long.valueOf(1L), warningCount.get());
+ report.invoke(queue, TimeUnit.SECONDS.toNanos(90L));
+ assertEquals(Long.valueOf(2L), warningCount.get());
+
+ final Field cursorField =
+
ConsensusPrefetchingQueue.class.getDeclaredField("nextExpectedSearchIndex");
+ cursorField.setAccessible(true);
+ ((AtomicLong) cursorField.get(queue)).incrementAndGet();
+ report.invoke(queue, TimeUnit.SECONDS.toNanos(91L));
+ report.invoke(queue, TimeUnit.SECONDS.toNanos(150L));
+ assertEquals(Long.valueOf(2L), warningCount.get());
+ report.invoke(queue, TimeUnit.SECONDS.toNanos(151L));
+ assertEquals(Long.valueOf(3L), warningCount.get());
+
+ reader.currentSearchIndex = 0L;
+ report.invoke(queue, TimeUnit.SECONDS.toNanos(250L));
+ report.invoke(queue, TimeUnit.SECONDS.toNanos(350L));
+ reader.currentSearchIndex = 10L;
+ // A backlog appearing after an idle period starts a new observation
window.
+ report.invoke(queue, TimeUnit.SECONDS.toNanos(450L));
+ report.invoke(queue, TimeUnit.SECONDS.toNanos(509L));
+ assertEquals(Long.valueOf(3L), warningCount.get());
+ queue.setActive(false);
+ report.invoke(queue, TimeUnit.SECONDS.toNanos(600L));
+ assertEquals(Long.valueOf(3L), warningCount.get());
+ } finally {
+ logger.detachAppender(appender);
+ logger.setLevel(originalLevel);
+ appender.stop();
+ if (queue != null) {
+ queue.close();
+ }
+
IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir);
+ }
+ }
+
@Test
public void testLateOldPrefetchDeregistrationKeepsReplacementTask() {
final ConsensusSubscriptionPrefetchExecutor executor =
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 192ed7c3e9c..204a557367e 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
@@ -56,6 +56,7 @@ import java.io.File;
import java.io.IOException;
import java.lang.reflect.Constructor;
import java.lang.reflect.Field;
+import java.lang.reflect.Method;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
@@ -183,6 +184,56 @@ public class ConsensusPrefetchingQueueWalBackpressureTest {
}
queue.drivePrefetchOnce();
assertNull(queue.poll("consumer"));
+ for (int round = 0; round < 10 && queue.getPendingEventCount() > 0;
round++) {
+ queue.drivePrefetchOnce();
+ }
+ assertEquals(0, queue.getPendingEventCount());
+
+ // Rebuilding at the unchanged cursor must position the reader past
covered history. The
+ // fragmented prefix is larger than a whole scan budget and must not
delay WAL takeover.
+ try (WALWriter writer = new WALWriter(liveWal, WALFileVersion.V3)) {
+ writeWalEntry(writer, 4L);
+ writeWalEntry(writer, 5L);
+ }
+ try (WALWriter ignored =
+ new WALWriter(
+ new File(
+ walDirectory,
+ WALFileUtils.getLogFileName(2L, 5L,
WALFileStatus.CONTAINS_SEARCH_INDEX)),
+ WALFileVersion.V3)) {
+ // Seal the new data file so replay can take over at index 4.
+ }
+ when(walNode.getCurrentSearchIndex()).thenReturn(5L);
+ when(walNode.getCurrentWALFileVersion()).thenReturn(2L);
+ // The factory must return an iterator that is ready for bounded replay
at the requested
+ // cursor, without requiring a separate advanceTo call from its caller.
+ try (ProgressWALIterator createdIterator =
queue.createSubscriptionWALIterator(4L)) {
+ final ProgressWALIterator.ScanBudget budget =
+ new ProgressWALIterator.ScanBudget(2, Long.MAX_VALUE);
+ assertTrue(createdIterator.hasNext(budget));
+ assertEquals(4L, createdIterator.next().getSearchIndex());
+ assertFalse(budget.hasYielded());
+ }
+ final Method reset =
+ ConsensusPrefetchingQueue.class.getDeclaredMethod(
+ "resetSubscriptionWALPosition", long.class);
+ reset.setAccessible(true);
+ reset.invoke(queue, 4L);
+ assertTrue(pendingEntries(queue).offer(createRequest(5L)));
+ queue.drivePrefetchOnce();
+ assertTrue(queue.toString(), queue.getCurrentReadSearchIndex() >= 5L);
+ for (int round = 0; round < 10 && queue.getCurrentReadSearchIndex() <
6L; round++) {
+ queue.drivePrefetchOnce();
+ }
+ assertEquals(6L, queue.getCurrentReadSearchIndex());
+ assertEquals(Arrays.asList(2L, 3L, 4L, 5L), conversions);
+ for (long expected = 4L; expected <= 5L; expected++) {
+ final SubscriptionEvent event = queue.poll("consumer");
+ assertNotNull(event);
+ assertEquals(expected,
event.getCommitContext().getWriterProgress().getLocalSeq());
+ assertTrue(queue.ack("consumer", event.getCommitContext()));
+ }
+ assertEquals(0L, queue.getWalGapSkippedEntries());
assertEquals(0L, memoryManager.getUsedMemorySizeInBytes());
verify(walNode, never()).rollWALFile();
} finally {
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 cd136e4cc72..370fca53796 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
@@ -119,6 +119,34 @@ public class ProgressWALIteratorTest {
}
}
+ @Test
+ public void testScanBudgetStartsWhenWalScanningBegins() throws Exception {
+ final Path dir = Files.createTempDirectory("progress-wal-delayed-scan");
+ final File walFile =
+ dir.resolve(WALFileUtils.getLogFileName(0, 0,
WALFileStatus.CONTAINS_SEARCH_INDEX))
+ .toFile();
+ final AtomicLong clock = new AtomicLong();
+ try {
+ try (WALWriter writer = new WALWriter(walFile, WALFileVersion.V3)) {
+ writer.write(searchableEntry(1L), singleEntryMeta(19, 1L, 1L, 1000L,
7, 1L));
+ writer.write(searchableEntry(2L), singleEntryMeta(19, 2L, 1L, 2000L,
7, 2L));
+ }
+ try (ProgressWALIterator iterator = new
ProgressWALIterator(dir.toFile(), 1L)) {
+ final ProgressWALIterator.ScanBudget budget =
+ new ProgressWALIterator.ScanBudget(100, 50L, clock::get);
+ // Queue housekeeping, realtime materialization, or a JVM pause can
consume a whole
+ // time slice before the first WAL probe. That must not starve replay.
+ clock.set(1000L);
+ assertTrue(iterator.hasNext(budget));
+ assertEquals(1L, iterator.next().getSearchIndex());
+ assertFalse(budget.hasYielded());
+ }
+ } finally {
+ Files.deleteIfExists(walFile.toPath());
+ Files.deleteIfExists(dir);
+ }
+ }
+
@Test
public void testElapsedScanBudgetYieldsWithoutFlushingIncompleteRequest()
throws Exception {
final Path dir = Files.createTempDirectory("progress-wal-scan-time-slice");
@@ -142,8 +170,7 @@ public class ProgressWALIteratorTest {
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);
+ new ProgressWALIterator.ScanBudget(100, 50L, () ->
clock.getAndAdd(50L));
assertFalse(iterator.hasNext(expired));
assertTrue(expired.hasYielded());
assertFalse(iterator.hasBufferedNext());
@@ -974,6 +1001,27 @@ public class ProgressWALIteratorTest {
assertTrue(iterator.hasNext());
assertEquals(6L, iterator.next().getSearchIndex());
}
+
+ // A freshly rebuilt replay iterator already has this lower bound, but
has not read or
+ // skipped any files yet. It must still fast-forward covered history
before a bounded scan.
+ try (ProgressWALIterator iterator = new
ProgressWALIterator(dir.toFile(), 6L)) {
+ iterator.advanceTo(6L, (physicalTime, nodeId, localSeq) -> true);
+ final ProgressWALIterator.ScanBudget budget =
+ new ProgressWALIterator.ScanBudget(2, Long.MAX_VALUE);
+ assertTrue(iterator.hasNext(budget));
+ assertEquals(6L, iterator.next().getSearchIndex());
+ assertFalse(budget.hasYielded());
+ }
+
+ // Equal-bound realignment must retain uncovered follower entries, just
like an increasing
+ // bound. They have no local search index and cannot be skipped using
the local cursor.
+ try (ProgressWALIterator iterator = new
ProgressWALIterator(dir.toFile(), 6L)) {
+ iterator.advanceTo(6L, (physicalTime, nodeId, localSeq) -> nodeId ==
7);
+ assertTrue(iterator.hasNext());
+ assertEquals(8, iterator.next().getNodeId());
+ assertTrue(iterator.hasNext());
+ assertEquals(6L, iterator.next().getSearchIndex());
+ }
} finally {
Files.deleteIfExists(localWal.toPath());
Files.deleteIfExists(followerWal.toPath());
