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 dd5d20400097..3d8efca6f47c 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 @@ -2028,7 +2028,7 @@ private MaterializationResult accumulateFromPending( } finally { // Advance once for the consumed prefix, including when a gap or memory block stops the // batch. Per-entry fast-forward can repeatedly inspect the same uncovered WAL file footer. - // Gap recovery creates its own iterator at the current cursor before reading WAL. + // Gap recovery realigns the same iterator at the current cursor before reading WAL. final ProgressWALIterator iterator = subscriptionWALIterator; if (nextExpectedSearchIndex.get() > startSearchIndex && Objects.nonNull(iterator) @@ -2059,10 +2059,16 @@ private MaterializationResult fillGapFromWAL( final int maxTablets, final long maxBatchBytes) { pendingWalGapRetryRequested = false; - resetSubscriptionWALPosition(fromIndex); if (seekGeneration.get() != expectedSeekGeneration || isClosed) { return MaterializationResult.STALE; } + // Preserve the reader position and any memory-blocked request across pending-queue gaps. + // Reopening here scans retained WAL prefixes again on every bounded replay round, and drops + // the request cached for a memory retry. advanceTo skips covered work without losing uncovered + // writer progress, including requests without a local search index. + if (Objects.nonNull(subscriptionWALIterator)) { + 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 @@ -2205,6 +2211,9 @@ private MaterializationResult pumpFromSubscriptionWAL( .PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_ERROR_READING_SUBSCRIPTION_WAL_A3888AC5, this, e); + // next() may have consumed a request whose materialization failed. Reusing this reader + // would lose that request, so recover from the unchanged cursor on the next round. + requestSubscriptionWalReset(nextExpectedSearchIndex.get(), expectedSeekGeneration); break; } } 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 54ea59fbb0a1..bb7b2e59371a 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 @@ -53,6 +53,7 @@ import org.junit.rules.TemporaryFolder; import java.io.File; +import java.io.IOException; import java.lang.reflect.Constructor; import java.lang.reflect.Field; import java.nio.ByteBuffer; @@ -61,6 +62,8 @@ import java.util.Collections; import java.util.List; import java.util.concurrent.BlockingQueue; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; @@ -332,17 +335,192 @@ public void testWalMemoryBlockRetriesWithoutRecreatingIterator() throws Exceptio } } + @Test + public void testPendingGapReplayDoesNotRescanPrefixAcrossBoundedRounds() throws Exception { + assertPendingGapReplay(false, false); + } + + @Test + public void testPendingGapReplayKeepsMemoryBlockedRequestAcrossRounds() throws Exception { + assertPendingGapReplay(true, false); + } + + @Test + public void testPendingGapReplayRetriesRequestAfterConversionFailure() throws Exception { + assertPendingGapReplay(false, true); + } + + private void assertPendingGapReplay( + final boolean memoryBackpressure, final boolean materializationFailure) 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-pending-gap-replay"); + final File walDirectory = temporaryFolder.newFolder("wal-pending-gap-replay"); + ConsensusPrefetchingQueue queue = null; + try { + final long oneTabletBytes = createTablet().ramBytesUsed(); + config.setSubscriptionConsensusBatchMaxWalEntries(2); + config.setSubscriptionConsensusBatchMaxTabletCount(1); + config.setSubscriptionConsensusBatchMaxSizeInBytes(oneTabletBytes * 16L); + config.setSubscriptionConsensusBatchMaxDelayInMs(60_000); + + final int replayedRequests = materializationFailure ? 7 : 6; + final int rounds = memoryBackpressure ? replayedRequests : (replayedRequests + 1) / 2; + final int requestCount = replayedRequests + 2 * rounds; + writeSealedWal(walDirectory, requestCount); + + 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 ConsensusLogToTabletConverter converter = mock(ConsensusLogToTabletConverter.class); + final AtomicBoolean failureInjected = new AtomicBoolean(); + when(converter.convert(any())) + .thenAnswer( + invocation -> { + final InsertNode insertNode = (InsertNode) invocation.getArgument(0); + if (materializationFailure + && insertNode.getSearchIndex() == 2L + && failureInjected.compareAndSet(false, true)) { + throw new IllegalStateException("transient conversion failure"); + } + return Collections.singletonList(createTablet()); + }); + when(converter.getDatabaseName()).thenReturn("db"); + + final AtomicLong closedIteratorBodyReads = new AtomicLong(); + 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) { + @Override + protected ProgressWALIterator createSubscriptionWALIterator( + final long startSearchIndex) { + return new ProgressWALIterator(walNode, startSearchIndex) { + @Override + public void close() throws IOException { + closedIteratorBodyReads.addAndGet(walBodyReadCount(this)); + super.close(); + } + }; + } + }; + final TrackingSubscriptionMemoryManager memoryManager = + new TrackingSubscriptionMemoryManager( + memoryBackpressure + ? oneTabletBytes + Math.max(1L, oneTabletBytes / 2L) + : oneTabletBytes * 16L); + queue.setSubscriptionMemoryManager(memoryManager); + + assertNull(queue.poll("consumer")); + // Admission failures leave a hole before newer pending requests. Keep a suffix queued so + // every bounded round enters fillGapFromWAL, including after ACK releases Tablet memory. + for (long index = replayedRequests + 1L; index <= requestCount; index++) { + assertTrue(pendingEntries(queue).offer(createRequest(index))); + } + + long deliveredLocalSeq = 0L; + for (int round = 0; round < rounds; round++) { + queue.drivePrefetchOnce(); + final long expectedLocalSeq = + materializationFailure + ? round * 2L + 1L + : memoryBackpressure ? round + 1L : (round + 1L) * 2L; + assertEquals(expectedLocalSeq + 1L, queue.getCurrentReadSearchIndex()); + assertMemoryBounded(queue, memoryManager); + if (memoryBackpressure) { + assertTrue(subscriptionWalIterator(queue).hasBufferedNext()); + } + while (deliveredLocalSeq < expectedLocalSeq) { + final SubscriptionEvent event = queue.poll("consumer"); + assertNotNull(event); + deliveredLocalSeq++; + assertEquals( + deliveredLocalSeq, event.getCommitContext().getWriterProgress().getLocalSeq()); + assertEquals( + 1000L + deliveredLocalSeq, + event.getCommitContext().getWriterProgress().getPhysicalTime()); + assertTrue(queue.ack("consumer", event.getCommitContext())); + } + assertEquals(0L, memoryManager.getUsedMemorySizeInBytes()); + } + + assertEquals(replayedRequests, deliveredLocalSeq); + assertEquals(replayedRequests, queue.getWalPathAcceptedEntries()); + assertEquals(0L, queue.getPendingPathAcceptedEntries()); + assertEquals(0L, queue.getWalGapSkippedEntries()); + assertEquals(0L, queue.getSubscriptionUncommittedEventCount()); + assertEquals(0L, queue.getRetainedTabletBytes()); + // Grouping reads one request ahead; a blocked request can require one further lookahead. + // Count actual body reads, including discarded iterators, rather than timing disk caches. + // An exceptional conversion failure needs one reset, rereading its three consumed bodies. + assertEquals( + replayedRequests + (memoryBackpressure ? 2L : 1L) + (materializationFailure ? 3L : 0L), + closedIteratorBodyReads.get() + walBodyReadCount(subscriptionWalIterator(queue))); + } finally { + if (queue != null) { + queue.close(); + } + config.setSubscriptionConsensusBatchMaxWalEntries(originalBatchMaxWalEntries); + config.setSubscriptionConsensusBatchMaxTabletCount(originalBatchMaxTabletCount); + config.setSubscriptionConsensusBatchMaxSizeInBytes(originalBatchMaxSize); + config.setSubscriptionConsensusBatchMaxDelayInMs(originalBatchMaxDelay); + IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir); + } + } + + private static int walBodyReadCount(final ProgressWALIterator iterator) { + try { + // All requests stay in one historical file, so this offset is the number of body reads. + final Field field = + ProgressWALIterator.class.getDeclaredField("consumedEntryCountInCurrentFile"); + field.setAccessible(true); + return field.getInt(iterator); + } catch (final ReflectiveOperationException e) { + throw new AssertionError(e); + } + } + private static void writeSealedWal(final File walDirectory) throws Exception { + writeSealedWal(walDirectory, REQUEST_COUNT); + } + + private static void writeSealedWal(final File walDirectory, final int requestCount) + throws Exception { final File historicalWal = new File( walDirectory, WALFileUtils.getLogFileName(0L, 0L, WALFileStatus.CONTAINS_SEARCH_INDEX)); final File liveWal = new File( walDirectory, - WALFileUtils.getLogFileName(1L, REQUEST_COUNT, WALFileStatus.CONTAINS_SEARCH_INDEX)); + WALFileUtils.getLogFileName(1L, requestCount, WALFileStatus.CONTAINS_SEARCH_INDEX)); try (WALWriter writer = new WALWriter(historicalWal, WALFileVersion.V3)) { - for (long searchIndex = 1L; searchIndex <= REQUEST_COUNT; searchIndex++) { + for (long searchIndex = 1L; searchIndex <= requestCount; searchIndex++) { writeWalEntry(writer, searchIndex); } }