Skip to content

Fix repeated WAL scans during subscription catch-up #18392

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Caideyipi wants to merge 2 commits into
base: master
Choose a base branch
from
Open

Fix repeated WAL scans during subscription catch-up #18392

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
9f0387a
Fix repeated WAL scans during subscription catch-up
Caideyipi Aug 4, 2026
30be225
Merge origin/master into fix/consensus-subscription-wal-catchup
Caideyipi Aug 7, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
87 changes: 23 additions & 64 deletions ...ain/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -195,9 +195,9 @@ public class ConsensusPrefetchingQueue {
private volatile ProgressWALIterator subscriptionWALIterator;

/**
* WAL cursor changes outside the iterator must not close/reset it from RPC threads because the
* prefetch worker may be reading it concurrently. Instead, the latest desired reset is recorded
* and applied by the next prefetch round after observing the expected seek generation.
* Seek requests must not close/reset the WAL iterator from RPC threads because the prefetch
* worker may be reading it concurrently. Instead, seek only records the latest desired reset and
* the queue's next prefetch round applies it after observing the new seek generation.
*/
private volatile long pendingSubscriptionWalResetSearchIndex = Long.MIN_VALUE;

Expand Down Expand Up @@ -1597,41 +1597,12 @@ private boolean isBeforeLocalCursor(final IndexedConsensusRequest request) {
return hasLocalSearchIndex(request) && request.getSearchIndex() < nextExpectedSearchIndex.get();
}

private boolean advanceLocalCursorIfPresent(final IndexedConsensusRequest request) {
private void advanceLocalCursorIfPresent(final IndexedConsensusRequest request) {
if (hasLocalSearchIndex(request)) {
nextExpectedSearchIndex.set(request.getSearchIndex() + 1);
return true;
}
return false;
}

private void advanceLocalCursorFromPendingIfPresent(
final IndexedConsensusRequest request, final long expectedSeekGeneration) {
if (advanceLocalCursorIfPresent(request)) {
// Pending delivery advances independently of the WAL reader. Raise its local lower bound in
// place so stale local requests are filtered without rebuilding and rescanning retained WAL.
final ProgressWALIterator iterator = subscriptionWALIterator;
if (Objects.nonNull(iterator) && seekGeneration.get() == expectedSeekGeneration) {
iterator.advanceTo(
nextExpectedSearchIndex.get(), this::isWriterProgressCoveredForWalFastForward);
}
}
}

private boolean isWriterProgressCoveredForWalFastForward(
final long physicalTime, final int nodeId, final long localSeq) {
final WriterProgress candidate = new WriterProgress(physicalTime, localSeq);
final WriterProgress recoveryProgress =
recoveryWriterProgressByWriter.get(new WriterId(consensusGroupId.toString(), nodeId));
if (Objects.nonNull(recoveryProgress)
&& compareWriterProgress(candidate, recoveryProgress) <= 0) {
return true;
}
final WriterProgress materializedProgress = materializedProgressByWriter.get(nodeId);
return Objects.nonNull(materializedProgress)
&& compareWriterProgress(candidate, materializedProgress) <= 0;
}

private MaterializationResult appendRealtimeRequest(
final IndexedConsensusRequest request,
final DeliveryBatchState batchState,
Expand Down Expand Up @@ -1707,12 +1678,12 @@ private MaterializationResult accumulateFromPending(

if (shouldSkipForRecoveryProgress(request)) {
skippedCount++;
advanceLocalCursorFromPendingIfPresent(request, expectedSeekGeneration);
advanceLocalCursorIfPresent(request);
continue;
}
if (shouldSkipForMaterializedProgress(request)) {
skippedCount++;
advanceLocalCursorFromPendingIfPresent(request, expectedSeekGeneration);
advanceLocalCursorIfPresent(request);
continue;
}

Expand All @@ -1724,7 +1695,7 @@ private MaterializationResult accumulateFromPending(
}
markMaterializedProgress(request);
processedCount++;
advanceLocalCursorFromPendingIfPresent(request, expectedSeekGeneration);
advanceLocalCursorIfPresent(request);
if (prefetchingQueue.size() >= MAX_PREFETCHING_QUEUE_SIZE) {
break;
}
Expand Down Expand Up @@ -1816,9 +1787,11 @@ private MaterializationResult tryCatchUpFromWAL(final long expectedSeekGeneratio
// Use the persistent linger batch so an unexpected runtime failure cannot orphan already
// reserved Tablets or advance replay progress past data that has become unreachable.
final DeliveryBatchState batchState = lingerBatch;
// Keep the iterator and its buffered next request across rounds. Reopening it here discards the
// request prepared by hasNext() and repeatedly re-reads, skips, and decompresses the same WAL
// segment. Pending-path cursor advances and seek operations request explicit realignment.
// Keep using the current iterator so its WAL reader cursor and buffered look-ahead request are
// preserved across bounded prefetch rounds. Rebuilding it here would replay all retained WAL
// entries before nextExpectedSearchIndex for every batch and make historical catch-up
// progressively slower. Explicit seek, WAL-gap recovery, and memory rollback still reset the
// iterator at their required positions.
final MaterializationResult materializationResult =
pumpFromSubscriptionWAL(
batchState, expectedSeekGeneration, maxWalEntries, maxTablets, maxBatchBytes);
Expand Down Expand Up @@ -1851,6 +1824,7 @@ private MaterializationResult pumpFromSubscriptionWAL(
return MaterializationResult.SUCCESS;
}

subscriptionWALIterator.refresh();
ensureSubscriptionWalReadable();

int entriesRead = 0;
Expand Down Expand Up @@ -1906,14 +1880,9 @@ private MaterializationResult pumpFromSubscriptionWAL(
}

private void ensureSubscriptionWalReadable() {
if (Objects.isNull(subscriptionWALIterator) || subscriptionWALIterator.hasNext()) {
return;
}

// Listing and sorting all retained WAL files is only necessary after the iterator is
// exhausted. While it still has a readable request, refreshing cannot affect the next result.
subscriptionWALIterator.refresh();
if (subscriptionWALIterator.hasNext() || !(consensusReqReader instanceof WALNode)) {
if (Objects.isNull(subscriptionWALIterator)
|| subscriptionWALIterator.hasNext()
|| !(consensusReqReader instanceof WALNode)) {
return;
}

Expand All @@ -1930,6 +1899,9 @@ private void ensureSubscriptionWalReadable() {
currentWalIndex);
((WALNode) consensusReqReader).rollWALFile();
resetSubscriptionWALPosition(nextExpectedSearchIndex.get());
if (Objects.nonNull(subscriptionWALIterator)) {
subscriptionWALIterator.refresh();
}
}

private void resetSubscriptionWALPosition(final long startSearchIndex) {
Expand All @@ -1947,11 +1919,6 @@ protected ProgressWALIterator createSubscriptionWALIterator(final long startSear
protected void onWalGapRetryScheduled() {}

private boolean hasReadableWalEntries() {
if (pendingSubscriptionWalResetSearchIndex != Long.MIN_VALUE) {
// Do not advance the stale iterator only to discard its buffered request when the next round
// applies the pending realignment. Returning true keeps the worker scheduled for that round.
return true;
}
return Objects.nonNull(subscriptionWALIterator) && subscriptionWALIterator.hasNext();
}

Expand Down Expand Up @@ -2078,10 +2045,7 @@ private boolean createAndEnqueueEvent(
SubscriptionPollResponseType.TABLETS.getType(),
payload,
commitContext,
SubscriptionAgent.broker()
.getColumnFilterMatcher(
topicName, SubscriptionAgent.consumer().isTableModel(consumerGroupId))
.isTimeSelected(),
SubscriptionAgent.broker().getColumnFilterMatcher(topicName).isTimeSelected(),
getTimeSelectedByTable(converter.getDatabaseName(), tablets));

// Install the ownership record before exposing the event to concurrent poll/ack threads.
Expand All @@ -2107,9 +2071,7 @@ private Map<String, Map<String, Boolean>> getTimeSelectedByTable(
return Collections.emptyMap();
}
final ColumnFilterMatcher matcher =
SubscriptionAgent.broker()
.getColumnFilterMatcher(
topicName, SubscriptionAgent.consumer().isTableModel(consumerGroupId));
SubscriptionAgent.broker().getColumnFilterMatcher(topicName);
final Map<String, Boolean> tableMap = new HashMap<>();
for (final Tablet tablet : tablets) {
if (Objects.nonNull(tablet) && Objects.nonNull(tablet.getTableName())) {
Expand Down Expand Up @@ -2251,10 +2213,7 @@ private void cleanUpEvent(final SubscriptionEvent event, final boolean force) {

private boolean ackMissingInFlightEvent(
final SubscriptionCommitContext commitContext, final boolean silent) {
// Late or duplicate ACKs touch the same concurrent lifecycle indexes and commit manager as the
// regular in-flight ACK path. A read lock is sufficient to fence seek/close transitions while
// allowing ACKs to proceed concurrently with a long-running WAL prefetch round.
acquireReadLock();
acquireWriteLock();
try {
if (!canAcceptCommitContext(commitContext, "ack", silent)) {
return false;
Expand Down Expand Up @@ -2301,7 +2260,7 @@ private boolean ackMissingInFlightEvent(
}
return true;
} finally {
releaseReadLock();
releaseWriteLock();
}
}

Expand Down
91 changes: 91 additions & 0 deletions .../iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueWalBackpressureTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,6 +78,90 @@ public class ConsensusPrefetchingQueueWalBackpressureTest {

@Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder();

@Test
public void testHistoricalCatchUpReusesWalIteratorAcrossBatches() 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 long originalBatchMaxSize = config.getSubscriptionConsensusBatchMaxSizeInBytes();
final int originalBatchMaxDelay = config.getSubscriptionConsensusBatchMaxDelayInMs();
final File systemDir = temporaryFolder.newFolder("system-wal-iterator-reuse");
final File walDirectory = temporaryFolder.newFolder("wal-iterator-reuse");
ConsensusPrefetchingQueue queue = null;
try {
config.setSubscriptionConsensusBatchMaxWalEntries(1);
config.setSubscriptionConsensusBatchMaxTabletCount(128);
config.setSubscriptionConsensusBatchMaxSizeInBytes(Long.MAX_VALUE);
config.setSubscriptionConsensusBatchMaxDelayInMs(0);

writeSealedWal(walDirectory);

final WALNode walNode = mock(WALNode.class);
when(walNode.getLogDirectory()).thenReturn(walDirectory);
when(walNode.getCurrentSearchIndex()).thenReturn((long) REQUEST_COUNT);
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 ConsensusLogToTabletConverter converter = mock(ConsensusLogToTabletConverter.class);
when(converter.convert(any())).thenReturn(Collections.singletonList(createTablet()));
when(converter.getDatabaseName()).thenReturn("db");

final DataRegionId regionId = new DataRegionId(1);
queue =
new ConsensusPrefetchingQueue(
"consumerGroup",
"topic",
TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE,
regionId,
serverImpl,
new SubscriptionWalRetentionPolicy(
"topic",
SubscriptionWalRetentionPolicy.UNBOUNDED,
SubscriptionWalRetentionPolicy.UNBOUNDED),
converter,
newCommitManager(systemDir),
new RegionProgress(Collections.emptyMap()),
1L,
1L,
true);

assertNull(queue.poll("consumer"));
final ProgressWALIterator initialIterator = subscriptionWalIterator(queue);
assertNotNull(initialIterator);

for (long expectedLocalSequence = 1L;
expectedLocalSequence <= REQUEST_COUNT;
expectedLocalSequence++) {
queue.drivePrefetchOnce();
assertEquals(initialIterator, subscriptionWalIterator(queue));
assertEquals(expectedLocalSequence + 1L, queue.getCurrentReadSearchIndex());

final SubscriptionEvent event = queue.poll("consumer");
assertNotNull(event);
assertEquals(
expectedLocalSequence, event.getCommitContext().getWriterProgress().getLocalSeq());
assertTrue(queue.ack("consumer", event.getCommitContext()));
}

assertEquals(REQUEST_COUNT, queue.getWalPathAcceptedEntries());
assertEquals(0, queue.getPrefetchedEventCount());
} finally {
if (queue != null) {
queue.close();
}
config.setSubscriptionConsensusBatchMaxWalEntries(originalBatchMaxWalEntries);
config.setSubscriptionConsensusBatchMaxTabletCount(originalBatchMaxTabletCount);
config.setSubscriptionConsensusBatchMaxSizeInBytes(originalBatchMaxSize);
config.setSubscriptionConsensusBatchMaxDelayInMs(originalBatchMaxDelay);
IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir);
}
}

@Test
public void testAckRecoversDrainedSuffixFromWalWithoutReoffer() throws Exception {
final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir();
Expand Down Expand Up @@ -288,6 +372,13 @@ private static BlockingQueue<IndexedConsensusRequest> pendingEntries(
return (BlockingQueue<IndexedConsensusRequest>) field.get(queue);
}

private static ProgressWALIterator subscriptionWalIterator(final ConsensusPrefetchingQueue queue)
throws Exception {
final Field field = ConsensusPrefetchingQueue.class.getDeclaredField("subscriptionWALIterator");
field.setAccessible(true);
return (ProgressWALIterator) field.get(queue);
}

private static ConsensusSubscriptionCommitManager newCommitManager(final File systemDir)
throws Exception {
IoTDBDescriptor.getInstance().getConfig().setSystemDir(systemDir.getAbsolutePath());
Expand Down
Loading