Skip to content

Fix subscription realtime retention under shared memory pressure #18870

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jt2594838 merged 2 commits into from
Oct 15, 2026
Merged

Fix subscription realtime retention under shared memory pressure #18870

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
5b7ba31
Fix subscription realtime retention under shared memory pressure
Caideyipi Oct 14, 2026
59b5a68
Merge master into fix/subscription-prefetch-memory-headroom
Caideyipi Oct 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
131 changes: 81 additions & 50 deletions ...ain/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 =
Expand Down Expand Up @@ -534,8 +534,10 @@ private void release(final IndexedConsensusRequest request) {
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() {
Expand Down Expand Up @@ -712,25 +714,38 @@ 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;
}
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;
}

Expand Down Expand Up @@ -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();
Expand Down Expand Up @@ -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<IndexedConsensusRequest> batch = drainPendingEntries(maxWalEntries);
final List<IndexedConsensusRequest> batch = snapshotPendingEntries(maxWalEntries);
if (!batch.isEmpty()) {
LOGGER.debug(
DataNodePipeMessages
Expand All @@ -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)) {
Expand All @@ -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();
Expand All @@ -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) {
Expand All @@ -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) {
Expand Down Expand Up @@ -1838,11 +1853,15 @@ private void resetRoundStateForSeek(final long newSeekGeneration) {
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;
}
Expand All @@ -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)) {
Expand Down Expand Up @@ -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
* <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,
Expand All @@ -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();

Expand All @@ -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;
}

Expand All @@ -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;
}
}
Expand Down Expand Up @@ -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.
*
* <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
*/
Expand Down Expand Up @@ -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 {
Expand Down
Loading
Loading