From b3f2d8af7e8d97651cf2e1be48a4da5913e03111 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Mon, 19 Oct 2026 12:33:04 +0800 Subject: [PATCH] [Subscription] Resume realtime delivery after WAL gap recovery --- .../consensus/ConsensusPrefetchingQueue.java | 23 +- ...ConsensusPrefetchingQueueDeliveryTest.java | 16 +- ...usPrefetchingQueueWalBackpressureTest.java | 219 +++++++++++++++++- 3 files changed, 245 insertions(+), 13 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 6ed389ddafbb..ae71c4fea2a6 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 @@ -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); @@ -2353,6 +2354,7 @@ private MaterializationResult tryCatchUpFromWAL( batchState, expectedSeekGeneration, maxWalEntries, + Long.MAX_VALUE, maxTablets, maxBatchBytes, walScanBudget); @@ -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) { @@ -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)) { @@ -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; @@ -2508,7 +2524,6 @@ private void advanceWalReplayCursorIfPresent(final IndexedConsensusRequest reque skippedEntries, totalSkippedEntries); } - nextExpectedSearchIndex.set(actualSearchIndex + 1); } private void ensureSubscriptionWalReadable(final ProgressWALIterator.ScanBudget walScanBudget) { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueDeliveryTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueDeliveryTest.java index e553a509766c..1f1a14535ca5 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueDeliveryTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueDeliveryTest.java @@ -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(); @@ -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()); @@ -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); } }; 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 204a557367e7..adcf18a50d82 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 @@ -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; @@ -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; @@ -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 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(); @@ -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)); @@ -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)) { @@ -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); @@ -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); }