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 a678646a7355..1cfffe2628f0 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 @@ -205,7 +205,7 @@ public class ConsensusPrefetchingQueue { */ private volatile long memoryBlockedEntryBytes = -1L; - /** Rejects the real-time fast path while this queue cannot safely materialize more data. */ + /** Rejects the real-time fast path during lifecycle transitions or a blocked writer backlog. */ private final AtomicBoolean realtimeAdmissionBlocked = new AtomicBoolean(false); private static final int MAX_PREFETCHING_QUEUE_SIZE = @@ -534,8 +534,10 @@ private void release(final IndexedConsensusRequest request) { requestMemoryManager.free(request); } - private void release(final List requests) { - requests.forEach(this::release); + private synchronized void consume(final IndexedConsensusRequest request) { + if (super.removeFirstOccurrence(request)) { + release(request); + } } private long getRetainedRequestBytes() { @@ -712,11 +714,29 @@ private void requestPrefetch() { } private void requestPrefetchForRealtimeEntry() { - if (prefetchingQueue.size() < MAX_PREFETCHING_QUEUE_SIZE) { + if (!hasReachedPrefetchLimit()) { requestPrefetch(); } } + private boolean hasReachedPrefetchLimit() { + // Reserve half of the active queue's fair share for speculative read-ahead. Retained bytes + // include in-flight events, so polling without ACK cannot bypass the memory target. Allow an + // empty delivery queue to make progress even when a single entry exceeds the target; the + // shared memory manager still enforces the hard allocation limit. + return prefetchingQueue.size() >= MAX_PREFETCHING_QUEUE_SIZE + || (!prefetchingQueue.isEmpty() + && retainedTabletBytes.get() + >= Math.max(1L, subscriptionMemoryHandle.getMemoryQuotaInBytes() / 2L)); + } + + private boolean hasReachedPrefetchLimit(final DeliveryBatchState batchState) { + // The current batch becomes one more event when flushed. Stop conversion before that event + // fills the delivery queue so the unprocessed pending suffix stays queued. + return hasReachedPrefetchLimit() + || (!batchState.isEmpty() && prefetchingQueue.size() >= MAX_PREFETCHING_QUEUE_SIZE - 1); + } + private SubscriptionQueueRejectionReason getRealtimeAdmissionRejectionReason() { if (!isActive || closeRequested || isClosed) { return SubscriptionQueueRejectionReason.INACTIVE_OR_CLOSED; @@ -724,13 +744,8 @@ private SubscriptionQueueRejectionReason getRealtimeAdmissionRejectionReason() { if (realtimeAdmissionBlocked.get()) { return realtimeAdmissionBlockReason; } - // Allow an empty queue to probe an entry even with a zero quota so materialization can report - // an explicit oversized-entry error. Pending requests have separate consensus memory - // accounting. - if (subscriptionMemoryHandle.getUsedMemorySizeInBytes() > 0L - && subscriptionMemoryHandle.getFreeMemorySizeInBytes() <= 0L) { - return toQueueRejectionReason(subscriptionMemoryHandle.inspectRejection(1L)); - } + // Pending requests have their own bounded queue and consensus memory accounting. Subscription + // memory pressure must stop materialization without forcing admitted realtime data into WAL. return SubscriptionQueueRejectionReason.NONE; } @@ -878,7 +893,7 @@ public SubscriptionEvent poll(final String consumerId, final RegionProgress regi return null; } final SubscriptionEvent event = pollInternal(consumerId); - if (Objects.nonNull(event) && prefetchingQueue.size() < MAX_PREFETCHING_QUEUE_SIZE) { + if (Objects.nonNull(event) && !hasReachedPrefetchLimit()) { requestPrefetch(); } else if (Objects.isNull(event) && shouldRecoverPrefetchBindingAfterEmptyPoll()) { requestPrefetch(); @@ -1665,24 +1680,22 @@ private PrefetchRoundResult drivePrefetchOnceInternal() { resetRoundStateForSeek(seekGeneration.get()); return PrefetchRoundResult.rescheduleNow(); } - if (prefetchingQueue.size() >= MAX_PREFETCHING_QUEUE_SIZE - || !realtimeEntriesByWriter.isEmpty()) { - if (!realtimeEntriesByWriter.isEmpty()) { - blockRealtimeAdmission(SubscriptionQueueRejectionReason.WRITER_BACKLOG); - } + if (!realtimeEntriesByWriter.isEmpty()) { + blockRealtimeAdmission(SubscriptionQueueRejectionReason.WRITER_BACKLOG); + return computeIdleRoundResult(walScanBudget); + } + unblockRealtimeAdmission(); + if (hasReachedPrefetchLimit()) { return computeIdleRoundResult(walScanBudget); } if (shouldWaitForSubscriptionMemory()) { - blockRealtimeAdmission(memoryBlockReason); if (!lingerBatch.isEmpty() && !flushBatch(lingerBatch, observedSeekGeneration)) { resetRoundStateForSeek(seekGeneration.get()); return PrefetchRoundResult.rescheduleNow(); } return PrefetchRoundResult.rescheduleAfter(MEMORY_RETRY_SLEEP_MS); } - unblockRealtimeAdmission(); - - final List batch = drainPendingEntries(maxWalEntries); + final List batch = snapshotPendingEntries(maxWalEntries); if (!batch.isEmpty()) { LOGGER.debug( DataNodePipeMessages @@ -1694,20 +1707,15 @@ private PrefetchRoundResult drivePrefetchOnceInternal() { nextExpectedSearchIndex.get(), prefetchingQueue.size()); - final MaterializationResult batchResult; - try { - batchResult = - accumulateFromPending( - batch, - lingerBatch, - observedSeekGeneration, - maxWalEntries, - maxTablets, - maxBatchBytes, - walScanBudget); - } finally { - pendingEntries.release(batch); - } + final MaterializationResult batchResult = + accumulateFromPending( + batch, + lingerBatch, + observedSeekGeneration, + maxWalEntries, + maxTablets, + maxBatchBytes, + walScanBudget); if (batchResult != MaterializationResult.SUCCESS) { if (batchResult == MaterializationResult.WAL_SCAN_YIELDED) { if (!lingerBatch.isEmpty() && !flushBatch(lingerBatch, observedSeekGeneration)) { @@ -1724,8 +1732,7 @@ private PrefetchRoundResult drivePrefetchOnceInternal() { } if (batchResult == MaterializationResult.MEMORY_BLOCKED) { // Publish already reserved Tablets immediately so ACK can release their memory. The - // blocked request and the drained suffix are recovered from WAL on a later round. - blockRealtimeAdmission(memoryBlockReason); + // blocked request and the unprocessed suffix remain in the bounded pending queue. if (!lingerBatch.isEmpty() && !flushBatch(lingerBatch, observedSeekGeneration)) { resetRoundStateForSeek(seekGeneration.get()); return PrefetchRoundResult.rescheduleNow(); @@ -1744,7 +1751,6 @@ private PrefetchRoundResult drivePrefetchOnceInternal() { return PrefetchRoundResult.rescheduleNow(); } if (walResult == MaterializationResult.MEMORY_BLOCKED) { - blockRealtimeAdmission(memoryBlockReason); return PrefetchRoundResult.rescheduleAfter(MEMORY_RETRY_SLEEP_MS); } if (walResult == MaterializationResult.STALE) { @@ -1759,6 +1765,15 @@ private PrefetchRoundResult drivePrefetchOnceInternal() { return PrefetchRoundResult.rescheduleNow(); } + // Publish reserved Tablets when read-ahead pauses. Leaving this batch lingering at the + // delivery limit would keep rescheduling pending work that cannot yet be materialized. + if (!lingerBatch.isEmpty() && hasReachedPrefetchLimit(lingerBatch)) { + if (!flushBatch(lingerBatch, observedSeekGeneration)) { + resetRoundStateForSeek(seekGeneration.get()); + return PrefetchRoundResult.rescheduleNow(); + } + } + if (!lingerBatch.isEmpty() && lingerBatch.firstTabletTimeMs > 0L) { final long lingerElapsedMs = System.currentTimeMillis() - lingerBatch.firstTabletTimeMs; if (lingerElapsedMs >= batchMaxDelayMs) { @@ -1838,11 +1853,15 @@ private void resetRoundStateForSeek(final long newSeekGeneration) { observedSeekGeneration = newSeekGeneration; } - private List drainPendingEntries(final int maxWalEntries) { + private List snapshotPendingEntries(final int maxWalEntries) { + // The queue read lock excludes lifecycle clears. Concurrent producers may only append, so + // retain each request and its consensus memory reservation until it is actually consumed. final List batch = new ArrayList<>(); - IndexedConsensusRequest next; - while (batch.size() < maxWalEntries && Objects.nonNull(next = pendingEntries.poll())) { - batch.add(next); + for (final IndexedConsensusRequest request : pendingEntries) { + if (batch.size() >= maxWalEntries) { + break; + } + batch.add(request); } return batch; } @@ -1858,7 +1877,7 @@ private PrefetchRoundResult computeIdleRoundResult( if (isClosed || !prefetchInitialized || !isActive) { return PrefetchRoundResult.dormant(); } - if (prefetchingQueue.size() >= MAX_PREFETCHING_QUEUE_SIZE) { + if (hasReachedPrefetchLimit()) { return PrefetchRoundResult.dormant(); } if (hasImmediatePrefetchableWork(walScanBudget)) { @@ -1985,7 +2004,10 @@ private MaterializationResult appendRealtimeRequest( * Accumulates tablets from pending entries into the linger buffer. When pending replay outruns * the local WAL reader, this method backfills the local-index gap from WAL before continuing. * - * @return the materialization result for the drained pending batch + *

Requests are removed only after materialization or deduplication. Memory pressure, delivery + * capacity, and temporary WAL gaps leave the unprocessed suffix available for realtime delivery. + * + * @return the materialization result for the pending batch */ private MaterializationResult accumulateFromPending( final List batch, @@ -2002,6 +2024,9 @@ private MaterializationResult accumulateFromPending( try { for (int index = 0; index < batch.size(); index++) { + if (hasReachedPrefetchLimit(lingerBatch)) { + break; + } final IndexedConsensusRequest request = batch.get(index); final long searchIndex = request.getSearchIndex(); @@ -2028,21 +2053,27 @@ private MaterializationResult accumulateFromPending( if (gapFillResult != MaterializationResult.SUCCESS) { return gapFillResult; } + if (hasReachedPrefetchLimit(lingerBatch)) { + break; + } } if (isBeforeLocalCursor(request)) { skippedCount++; + pendingEntries.consume(request); continue; } if (shouldSkipForRecoveryProgress(request)) { skippedCount++; advanceLocalCursorIfPresent(request); + pendingEntries.consume(request); continue; } if (shouldSkipForMaterializedProgress(request)) { skippedCount++; advanceLocalCursorIfPresent(request); + pendingEntries.consume(request); continue; } @@ -2055,7 +2086,8 @@ private MaterializationResult accumulateFromPending( markMaterializedProgress(request); processedCount++; advanceLocalCursorIfPresent(request); - if (prefetchingQueue.size() >= MAX_PREFETCHING_QUEUE_SIZE) { + pendingEntries.consume(request); + if (hasReachedPrefetchLimit(lingerBatch)) { break; } } @@ -2089,10 +2121,9 @@ private MaterializationResult accumulateFromPending( * Fills a gap in the pending queue by reading entries from WAL so the internal local replay * cursor stays contiguous even when pending delivery jumps ahead of the WAL iterator. * - *

Temporary WAL visibility lag is treated as a normal back-pressure condition: once a drained - * pending batch encounters an unresolved local-index gap, the queue backs off and lets the - * affected suffix fall back to the WAL path on later rounds. This keeps replay contiguous without - * requeueing the drained batch back into the bounded pending queue. + *

Temporary WAL visibility lag is treated as a normal back-pressure condition: an unresolved + * local-index gap pauses materialization while the affected suffix remains in the bounded pending + * queue. Later rounds backfill the gap before resuming pending delivery. * * @return the materialization result for the WAL gap */ @@ -2232,7 +2263,7 @@ private MaterializationResult pumpFromSubscriptionWAL( int entriesRead = 0; while (entriesRead < maxWalEntries - && prefetchingQueue.size() < MAX_PREFETCHING_QUEUE_SIZE + && !hasReachedPrefetchLimit(batchState) && subscriptionWALIterator.hasNext(walScanBudget)) { reportUnreadableWalReplayIfNecessary(); try { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueDataNodeMemoryTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueDataNodeMemoryTest.java index 16f675a4da26..7a99e139733d 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueDataNodeMemoryTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueDataNodeMemoryTest.java @@ -19,6 +19,7 @@ package org.apache.iotdb.db.subscription.broker.consensus; +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.consensus.common.request.IndexedConsensusRequest; @@ -50,6 +51,7 @@ import java.lang.reflect.Constructor; import java.lang.reflect.Field; import java.lang.reflect.Method; +import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.List; @@ -76,6 +78,149 @@ public class ConsensusPrefetchingQueueDataNodeMemoryTest { @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder(); + @Test + public void testFullNodeMemoryDoesNotRejectRealtimeRequests() throws Exception { + final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); + final CommonConfig config = CommonDescriptor.getInstance().getConfig(); + final int originalBatchMaxDelay = config.getSubscriptionConsensusBatchMaxDelayInMs(); + final List queues = new ArrayList<>(); + final long oneTabletBytes = PipeMemoryWeightUtil.calculateTabletSizeInBytes(createTablet()); + final SubscriptionMemoryManager memoryManager = + new SubscriptionMemoryManager(oneTabletBytes * 8L); + try { + config.setSubscriptionConsensusBatchMaxDelayInMs(0); + final ConsensusSubscriptionCommitManager commitManager = + newCommitManager(temporaryFolder.newFolder("full-node-realtime-admission")); + for (int region = 1; region <= 4; region++) { + final FakeConsensusReqReader reader = new FakeConsensusReqReader(); + reader.currentSearchIndex = 2L; + final ConsensusPrefetchingQueue queue = + newQueue( + "consumerGroup", + new DataRegionId(region), + reader, + newConverter(new AtomicInteger()), + commitManager, + TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE); + queues.add(queue); + queue.setSubscriptionMemoryManager(memoryManager); + assertNull(queue.poll("consumer")); + assertTrue(pendingEntries(queue).offer(createRequest(1L))); + queue.drivePrefetchOnce(); + } + // Exhaust the remaining budget through another subscription owner. Pending requests use + // consensus memory and must remain available for direct delivery after materialization + // resumes. + assertTrue(memoryManager.tryAllocate(oneTabletBytes * 4L)); + assertEquals(0L, memoryManager.getFreeMemorySizeInBytes()); + for (final ConsensusPrefetchingQueue queue : queues) { + assertTrue(pendingEntries(queue).offer(createRequest(2L))); + queue.drivePrefetchOnce(); + assertEquals(1, pendingEntries(queue).size()); + assertEquals("false", queue.coreReportMessage().get("realtimeAdmissionBlocked")); + final SubscriptionEvent event = queue.poll("consumer"); + assertNotNull(event); + assertTrue(queue.ack("consumer", event.getCommitContext())); + } + memoryManager.release(oneTabletBytes * 4L); + for (final ConsensusPrefetchingQueue queue : queues) { + queue.drivePrefetchOnce(); + final SubscriptionEvent event = queue.poll("consumer"); + assertNotNull(event); + assertEquals(2L, event.getCommitContext().getWriterProgress().getLocalSeq()); + assertTrue(queue.ack("consumer", event.getCommitContext())); + assertEquals(2L, queue.getPendingPathAcceptedEntries()); + assertEquals(0L, queue.getWalPathAcceptedEntries()); + assertEquals(0L, queue.getRealtimeAdmissionRejectionCount()); + assertEquals(0L, queue.getRetainedRequestBytes()); + } + assertEquals(0L, memoryManager.getUsedMemorySizeInBytes()); + } finally { + queues.forEach(ConsensusPrefetchingQueue::close); + memoryManager.release(oneTabletBytes * 4L); + config.setSubscriptionConsensusBatchMaxDelayInMs(originalBatchMaxDelay); + IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir); + } + } + + @Test + public void testFourRegionsBoundReadAheadAndResumePendingDeliveryAfterAck() throws Exception { + final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); + final CommonConfig config = CommonDescriptor.getInstance().getConfig(); + final int originalBatchMaxDelay = config.getSubscriptionConsensusBatchMaxDelayInMs(); + final int originalMaxTablets = config.getSubscriptionConsensusBatchMaxTabletCount(); + final List queues = new ArrayList<>(); + final List conversionCounts = new ArrayList<>(); + final long oneTabletBytes = PipeMemoryWeightUtil.calculateTabletSizeInBytes(createTablet()); + final SubscriptionMemoryManager memoryManager = + new SubscriptionMemoryManager(oneTabletBytes * 16L); + final int requestCount = 12; + try { + config.setSubscriptionConsensusBatchMaxDelayInMs(0); + config.setSubscriptionConsensusBatchMaxTabletCount(1); + final ConsensusSubscriptionCommitManager commitManager = + newCommitManager(temporaryFolder.newFolder("four-region-read-ahead")); + for (int region = 1; region <= 4; region++) { + final FakeConsensusReqReader reader = new FakeConsensusReqReader(); + reader.currentSearchIndex = requestCount; + final AtomicInteger conversionCount = new AtomicInteger(); + conversionCounts.add(conversionCount); + final ConsensusPrefetchingQueue queue = + newQueue( + "consumerGroup", + new DataRegionId(region), + reader, + newConverter(conversionCount), + commitManager, + TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE); + queues.add(queue); + queue.setSubscriptionMemoryManager(memoryManager); + assertNull(queue.poll("consumer")); + for (long index = 1L; index <= requestCount; index++) { + assertTrue(pendingEntries(queue).offer(createRequest(index))); + } + } + + for (int delivered = 0; delivered < requestCount; delivered += 2) { + for (int region = 0; region < queues.size(); region++) { + final ConsensusPrefetchingQueue queue = queues.get(region); + queue.drivePrefetchOnce(); + assertEquals(2, queue.getPrefetchedEventCount()); + assertEquals(requestCount - delivered - 2, pendingEntries(queue).size()); + assertEquals(delivered + 2, conversionCounts.get(region).get()); + // A paused consumer must stop further conversion while preserving the pending suffix. + queue.drivePrefetchOnce(); + assertEquals(delivered + 2, conversionCounts.get(region).get()); + } + assertTrue( + memoryManager.getUsedMemorySizeInBytes() + <= memoryManager.getTotalMemorySizeInBytes() / 2L); + assertEquals(0L, memoryManager.getOvercommitSizeInBytes()); + for (final ConsensusPrefetchingQueue queue : queues) { + for (long index = delivered + 1L; index <= delivered + 2L; index++) { + final SubscriptionEvent event = queue.poll("consumer"); + assertNotNull(event); + assertEquals(index, event.getCommitContext().getWriterProgress().getLocalSeq()); + assertTrue(queue.ack("consumer", event.getCommitContext())); + } + } + assertEquals(0L, memoryManager.getUsedMemorySizeInBytes()); + } + for (final ConsensusPrefetchingQueue queue : queues) { + assertEquals(requestCount, queue.getPendingPathAcceptedEntries()); + assertEquals(0L, queue.getWalPathAcceptedEntries()); + assertEquals(0L, queue.getWalGapSkippedEntries()); + assertEquals(0L, queue.getRetainedRequestBytes()); + assertEquals(0L, queue.getRealtimeAdmissionRejectionCount()); + } + } finally { + queues.forEach(ConsensusPrefetchingQueue::close); + config.setSubscriptionConsensusBatchMaxDelayInMs(originalBatchMaxDelay); + config.setSubscriptionConsensusBatchMaxTabletCount(originalMaxTablets); + IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir); + } + } + @Test public void testLargeQueueQuotaDoesNotBlockSmallQueueAndPreservesPendingEntries() throws Exception { @@ -133,15 +278,12 @@ public void testLargeQueueQuotaDoesNotBlockSmallQueueAndPreservesPendingEntries( assertEquals(1, largeConversions.get()); assertEquals(1, pendingEntries(largeQueue).size()); assertEquals(retainedPendingBytes, largeQueue.getRetainedRequestBytes()); - assertFalse(pendingEntries(largeQueue).offer(createRequest(3L))); + assertTrue(pendingEntries(largeQueue).offer(createRequest(3L))); assertEquals( - SubscriptionQueueRejectionReason.SUBSCRIPTION_MEMORY_QUOTA, + SubscriptionQueueRejectionReason.NONE, ((SubscriptionQueueAdmission) pendingEntries(largeQueue)).getLastRejectionReason()); - assertEquals(1L, largeQueue.getRealtimeAdmissionRejectionCount()); - assertEquals( - 1L, - largeQueue.getRealtimeAdmissionRejectionCount( - SubscriptionQueueRejectionReason.SUBSCRIPTION_MEMORY_QUOTA)); + assertEquals(0L, largeQueue.getRealtimeAdmissionRejectionCount()); + assertEquals(2, pendingEntries(largeQueue).size()); smallReader.currentSearchIndex = 1L; assertTrue(pendingEntries(smallQueue).offer(createRequest(1L))); @@ -158,11 +300,21 @@ public void testLargeQueueQuotaDoesNotBlockSmallQueueAndPreservesPendingEntries( assertTrue(largeQueue.ack("largeConsumer", largeEvent.getCommitContext())); largeQueue.drivePrefetchOnce(); assertEquals(2, largeConversions.get()); - assertEquals(0, pendingEntries(largeQueue).size()); - assertEquals(0L, largeQueue.getRetainedRequestBytes()); + assertEquals(1, pendingEntries(largeQueue).size()); + assertTrue(largeQueue.getRetainedRequestBytes() > 0L); final SubscriptionEvent secondLargeEvent = largeQueue.poll("largeConsumer"); assertNotNull(secondLargeEvent); assertTrue(largeQueue.ack("largeConsumer", secondLargeEvent.getCommitContext())); + largeQueue.drivePrefetchOnce(); + final SubscriptionEvent thirdLargeEvent = largeQueue.poll("largeConsumer"); + assertNotNull(thirdLargeEvent); + assertEquals(3L, thirdLargeEvent.getCommitContext().getWriterProgress().getLocalSeq()); + assertTrue(largeQueue.ack("largeConsumer", thirdLargeEvent.getCommitContext())); + assertEquals(3, largeConversions.get()); + assertEquals(0, pendingEntries(largeQueue).size()); + assertEquals(0L, largeQueue.getRetainedRequestBytes()); + assertEquals(3L, largeQueue.getPendingPathAcceptedEntries()); + assertEquals(0L, largeQueue.getWalPathAcceptedEntries()); assertEquals(0L, memoryManager.getUsedMemorySizeInBytes()); } finally { if (largeQueue != null) { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueTest.java index 7f9c4139040c..eb2c86091e82 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueTest.java @@ -1793,11 +1793,18 @@ protected ProgressWALIterator createSubscriptionWALIterator( queue.drivePrefetchOnce(); assertEquals(2, conversionCount.get()); assertEquals(3L, queue.getCurrentReadSearchIndex()); - assertTrue(pendingEntries(queue).isEmpty()); + assertEquals(1, pendingEntries(queue).size()); + assertTrue(queue.getRetainedRequestBytes() > 0L); + verify(iterator, times(2)).next(); queue.drivePrefetchOnce(); - assertEquals(4, conversionCount.get()); - assertEquals(5L, queue.getCurrentReadSearchIndex()); + assertEquals(5, conversionCount.get()); + assertEquals(6L, queue.getCurrentReadSearchIndex()); + assertTrue(pendingEntries(queue).isEmpty()); + assertEquals(0L, queue.getRetainedRequestBytes()); + assertEquals(1L, queue.getPendingPathAcceptedEntries()); + assertEquals(4L, queue.getWalPathAcceptedEntries()); + verify(iterator, times(4)).next(); queue.drivePrefetchOnce(); assertEquals(5, conversionCount.get()); @@ -2556,9 +2563,12 @@ public void testCleanupReconcilesUnindexedTabletReservation() throws Exception { @Test public void testMemoryBackpressureStopsFurtherMaterializationUntilAck() throws Exception { final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); + final int originalBatchMaxDelay = + CommonDescriptor.getInstance().getConfig().getSubscriptionConsensusBatchMaxDelayInMs(); final File systemDir = temporaryFolder.newFolder("system-memory-backpressure"); ConsensusPrefetchingQueue queue = null; try { + CommonDescriptor.getInstance().getConfig().setSubscriptionConsensusBatchMaxDelayInMs(0); final DataRegionId regionId = new DataRegionId(1); final FakeConsensusReqReader reader = new FakeConsensusReqReader(); final IoTConsensusServerImpl serverImpl = mock(IoTConsensusServerImpl.class); @@ -2617,41 +2627,34 @@ public void testMemoryBackpressureStopsFurtherMaterializationUntilAck() throws E assertEquals(oneTabletBytes, queue.getRetainedTabletBytes()); assertTrue(memoryManager.getFreeMemorySizeInBytes() > 0L); assertEquals(1, queue.getPrefetchedEventCount()); - assertTrue(pendingEntries(queue).isEmpty()); - assertEquals("true", queue.coreReportMessage().get("realtimeAdmissionBlocked")); - assertFalse(pendingEntries(queue).offer(createRequest(4L))); + assertEquals(2, pendingEntries(queue).size()); + assertTrue(queue.getRetainedRequestBytes() > 0L); + assertEquals("false", queue.coreReportMessage().get("realtimeAdmissionBlocked")); + assertTrue(pendingEntries(queue).offer(createRequest(4L))); + reader.currentSearchIndex = 4L; queue.drivePrefetchOnce(); assertEquals(2, conversionCount.get()); assertEquals(oneTabletBytes, queue.getRetainedTabletBytes()); - final SubscriptionEvent event = queue.poll("consumer"); - assertNotNull(event); - assertTrue(queue.ack("consumer", event.getCommitContext())); - assertEquals(0L, queue.getRetainedTabletBytes()); - - queue.drivePrefetchOnce(); - assertEquals("false", queue.coreReportMessage().get("realtimeAdmissionBlocked")); - assertTrue(pendingEntries(queue).offer(secondRequest)); - assertTrue(pendingEntries(queue).offer(thirdRequest)); - queue.drivePrefetchOnce(); - assertEquals(4, conversionCount.get()); - assertEquals(3L, queue.getCurrentReadSearchIndex()); - assertEquals(oneTabletBytes, queue.getRetainedTabletBytes()); - assertEquals(1, queue.getPrefetchedEventCount()); - - final SubscriptionEvent secondEvent = queue.poll("consumer"); - assertNotNull(secondEvent); - assertTrue(queue.ack("consumer", secondEvent.getCommitContext())); - queue.drivePrefetchOnce(); - assertTrue(pendingEntries(queue).offer(thirdRequest)); - queue.drivePrefetchOnce(); + for (long index = 1L; index <= 4L; index++) { + final SubscriptionEvent event = queue.poll("consumer"); + assertNotNull(event); + assertEquals(index, event.getCommitContext().getWriterProgress().getLocalSeq()); + assertTrue(queue.ack("consumer", event.getCommitContext())); + assertEquals(0L, queue.getRetainedTabletBytes()); + queue.drivePrefetchOnce(); + } - assertEquals(5, conversionCount.get()); - assertEquals(4L, queue.getCurrentReadSearchIndex()); + assertEquals(7, conversionCount.get()); + assertEquals(5L, queue.getCurrentReadSearchIndex()); assertEquals("0", queue.coreReportMessage().get("pendingEntriesSize")); assertEquals("0", queue.coreReportMessage().get("bufferedRealtimeEntryCount")); - assertEquals(oneTabletBytes, queue.getRetainedTabletBytes()); + assertEquals(0L, queue.getRetainedRequestBytes()); + assertEquals(0L, queue.getRetainedTabletBytes()); + assertEquals(4L, queue.getPendingPathAcceptedEntries()); + assertEquals(0L, queue.getWalPathAcceptedEntries()); + assertEquals(0L, queue.getRealtimeAdmissionRejectionCount()); queue.close(); queue = null; @@ -2660,6 +2663,9 @@ public void testMemoryBackpressureStopsFurtherMaterializationUntilAck() throws E if (queue != null) { queue.close(); } + CommonDescriptor.getInstance() + .getConfig() + .setSubscriptionConsensusBatchMaxDelayInMs(originalBatchMaxDelay); IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir); } } @@ -2678,11 +2684,11 @@ public void testPrefetchQueueCapacityDoesNotDisableRealtimeAdmission() throws Ex try { final int prefetchingQueueCapacity = getMaxPrefetchingQueueSize(); final int requestCount = prefetchingQueueCapacity + 1; - CommonDescriptor.getInstance().getConfig().setSubscriptionConsensusBatchMaxDelayInMs(0); + CommonDescriptor.getInstance().getConfig().setSubscriptionConsensusBatchMaxDelayInMs(60_000); CommonDescriptor.getInstance().getConfig().setSubscriptionConsensusBatchMaxTabletCount(1); CommonDescriptor.getInstance() .getConfig() - .setSubscriptionConsensusBatchMaxWalEntries(prefetchingQueueCapacity); + .setSubscriptionConsensusBatchMaxWalEntries(requestCount); final DataRegionId regionId = new DataRegionId(1); final FakeConsensusReqReader reader = new FakeConsensusReqReader(); @@ -2719,25 +2725,30 @@ public void testPrefetchQueueCapacityDoesNotDisableRealtimeAdmission() throws Ex true); queue.setSubscriptionMemoryManager(new SubscriptionMemoryManager(16L * 1024 * 1024)); - reader.currentSearchIndex = requestCount; + reader.currentSearchIndex = requestCount + 1L; assertNull(queue.poll("consumer")); - for (long searchIndex = 1L; searchIndex <= prefetchingQueueCapacity; searchIndex++) { + for (long searchIndex = 1L; searchIndex <= requestCount; searchIndex++) { assertTrue(pendingEntries(queue).offer(createRequest(searchIndex))); } queue.drivePrefetchOnce(); assertEquals(prefetchingQueueCapacity, queue.getPrefetchedEventCount()); + assertEquals(1, pendingEntries(queue).size()); + assertTrue(queue.getRetainedRequestBytes() > 0L); assertEquals("false", queue.coreReportMessage().get("realtimeAdmissionBlocked")); - assertTrue(pendingEntries(queue).offer(createRequest(requestCount))); + assertTrue(pendingEntries(queue).offer(createRequest(requestCount + 1L))); assertNotNull(queue.poll("consumer")); queue.drivePrefetchOnce(); assertEquals(prefetchingQueueCapacity, queue.getPrefetchedEventCount()); assertEquals(requestCount, conversionCount.get()); - assertTrue(pendingEntries(queue).isEmpty()); + assertEquals(1, pendingEntries(queue).size()); assertEquals("false", queue.coreReportMessage().get("realtimeAdmissionBlocked")); + queue.close(); + assertEquals(0L, queue.getRetainedRequestBytes()); + queue = null; } finally { if (queue != null) { queue.close(); @@ -2818,16 +2829,18 @@ public void testWideTablePausedConsumerKeepsMaterializedMemoryBounded() throws E assertEquals(oneTabletBytes * 2L, queue.getRetainedTabletBytes()); assertTrue(queue.getRetainedTabletBytes() <= queue.getSubscriptionMemoryLimitInBytes()); assertEquals(1, queue.getPrefetchedEventCount()); - assertEquals("0", queue.coreReportMessage().get("pendingEntriesSize")); + assertEquals( + String.valueOf(writeCount - 2), queue.coreReportMessage().get("pendingEntriesSize")); + assertTrue(queue.getRetainedRequestBytes() > 0L); assertEquals("0", queue.coreReportMessage().get("bufferedRealtimeEntryCount")); - assertEquals("true", queue.coreReportMessage().get("realtimeAdmissionBlocked")); + assertEquals("false", queue.coreReportMessage().get("realtimeAdmissionBlocked")); for (int round = 0; round < backpressureVerificationRounds; round++) { queue.drivePrefetchOnce(); } assertEquals(3, conversionCount.get()); assertEquals(oneTabletBytes * 2L, queue.getRetainedTabletBytes()); - assertFalse(pendingEntries(queue).offer(createRequest(writeCount + 1L))); + assertTrue(pendingEntries(queue).offer(createRequest(writeCount + 1L))); final SubscriptionEvent pausedEvent = queue.poll("pausedConsumer"); assertNotNull(pausedEvent); @@ -2838,8 +2851,10 @@ public void testWideTablePausedConsumerKeepsMaterializedMemoryBounded() throws E } assertEquals(3, conversionCount.get()); assertEquals(oneTabletBytes * 2L, queue.getRetainedTabletBytes()); + assertEquals(writeCount - 1, pendingEntries(queue).size()); queue.close(); + assertEquals(0L, queue.getRetainedRequestBytes()); queue = null; assertEquals(0L, memoryManager.getUsedMemorySizeInBytes()); } finally { 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 377f20801c4d..192ed7c3e9c8 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 @@ -197,7 +197,7 @@ public void testYieldedGapScanKeepsIteratorAndDeliversEachRequestOnce() throws E } @Test - public void testAckRecoversDrainedSuffixFromWalWithoutReoffer() throws Exception { + public void testAckResumesPendingSuffixWithoutWalReplay() throws Exception { final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); final CommonConfig config = CommonDescriptor.getInstance().getConfig(); final int originalBatchMaxWalEntries = config.getSubscriptionConsensusBatchMaxWalEntries(); @@ -213,7 +213,7 @@ public void testAckRecoversDrainedSuffixFromWalWithoutReoffer() throws Exception config.setSubscriptionConsensusBatchMaxWalEntries(128); config.setSubscriptionConsensusBatchMaxTabletCount(128); config.setSubscriptionConsensusBatchMaxSizeInBytes(oneTabletBytes * 16L); - config.setSubscriptionConsensusBatchMaxDelayInMs(60_000); + config.setSubscriptionConsensusBatchMaxDelayInMs(0); writeSealedWal(walDirectory); @@ -272,8 +272,9 @@ public void testAckRecoversDrainedSuffixFromWalWithoutReoffer() throws Exception assertEquals(Arrays.asList(1L, 2L), conversionAttempts); assertEquals(2L, queue.getCurrentReadSearchIndex()); assertEquals(1, queue.getPrefetchedEventCount()); - assertEquals(0, pendingEntries(queue).size()); - assertEquals("true", queue.coreReportMessage().get("realtimeAdmissionBlocked")); + assertEquals(2, pendingEntries(queue).size()); + assertTrue(queue.getRetainedRequestBytes() > 0L); + assertEquals("false", queue.coreReportMessage().get("realtimeAdmissionBlocked")); assertEquals(oneTabletBytes, memoryManager.getUsedMemorySizeInBytes()); assertMemoryBounded(queue, memoryManager); @@ -302,8 +303,10 @@ public void testAckRecoversDrainedSuffixFromWalWithoutReoffer() throws Exception assertEquals(Arrays.asList(1L, 2L, 3L), deliveredLocalSequences); assertEquals(Arrays.asList(1L, 2L, 2L, 3L, 3L), conversionAttempts); - assertEquals(1L, queue.getPendingPathAcceptedEntries()); - assertEquals(2L, queue.getWalPathAcceptedEntries()); + assertEquals(3L, queue.getPendingPathAcceptedEntries()); + assertEquals(0L, queue.getWalPathAcceptedEntries()); + assertEquals(0L, queue.getRetainedRequestBytes()); + assertEquals(0L, queue.getRealtimeAdmissionRejectionCount()); assertEquals(4L, queue.getCurrentReadSearchIndex()); assertEquals(0, queue.getPrefetchedEventCount()); assertEquals(0L, queue.getSubscriptionUncommittedEventCount()); @@ -553,7 +556,8 @@ public void close() throws IOException { 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++) { + // Leave the gap unresolved through the last round so this test measures only WAL work. + for (long index = replayedRequests + 3L; index <= requestCount; index++) { assertTrue(pendingEntries(queue).offer(createRequest(index))); } @@ -594,6 +598,8 @@ public void close() throws IOException { assertEquals(0L, queue.getWalGapSkippedEntries()); assertEquals(0L, queue.getSubscriptionUncommittedEventCount()); assertEquals(0L, queue.getRetainedTabletBytes()); + assertEquals(2 * maxRounds - 2, pendingEntries(queue).size()); + assertTrue(queue.getRetainedRequestBytes() > 0L); if (memoryBackpressure) { assertTrue(queue.getSubscriptionMemoryRejectionCount() > 0L); }