This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 3f1706cbce0 Fix subscription realtime retention under shared memory
pressure (#18870)
3f1706cbce0 is described below
commit 3f1706cbce05445ddd84e4867097f3c575bd02c8
Author: Caideyipi <[email protected]>
AuthorDate: Thu Oct 15 10:35:43 2026 +0800
Fix subscription realtime retention under shared memory pressure (#18870)
---
.../consensus/ConsensusPrefetchingQueue.java | 131 ++++++++++------
...onsensusPrefetchingQueueDataNodeMemoryTest.java | 170 +++++++++++++++++++--
.../consensus/ConsensusPrefetchingQueueTest.java | 93 ++++++-----
...nsensusPrefetchingQueueWalBackpressureTest.java | 20 ++-
4 files changed, 309 insertions(+), 105 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 c19420c130c..1dd733b4cae 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 @@ public class ConsensusPrefetchingQueue {
requestMemoryManager.free(request);
}
- private void release(final List<IndexedConsensusRequest> requests) {
- requests.forEach(this::release);
+ private synchronized void consume(final IndexedConsensusRequest request) {
+ if (super.removeFirstOccurrence(request)) {
+ release(request);
+ }
}
private long getRetainedRequestBytes() {
@@ -751,11 +753,29 @@ public class ConsensusPrefetchingQueue {
}
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;
@@ -763,13 +783,8 @@ public class ConsensusPrefetchingQueue {
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;
}
@@ -917,7 +932,7 @@ public class ConsensusPrefetchingQueue {
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();
@@ -1704,24 +1719,22 @@ public class ConsensusPrefetchingQueue {
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<IndexedConsensusRequest> batch =
drainPendingEntries(maxWalEntries);
+ final List<IndexedConsensusRequest> batch =
snapshotPendingEntries(maxWalEntries);
if (!batch.isEmpty()) {
LOGGER.debug(
DataNodePipeMessages
@@ -1733,20 +1746,15 @@ public class ConsensusPrefetchingQueue {
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)) {
@@ -1763,8 +1771,7 @@ public class ConsensusPrefetchingQueue {
}
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();
@@ -1783,7 +1790,6 @@ public class ConsensusPrefetchingQueue {
return PrefetchRoundResult.rescheduleNow();
}
if (walResult == MaterializationResult.MEMORY_BLOCKED) {
- blockRealtimeAdmission(memoryBlockReason);
return PrefetchRoundResult.rescheduleAfter(MEMORY_RETRY_SLEEP_MS);
}
if (walResult == MaterializationResult.STALE) {
@@ -1798,6 +1804,15 @@ public class ConsensusPrefetchingQueue {
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) {
@@ -1877,11 +1892,15 @@ public class ConsensusPrefetchingQueue {
observedSeekGeneration = newSeekGeneration;
}
- private List<IndexedConsensusRequest> drainPendingEntries(final int
maxWalEntries) {
+ private List<IndexedConsensusRequest> 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<IndexedConsensusRequest> 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;
}
@@ -1897,7 +1916,7 @@ public class ConsensusPrefetchingQueue {
if (isClosed || !prefetchInitialized || !isActive) {
return PrefetchRoundResult.dormant();
}
- if (prefetchingQueue.size() >= MAX_PREFETCHING_QUEUE_SIZE) {
+ if (hasReachedPrefetchLimit()) {
return PrefetchRoundResult.dormant();
}
if (hasImmediatePrefetchableWork(walScanBudget)) {
@@ -2024,7 +2043,10 @@ public class ConsensusPrefetchingQueue {
* 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
+ * <p>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<IndexedConsensusRequest> batch,
@@ -2041,6 +2063,9 @@ public class ConsensusPrefetchingQueue {
try {
for (int index = 0; index < batch.size(); index++) {
+ if (hasReachedPrefetchLimit(lingerBatch)) {
+ break;
+ }
final IndexedConsensusRequest request = batch.get(index);
final long searchIndex = request.getSearchIndex();
@@ -2067,21 +2092,27 @@ public class ConsensusPrefetchingQueue {
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;
}
@@ -2094,7 +2125,8 @@ public class ConsensusPrefetchingQueue {
markMaterializedProgress(request);
processedCount++;
advanceLocalCursorIfPresent(request);
- if (prefetchingQueue.size() >= MAX_PREFETCHING_QUEUE_SIZE) {
+ pendingEntries.consume(request);
+ if (hasReachedPrefetchLimit(lingerBatch)) {
break;
}
}
@@ -2128,10 +2160,9 @@ public class ConsensusPrefetchingQueue {
* 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.
*
- * <p>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.
+ * <p>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
*/
@@ -2271,7 +2302,7 @@ public class ConsensusPrefetchingQueue {
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 16f675a4da2..7a99e139733 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.io.File;
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<ConsensusPrefetchingQueue> 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<ConsensusPrefetchingQueue> queues = new ArrayList<>();
+ final List<AtomicInteger> 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 class ConsensusPrefetchingQueueDataNodeMemoryTest {
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 class ConsensusPrefetchingQueueDataNodeMemoryTest {
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 c8bd2734e91..32f8e5737fd 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
@@ -1808,11 +1808,18 @@ public class ConsensusPrefetchingQueueTest {
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());
@@ -2571,9 +2578,12 @@ public class ConsensusPrefetchingQueueTest {
@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);
@@ -2632,41 +2642,34 @@ public class ConsensusPrefetchingQueueTest {
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;
@@ -2675,6 +2678,9 @@ public class ConsensusPrefetchingQueueTest {
if (queue != null) {
queue.close();
}
+ CommonDescriptor.getInstance()
+ .getConfig()
+ .setSubscriptionConsensusBatchMaxDelayInMs(originalBatchMaxDelay);
IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir);
}
}
@@ -2693,11 +2699,11 @@ public class ConsensusPrefetchingQueueTest {
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();
@@ -2734,25 +2740,30 @@ public class ConsensusPrefetchingQueueTest {
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();
@@ -2833,16 +2844,18 @@ public class ConsensusPrefetchingQueueTest {
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);
@@ -2853,8 +2866,10 @@ public class ConsensusPrefetchingQueueTest {
}
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 377f20801c4..192ed7c3e9c 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 class ConsensusPrefetchingQueueWalBackpressureTest {
}
@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 class ConsensusPrefetchingQueueWalBackpressureTest {
config.setSubscriptionConsensusBatchMaxWalEntries(128);
config.setSubscriptionConsensusBatchMaxTabletCount(128);
config.setSubscriptionConsensusBatchMaxSizeInBytes(oneTabletBytes * 16L);
- config.setSubscriptionConsensusBatchMaxDelayInMs(60_000);
+ config.setSubscriptionConsensusBatchMaxDelayInMs(0);
writeSealedWal(walDirectory);
@@ -272,8 +272,9 @@ public class ConsensusPrefetchingQueueWalBackpressureTest {
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 class ConsensusPrefetchingQueueWalBackpressureTest {
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 class ConsensusPrefetchingQueueWalBackpressureTest {
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 class ConsensusPrefetchingQueueWalBackpressureTest {
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);
}
