Skip to content

[Subscription] Resume realtime delivery after WAL gap recovery #18912

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

Merged
jt2594838 merged 1 commit into from
Oct 19, 2026
Merged

[Subscription] Resume realtime delivery after WAL gap recovery #18912

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
23 changes: 19 additions & 4 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 @@ -2286,14 +2286,15 @@ private MaterializationResult fillGapFromWAL(
subscriptionWALIterator.advanceTo(fromIndex, this::isWriterProgressCoveredForWalFastForward);
}
// A pending-queue jump can span millions of WAL entries when real-time delivery has overflowed.
// Keep gap recovery within the normal per-round budget so sparse topic filtering cannot hold
// the
// queue read lock indefinitely and block unsubscribe from acquiring the close write lock.
// Stop replay at the pending boundary. Admitted requests should use direct delivery instead of
// being replayed and discarded as duplicates. Gap recovery still shares the normal scan budget,
// so sparse filtering cannot hold the read lock indefinitely and block unsubscribe.
final MaterializationResult pumpResult =
pumpFromSubscriptionWAL(
batchState,
expectedSeekGeneration,
maxWalEntries,
toIndex,
maxTablets,
maxBatchBytes,
walScanBudget);
Expand Down Expand Up @@ -2353,6 +2354,7 @@ private MaterializationResult tryCatchUpFromWAL(
batchState,
expectedSeekGeneration,
maxWalEntries,
Long.MAX_VALUE,
maxTablets,
maxBatchBytes,
walScanBudget);
Expand Down Expand Up @@ -2380,6 +2382,7 @@ private MaterializationResult pumpFromSubscriptionWAL(
final DeliveryBatchState batchState,
final long expectedSeekGeneration,
final int maxWalEntries,
final long endSearchIndexExclusive,
final int maxTablets,
final long maxBatchBytes,
final ProgressWALIterator.ScanBudget walScanBudget) {
Expand All @@ -2406,6 +2409,14 @@ private MaterializationResult pumpFromSubscriptionWAL(
reportUnreadableWalReplayIfNecessary();
try {
final IndexedConsensusRequest walEntry = subscriptionWALIterator.next();
if (hasLocalSearchIndex(walEntry) && walEntry.getSearchIndex() >= endSearchIndexExclusive) {
// A sparse WAL can jump past the pending boundary. Preserve its next request while
// handing the admitted entry back to the pending path, and report only the missing gap.
subscriptionWALIterator.retry(walEntry);
reportWalGapBefore(endSearchIndexExclusive);
nextExpectedSearchIndex.set(endSearchIndexExclusive);
break;
}
entriesRead++;

if (isBeforeLocalCursor(walEntry)) {
Expand Down Expand Up @@ -2495,6 +2506,11 @@ private void advanceWalReplayCursorIfPresent(final IndexedConsensusRequest reque
}

final long actualSearchIndex = request.getSearchIndex();
reportWalGapBefore(actualSearchIndex);
nextExpectedSearchIndex.set(actualSearchIndex + 1);
}

private void reportWalGapBefore(final long actualSearchIndex) {
final long expectedSearchIndex = nextExpectedSearchIndex.get();
if (actualSearchIndex > expectedSearchIndex) {
final long skippedEntries = actualSearchIndex - expectedSearchIndex;
Expand All @@ -2508,7 +2524,6 @@ private void advanceWalReplayCursorIfPresent(final IndexedConsensusRequest reque
skippedEntries,
totalSkippedEntries);
}
nextExpectedSearchIndex.set(actualSearchIndex + 1);
}

private void ensureSubscriptionWalReadable(final ProgressWALIterator.ScanBudget walScanBudget) {
Expand Down
16 changes: 15 additions & 1 deletion .../apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueDeliveryTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -257,7 +257,9 @@ public void testMixedPendingAndWalBatchRetriesRedeliveryUntilAllMembersAreReadab
config.setSubscriptionConsensusBatchMaxTabletCount(2);
writeWal(1, true);
currentSearchIndex.set(2);
final ConsensusPrefetchingQueue queue = newQueue();
// Keep the mixed batch deterministic when first-use class loading exhausts a scan slice.
// Bounded scan yielding is covered by ConsensusPrefetchingQueueWalBackpressureTest.
final ConsensusPrefetchingQueue queue = newQueue(true);
assertNull(queue.poll(CONSUMER));
assertTrue(pendingEntries(queue).offer(request(2, 2, WRITER)));
queue.drivePrefetchOnce();
Expand Down Expand Up @@ -591,6 +593,10 @@ public void testLateDeliveryCallbackCannotDetachAnotherAttemptAndCloseWaitsForSe
}

private ConsensusPrefetchingQueue newQueue() {
return newQueue(false);
}

private ConsensusPrefetchingQueue newQueue(final boolean unboundedWalFixture) {
final IoTConsensusServerImpl server = mock(IoTConsensusServerImpl.class);
when(server.getConsensusReqReader()).thenReturn(walNode);
when(server.getWriterSafeFrontierTracker()).thenReturn(new WriterSafeFrontierTracker());
Expand All @@ -614,6 +620,14 @@ private ConsensusPrefetchingQueue newQueue() {
@Override
protected ProgressWALIterator createSubscriptionWALIterator(final long startSearchIndex) {
walIteratorCreations.incrementAndGet();
if (unboundedWalFixture) {
return new ProgressWALIterator(walNode, startSearchIndex) {
@Override
boolean hasNext(final ScanBudget scanBudget) {
return super.hasNext(null);
}
};
}
return super.createSubscriptionWALIterator(startSearchIndex);
}
};
Expand Down
219 changes: 211 additions & 8 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 @@ -22,11 +22,16 @@
import org.apache.iotdb.commons.conf.CommonConfig;
import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.apache.iotdb.commons.consensus.DataRegionId;
import org.apache.iotdb.commons.memory.AtomicLongMemoryBlock;
import org.apache.iotdb.commons.memory.IMemoryBlock;
import org.apache.iotdb.consensus.common.request.IndexedConsensusRequest;
import org.apache.iotdb.consensus.iot.IoTConsensusServerImpl;
import org.apache.iotdb.consensus.iot.SubscriptionWalRetentionPolicy;
import org.apache.iotdb.consensus.iot.WriterSafeFrontierTracker;
import org.apache.iotdb.consensus.iot.logdispatcher.IoTConsensusMemoryManager;
import org.apache.iotdb.consensus.iot.subscription.SubscriptionQueueRejectionReason;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryWeightUtil;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowNode;
import org.apache.iotdb.db.queryengine.plan.statement.StatementTestUtils;
Expand Down Expand Up @@ -62,6 +67,7 @@
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
Expand Down Expand Up @@ -247,6 +253,181 @@ public void testYieldedGapScanKeepsIteratorAndDeliversEachRequestOnce() throws E
}
}

@Test
public void testWalGapRecoveryResumesRealtimeAfterConsensusMemoryRejection() throws Exception {
assertWalGapRecoveryResumesRealtime(false, false, false);
}

@Test
public void testWalGapRecoveryResumesRealtimeAfterAck() throws Exception {
assertWalGapRecoveryResumesRealtime(true, false, false);
}

@Test
public void testWalGapRecoveryResumesPendingWhenWalStartsAfterGap() throws Exception {
assertWalGapRecoveryResumesRealtime(false, true, false);
}

@Test
public void testWalGapRecoveryPreservesFollowerBeforePendingBoundary() throws Exception {
assertWalGapRecoveryResumesRealtime(false, false, true);
}

private void assertWalGapRecoveryResumesRealtime(
final boolean memoryBackpressure,
final boolean missingWalGap,
final boolean followerBeforePending)
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 long originalWatermarkInterval = config.getSubscriptionConsensusWatermarkIntervalMs();
final IoTConsensusMemoryManager requestMemoryManager = IoTConsensusMemoryManager.getInstance();
final IMemoryBlock originalRequestMemoryBlock = requestMemoryManager.getMemoryBlock();
final double originalQueueRatio = requestMemoryManager.getMaxMemoryRatioForQueue();
final AtomicLongMemoryBlock requestMemoryBlock =
new AtomicLongMemoryBlock("subscription-gap-recovery", null, 1_000_000L);
final File systemDir = temporaryFolder.newFolder("system-gap-recovery");
final File walDirectory = temporaryFolder.newFolder("wal-gap-recovery");
final int requestCount = 8;
ConsensusPrefetchingQueue queue = null;
try {
config.setSubscriptionConsensusBatchMaxWalEntries(128);
config.setSubscriptionConsensusBatchMaxTabletCount(1);
config.setSubscriptionConsensusBatchMaxDelayInMs(0);
config.setSubscriptionConsensusWatermarkIntervalMs(0L);
writeSealedWal(walDirectory, requestCount, missingWalGap ? 2L : 1L, followerBeforePending);

final WALNode walNode = mock(WALNode.class);
when(walNode.getLogDirectory()).thenReturn(walDirectory);
when(walNode.getCurrentSearchIndex()).thenReturn((long) requestCount);
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");
final DataRegionId regionId = new DataRegionId(1);
final ConsensusSubscriptionCommitManager commitManager = newCommitManager(systemDir);
queue =
new ConsensusPrefetchingQueue(
"consumerGroup",
"topic",
followerBeforePending
? TopicConstant.ORDER_MODE_MULTI_WRITER_VALUE
: TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE,
regionId,
serverImpl,
new SubscriptionWalRetentionPolicy(
"topic",
SubscriptionWalRetentionPolicy.UNBOUNDED,
SubscriptionWalRetentionPolicy.UNBOUNDED),
converter,
commitManager,
new RegionProgress(Collections.emptyMap()),
1L,
1L,
true);
if (followerBeforePending) {
queue.setActiveWriterNodeIds(Set.of(WRITER_NODE_ID, WRITER_NODE_ID + 1));
}
final long oneTabletBytes = PipeMemoryWeightUtil.calculateTabletSizeInBytes(createTablet());
final TrackingSubscriptionMemoryManager memoryManager =
new TrackingSubscriptionMemoryManager(
memoryBackpressure ? oneTabletBytes * 4L : oneTabletBytes * requestCount * 4L);
queue.setSubscriptionMemoryManager(memoryManager);
assertNull(queue.poll("consumer"));

// A rejected request creates a WAL-only hole. Once consensus memory is released, subsequent
// requests are admitted again and must be delivered from pending after that hole is filled.
requestMemoryManager.init(requestMemoryBlock, 1.0);
requestMemoryBlock.forceAllocateWithoutLimitation(1_000_000L);
assertFalse(pendingEntries(queue).offer(createRequest(1L)));
assertEquals(
SubscriptionQueueRejectionReason.CONSENSUS_REQUEST_MEMORY_LIMIT.getCode(),
queue.getLastAdmissionRejectionCode());
requestMemoryBlock.release(1_000_000L);
for (long index = 2L; index <= requestCount; index++) {
assertTrue(pendingEntries(queue).offer(createRequest(index)));
}

long deliveredLocalSeq = missingWalGap ? 1L : 0L;
for (int round = 0; round < 32 && deliveredLocalSeq < requestCount; round++) {
queue.drivePrefetchOnce();
assertMemoryBounded(queue, memoryManager);
SubscriptionEvent event;
while ((event = queue.poll("consumer")) != null) {
assertEquals(WRITER_NODE_ID, event.getCommitContext().getWriterId().getNodeId());
deliveredLocalSeq++;
assertEquals(
deliveredLocalSeq, event.getCommitContext().getWriterProgress().getLocalSeq());
assertTrue(queue.ack("consumer", event.getCommitContext()));
}
assertEquals(
followerBeforePending ? oneTabletBytes : 0L, memoryManager.getUsedMemorySizeInBytes());
}
assertEquals(requestCount, deliveredLocalSeq);
if (followerBeforePending) {
// The replicated writer's later timestamp lets local requests drain in order. Removing
// the local writer then releases that final follower entry without reading it again.
queue.setActiveWriterNodeIds(Collections.singleton(WRITER_NODE_ID + 1));
queue.drivePrefetchOnce();
final SubscriptionEvent follower = queue.poll("consumer");
assertNotNull(follower);
assertEquals(WRITER_NODE_ID + 1, follower.getCommitContext().getWriterId().getNodeId());
assertEquals(
new WriterProgress(2000L, 1L), follower.getCommitContext().getWriterProgress());
assertTrue(queue.ack("consumer", follower.getCommitContext()));
assertNull(queue.poll("consumer"));
assertEquals(0L, memoryManager.getUsedMemorySizeInBytes());
}
assertEquals(
missingWalGap ? 0L : (followerBeforePending ? 2L : 1L),
queue.getWalPathAcceptedEntries());
assertEquals(requestCount - 1L, queue.getPendingPathAcceptedEntries());
assertEquals(1L, queue.getRealtimeAdmissionRejectionCount());
assertEquals(missingWalGap ? 1L : 0L, queue.getWalGapSkippedEntries());
assertEquals(0L, queue.getRetainedRequestBytes());
assertEquals(0L, requestMemoryBlock.getUsedMemoryInBytes());
assertTrue(pendingEntries(queue).isEmpty());
assertEquals(
new WriterProgress(1000L + requestCount, requestCount),
commitManager
.getCommittedRegionProgress("consumerGroup", "topic", regionId)
.getWriterPositions()
.get(new WriterId(regionId.toString(), WRITER_NODE_ID)));
if (!memoryBackpressure) {
assertEquals(
missingWalGap
? Arrays.asList(2L, 3L, 4L, 5L, 6L, 7L, 8L)
: (followerBeforePending
? Arrays.asList(1L, -1L, 2L, 3L, 4L, 5L, 6L, 7L, 8L)
: Arrays.asList(1L, 2L, 3L, 4L, 5L, 6L, 7L, 8L)),
conversions);
}
} finally {
if (queue != null) {
queue.close();
}
requestMemoryManager.init(originalRequestMemoryBlock, originalQueueRatio);
config.setSubscriptionConsensusBatchMaxWalEntries(originalBatchMaxWalEntries);
config.setSubscriptionConsensusBatchMaxTabletCount(originalBatchMaxTabletCount);
config.setSubscriptionConsensusBatchMaxDelayInMs(originalBatchMaxDelay);
config.setSubscriptionConsensusWatermarkIntervalMs(originalWatermarkInterval);
IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir);
}
}

