From f3a6326f569ab24ab5b2caa7b5dc4dbfa3b09591 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Thu, 15 Oct 2026 18:15:22 +0800 Subject: [PATCH 1/3] [Subscription] Fix WAL replay progress after realtime gaps --- docs/Subscription-WAL-Replay.md | 115 ++++++++++++++++++ .../iotdb/db/i18n/DataNodePipeMessages.java | 2 + .../iotdb/db/i18n/DataNodePipeMessages.java | 2 + .../consensus/ConsensusPrefetchingQueue.java | 58 ++++++++- .../broker/consensus/ProgressWALIterator.java | 42 ++++++- .../ConsensusPrefetchingQueueTest.java | 99 +++++++++++++++ ...usPrefetchingQueueWalBackpressureTest.java | 42 +++++++ .../consensus/ProgressWALIteratorTest.java | 52 +++++++- 8 files changed, 403 insertions(+), 9 deletions(-) create mode 100644 docs/Subscription-WAL-Replay.md diff --git a/docs/Subscription-WAL-Replay.md b/docs/Subscription-WAL-Replay.md new file mode 100644 index 000000000000..725a7f823154 --- /dev/null +++ b/docs/Subscription-WAL-Replay.md @@ -0,0 +1,115 @@ + + +# Consensus subscription WAL replay + +Realtime requests can be rejected by the shared Consensus request memory pool or by +queue admission. The queue then needs WAL replay to fill the missing local search +indexes before delivering later realtime requests. `WAL replay required` describes +this requirement; it does not promise that WAL is already readable or that replay +will finish within a fixed time. `lastAdmissionRejectionCode` records the most +recent rejection and can remain set after the pressure has passed. + +## Replay progress and latency + +WAL scanning uses a shared per-round physical-entry budget and a cooperative +50 ms time slice. The time slice starts at the first WAL scan operation, after +queue housekeeping. A new iterator may skip historical files only when every +writer's progress in those files is already covered. An equal local search-index +lower bound still permits this realignment; the lower bound alone does not locate +the reader. Uncovered replicated requests remain replayable even when they have +no local search index. Reader position and incomplete request fragments survive +a yielded scan. + +The retry delays for a WAL visibility gap and a memory block are 10 ms and 100 ms, +respectively. These are scheduling delays, not end-to-end recovery deadlines. +WAL visibility, retained history, disk throughput, worker contention, entry size, +and consumer ACK latency all affect recovery. A single I/O operation can exceed +the cooperative time slice. There is no finite recovery-time guarantee while a +required WAL file remains unreadable or memory remains unavailable. Do not skip +that file or advance the cursor to satisfy a latency target. + +For a workload whose normal per-region delivery interval is below 5 seconds, +validate recovery against that same target under sustained writes and temporary +memory pressure. Treat 60 seconds without delivery while lag is positive as an +investigation threshold, not an acceptable recovery time. A cursor that makes no +progress for 60 seconds while lag remains positive also produces a server WARN, +at most once every 30 seconds until cursor progress resumes. The WARN and periodic +queue statistics include the last completed prefetch duration/result and WAL +reader state: scan yield, temporary unreadability, buffered request, or no +buffered request, together with file version, entry offset, and cumulative body +read attempts for that iterator. `SCAN_YIELDED` with advancing file/entry offsets +and increasing body reads identifies historical scanning despite a frozen +delivery cursor. Read retries can also increase the attempt counter, so interpret +it together with the reader state and position. + +`prefetchDurationMs` measures the currently running round. Periodic statistics +are emitted near the start of that round, so a value of zero is expected there. +Use `lastPrefetchDurationMs` for the last completed round and `walReplayState` to +interpret replay activity. The reader state describes its latest WAL probe; a +round blocked before probing WAL can retain that state. Correlate consecutive +samples with `memoryBlockedEntryBytes`, `realtimeAdmissionBlocked`, and reader +position changes. `NO_BUFFERED_REQUEST` alone does not imply an empty WAL backlog. + +## Monitoring each queue + +Evaluate delivery and lag for each `(name, region)` queue on each DataNode. +Consumer-group totals and consumer-level poll intervals can hide a stalled region +when other regions keep delivering. An example Prometheus alert expression is: + +```promql +(subscription_consensus_delivery_idle_time_ms >= 60000) +and (subscription_consensus_lag > 0) +and (subscription_consensus_active == 1) +and (subscription_consensus_initialized == 1) +``` + +These gauges use the same labels, so the expression matches the full queue and +DataNode label set. Consider a short `for` interval appropriate to the scrape +period. Capture prefetch worker stacks and a wall-clock profile when the alert +fires, along with queue statistics and WAL I/O information. The server warning +requires the prefetch worker to run; the delivery-idle gauge can also detect a +worker blocked inside I/O or waiting for a lock. + +## Capacity under sustained writes + +There are separate budgets for retained realtime requests in Consensus memory and +materialized subscription Tablets. Free subscription memory does not prove that +Consensus request admission succeeded, and a historical rejection does not prove +that materialization is currently blocked. + +For a subscription materialization budget `M` and `N` active queues, each queue's +quota is approximately `M / N`. Half of that quota is protected from borrowing by +other queues. One queue's theoretical maximum is approximately +`M - (N - 1) * M / (2 * N)`, but actual availability falls as other queues retain +data. Size writes and consumer ACK cadence for the per-queue quota under concurrent +load, rather than relying on the theoretical borrowing maximum. With a budget of +987,063,313 bytes and ten active queues, the quota is about 94.1 MiB and the +protected share is about 47.1 MiB per queue. + +Reduce write batch sizes or large fields when individual materialized requests +consume much of the quota. Size subscription memory for peak concurrent retained +Tablets and ACK latency, and leave headroom in the shared Consensus pool for the +write path and retained realtime requests. More consumers in one group do not +increase a single region's WAL scan concurrency. Validate sustained WAL replay +throughput above that region's write rate, retention sufficient for the measured +catch-up time, and recovery after rejected realtime requests before accepting a +capacity configuration. 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 9d400f9d40b1..d03146c60b73 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 4891920e079f..0c2282616d84 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 4e71f2e2fe97..4ebadbad2006 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; @@ -1031,8 +1042,7 @@ private boolean initPrefetchUnderInitializationLock(final RegionProgress regionP 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(); @@ -1617,6 +1627,8 @@ public SubscriptionEvent pollTablets( 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; @@ -1649,6 +1661,8 @@ public PrefetchRoundResult drivePrefetchOnce() { return result; } finally { final long durationMs = timeTracker.endPrefetch(); + lastPrefetchDurationMs = durationMs; + lastPrefetchResult = Objects.nonNull(result) ? result.getType() : null; if (debugEnabled) { LOGGER.debug( DataNodePipeMessages @@ -1859,6 +1873,8 @@ private PrefetchRoundResult drivePrefetchOnceInternal() { } return PrefetchRoundResult.rescheduleAfter(100L); } finally { + final ProgressWALIterator iterator = subscriptionWALIterator; + walReplayState = Objects.nonNull(iterator) ? iterator.toString() : "UNAVAILABLE"; releaseReadLock(); } } @@ -1869,6 +1885,8 @@ private void logPeriodicStatsIfNecessary() { return; } + reportReplayStallIfNecessary(System.nanoTime()); + final long currentPendingAcceptedEntries = pendingPathAcceptedEntries.get(); final long currentWalAcceptedEntries = walPathAcceptedEntries.get(); LOGGER.info( @@ -1893,7 +1911,36 @@ private void logPeriodicStatsIfNecessary() { 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); @@ -2469,6 +2516,10 @@ private void resetSubscriptionWALPosition(final long startSearchIndex) { reportedSkippedBrokenWalFileCount = 0; reportedSkippedBrokenWalEntryCount = 0L; subscriptionWALIterator = createSubscriptionWALIterator(startSearchIndex); + if (Objects.nonNull(subscriptionWALIterator)) { + subscriptionWALIterator.advanceTo( + startSearchIndex, this::isWriterProgressCoveredForWalFastForward); + } } protected ProgressWALIterator createSubscriptionWALIterator(final long startSearchIndex) { @@ -5093,6 +5144,9 @@ public Map coreReportMessage() { 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 3f5186893eea..13568fd00be9 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 interface WriterProgressCoverage { /** * 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 @@ static final class ScanBudget { 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 @@ boolean hasYielded() { private IndexedConsensusRequest nextReady; private boolean scanYielded; + private long bodyReadAttempts; public ProgressWALIterator(final File logDirectory) { this(logDirectory, Long.MIN_VALUE); @@ -279,9 +286,13 @@ void retry(final IndexedConsensusRequest request) { */ 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 @@ boolean isWaitingForReadableWal() { 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 @@ private IndexedConsensusRequest advance(final ScanBudget scanBudget) throws IOEx // 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 cc30193647af..d50795cc3518 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 @@ -35,6 +35,7 @@ import org.apache.iotdb.consensus.iot.subscription.SubscriptionQueueAdmission; 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; @@ -56,6 +57,10 @@ 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; @@ -63,6 +68,7 @@ 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; @@ -112,6 +118,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 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 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 192ed7c3e9c8..c37207d9417d 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.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,47 @@ public void testYieldedGapScanKeepsIteratorAndDeliversEachRequestOnce() throws E } 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); + 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 cd136e4cc726..370fca537965 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 void testBoundedScanKeepsSkippedPrefixAndPendingFragmentsAcrossRounds() t } } + @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 void testElapsedScanBudgetYieldsWithoutFlushingIncompleteRequest() throws 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 void testAdvanceToSkipsOnlyFullyCoveredWalFiles() throws Exception { 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()); From 3de706ce0005bde86e09520f49b12b2234ea57a3 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Thu, 15 Oct 2026 18:35:04 +0800 Subject: [PATCH 2/3] Remove subscription WAL replay documentation --- docs/Subscription-WAL-Replay.md | 115 -------------------------------- 1 file changed, 115 deletions(-) delete mode 100644 docs/Subscription-WAL-Replay.md diff --git a/docs/Subscription-WAL-Replay.md b/docs/Subscription-WAL-Replay.md deleted file mode 100644 index 725a7f823154..000000000000 --- a/docs/Subscription-WAL-Replay.md +++ /dev/null @@ -1,115 +0,0 @@ - - -# Consensus subscription WAL replay - -Realtime requests can be rejected by the shared Consensus request memory pool or by -queue admission. The queue then needs WAL replay to fill the missing local search -indexes before delivering later realtime requests. `WAL replay required` describes -this requirement; it does not promise that WAL is already readable or that replay -will finish within a fixed time. `lastAdmissionRejectionCode` records the most -recent rejection and can remain set after the pressure has passed. - -## Replay progress and latency - -WAL scanning uses a shared per-round physical-entry budget and a cooperative -50 ms time slice. The time slice starts at the first WAL scan operation, after -queue housekeeping. A new iterator may skip historical files only when every -writer's progress in those files is already covered. An equal local search-index -lower bound still permits this realignment; the lower bound alone does not locate -the reader. Uncovered replicated requests remain replayable even when they have -no local search index. Reader position and incomplete request fragments survive -a yielded scan. - -The retry delays for a WAL visibility gap and a memory block are 10 ms and 100 ms, -respectively. These are scheduling delays, not end-to-end recovery deadlines. -WAL visibility, retained history, disk throughput, worker contention, entry size, -and consumer ACK latency all affect recovery. A single I/O operation can exceed -the cooperative time slice. There is no finite recovery-time guarantee while a -required WAL file remains unreadable or memory remains unavailable. Do not skip -that file or advance the cursor to satisfy a latency target. - -For a workload whose normal per-region delivery interval is below 5 seconds, -validate recovery against that same target under sustained writes and temporary -memory pressure. Treat 60 seconds without delivery while lag is positive as an -investigation threshold, not an acceptable recovery time. A cursor that makes no -progress for 60 seconds while lag remains positive also produces a server WARN, -at most once every 30 seconds until cursor progress resumes. The WARN and periodic -queue statistics include the last completed prefetch duration/result and WAL -reader state: scan yield, temporary unreadability, buffered request, or no -buffered request, together with file version, entry offset, and cumulative body -read attempts for that iterator. `SCAN_YIELDED` with advancing file/entry offsets -and increasing body reads identifies historical scanning despite a frozen -delivery cursor. Read retries can also increase the attempt counter, so interpret -it together with the reader state and position. - -`prefetchDurationMs` measures the currently running round. Periodic statistics -are emitted near the start of that round, so a value of zero is expected there. -Use `lastPrefetchDurationMs` for the last completed round and `walReplayState` to -interpret replay activity. The reader state describes its latest WAL probe; a -round blocked before probing WAL can retain that state. Correlate consecutive -samples with `memoryBlockedEntryBytes`, `realtimeAdmissionBlocked`, and reader -position changes. `NO_BUFFERED_REQUEST` alone does not imply an empty WAL backlog. - -## Monitoring each queue - -Evaluate delivery and lag for each `(name, region)` queue on each DataNode. -Consumer-group totals and consumer-level poll intervals can hide a stalled region -when other regions keep delivering. An example Prometheus alert expression is: - -```promql -(subscription_consensus_delivery_idle_time_ms >= 60000) -and (subscription_consensus_lag > 0) -and (subscription_consensus_active == 1) -and (subscription_consensus_initialized == 1) -``` - -These gauges use the same labels, so the expression matches the full queue and -DataNode label set. Consider a short `for` interval appropriate to the scrape -period. Capture prefetch worker stacks and a wall-clock profile when the alert -fires, along with queue statistics and WAL I/O information. The server warning -requires the prefetch worker to run; the delivery-idle gauge can also detect a -worker blocked inside I/O or waiting for a lock. - -## Capacity under sustained writes - -There are separate budgets for retained realtime requests in Consensus memory and -materialized subscription Tablets. Free subscription memory does not prove that -Consensus request admission succeeded, and a historical rejection does not prove -that materialization is currently blocked. - -For a subscription materialization budget `M` and `N` active queues, each queue's -quota is approximately `M / N`. Half of that quota is protected from borrowing by -other queues. One queue's theoretical maximum is approximately -`M - (N - 1) * M / (2 * N)`, but actual availability falls as other queues retain -data. Size writes and consumer ACK cadence for the per-queue quota under concurrent -load, rather than relying on the theoretical borrowing maximum. With a budget of -987,063,313 bytes and ten active queues, the quota is about 94.1 MiB and the -protected share is about 47.1 MiB per queue. - -Reduce write batch sizes or large fields when individual materialized requests -consume much of the quota. Size subscription memory for peak concurrent retained -Tablets and ACK latency, and leave headroom in the shared Consensus pool for the -write path and retained realtime requests. More consumers in one group do not -increase a single region's WAL scan concurrency. Validate sustained WAL replay -throughput above that region's write rate, retention sufficient for the measured -catch-up time, and recovery after rejected realtime requests before accepting a -capacity configuration. From 60e3ae390ef403f5000af7d024b9c24ebab0aedc Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Thu, 15 Oct 2026 18:56:34 +0800 Subject: [PATCH 3/3] [Subscription] Initialize WAL position inside iterator factory --- .../broker/consensus/ConsensusPrefetchingQueue.java | 11 ++++++----- .../ConsensusPrefetchingQueueWalBackpressureTest.java | 9 +++++++++ 2 files changed, 15 insertions(+), 5 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 4ebadbad2006..67655f06e315 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 @@ -2516,15 +2516,16 @@ private void resetSubscriptionWALPosition(final long startSearchIndex) { reportedSkippedBrokenWalFileCount = 0; reportedSkippedBrokenWalEntryCount = 0L; subscriptionWALIterator = createSubscriptionWALIterator(startSearchIndex); - if (Objects.nonNull(subscriptionWALIterator)) { - subscriptionWALIterator.advanceTo( - startSearchIndex, this::isWriterProgressCoveredForWalFastForward); - } } 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; } 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 c37207d9417d..204a557367e7 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 @@ -205,6 +205,15 @@ public void testYieldedGapScanKeepsIteratorAndDeliversEachRequestOnce() throws E } 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);