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 f535328cdca6..5a88845242ff 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 @@ -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; @@ -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, @@ -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; } @@ -1724,7 +1695,7 @@ private MaterializationResult accumulateFromPending( } markMaterializedProgress(request); processedCount++; - advanceLocalCursorFromPendingIfPresent(request, expectedSeekGeneration); + advanceLocalCursorIfPresent(request); if (prefetchingQueue.size() >= MAX_PREFETCHING_QUEUE_SIZE) { break; } @@ -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); @@ -1851,6 +1824,7 @@ private MaterializationResult pumpFromSubscriptionWAL( return MaterializationResult.SUCCESS; } + subscriptionWALIterator.refresh(); ensureSubscriptionWalReadable(); int entriesRead = 0; @@ -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; } @@ -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) { @@ -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(); } @@ -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. @@ -2107,9 +2071,7 @@ private Map> getTimeSelectedByTable( return Collections.emptyMap(); } final ColumnFilterMatcher matcher = - SubscriptionAgent.broker() - .getColumnFilterMatcher( - topicName, SubscriptionAgent.consumer().isTableModel(consumerGroupId)); + SubscriptionAgent.broker().getColumnFilterMatcher(topicName); final Map tableMap = new HashMap<>(); for (final Tablet tablet : tablets) { if (Objects.nonNull(tablet) && Objects.nonNull(tablet.getTableName())) { @@ -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; @@ -2301,7 +2260,7 @@ private boolean ackMissingInFlightEvent( } return true; } finally { - releaseReadLock(); + releaseWriteLock(); } } 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 b3783beab43d..8aa40a042ed7 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 @@ -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(); @@ -288,6 +372,13 @@ private static BlockingQueue pendingEntries( return (BlockingQueue) 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());