@Test
public void testAckResumesPendingSuffixWithoutWalReplay() throws Exception {
final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir();
Expand Down Expand Up @@ -690,6 +871,21 @@ private static void writeSealedWal(final File walDirectory) throws Exception {

private static void writeSealedWal(final File walDirectory, final int requestCount)
throws Exception {
writeSealedWal(walDirectory, requestCount, 1L);
}

private static void writeSealedWal(
final File walDirectory, final int requestCount, final long firstSearchIndex)
throws Exception {
writeSealedWal(walDirectory, requestCount, firstSearchIndex, false);
}

private static void writeSealedWal(
final File walDirectory,
final int requestCount,
final long firstSearchIndex,
final boolean followerBeforePending)
throws Exception {
final File historicalWal =
new File(
walDirectory, WALFileUtils.getLogFileName(0L, 0L, WALFileStatus.CONTAINS_SEARCH_INDEX));
Expand All @@ -699,8 +895,11 @@ private static void writeSealedWal(final File walDirectory, final int requestCou
WALFileUtils.getLogFileName(1L, requestCount, WALFileStatus.CONTAINS_SEARCH_INDEX));

try (WALWriter writer = new WALWriter(historicalWal, WALFileVersion.V3)) {
for (long searchIndex = 1L; searchIndex <= requestCount; searchIndex++) {
for (long searchIndex = firstSearchIndex; searchIndex <= requestCount; searchIndex++) {
writeWalEntry(writer, searchIndex);
if (searchIndex == 1L && followerBeforePending) {
writeWalEntry(writer, -1L, 2000L, WRITER_NODE_ID + 1, 1L);
}
}
}
try (WALWriter ignored = new WALWriter(liveWal, WALFileVersion.V3)) {
Expand All @@ -710,6 +909,16 @@ private static void writeSealedWal(final File walDirectory, final int requestCou

private static void writeWalEntry(final WALWriter writer, final long searchIndex)
throws Exception {
writeWalEntry(writer, searchIndex, 1000L + searchIndex, WRITER_NODE_ID, searchIndex);
}

private static void writeWalEntry(
final WALWriter writer,
final long searchIndex,
final long physicalTime,
final int writerNodeId,
final long localSeq)
throws Exception {
final RelationalInsertRowNode insertNode =
StatementTestUtils.genInsertRowNode(Math.toIntExact(searchIndex));
insertNode.setSearchIndex(searchIndex);
Expand All @@ -719,13 +928,7 @@ private static void writeWalEntry(final WALWriter writer, final long searchIndex
walEntry.serialize(new WALByteBufferForTest(buffer));

final WALMetaData metaData = new WALMetaData();
metaData.add(
walEntry.serializedSize(),
searchIndex,
1L,
1000L + searchIndex,
WRITER_NODE_ID,
searchIndex);
metaData.add(walEntry.serializedSize(), searchIndex, 1L, physicalTime, writerNodeId, localSeq);
writer.write(buffer, metaData);
}

Expand Down
Loading