Skip to content

Fix subscription prefetch wakeups for late replicated WAL entries #19194

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

Open
Caideyipi wants to merge 2 commits into
base: master
Choose a base branch
from
Open

Fix subscription prefetch wakeups for late replicated WAL entries #19194

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
2c5ac47
Fix subscription prefetch wakeup for late replicated WAL entries
Caideyipi Nov 9, 2026
3a343fa
Throttle subscription prefetch empty-poll wakeups
Caideyipi Nov 9, 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
27 changes: 24 additions & 3 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 @@ -115,6 +115,8 @@

private static final Logger LOGGER = LoggerFactory.getLogger(ConsensusPrefetchingQueue.class);

static final long EMPTY_POLL_PREFETCH_WAKEUP_INTERVAL_NANOS = TimeUnit.MILLISECONDS.toNanos(100L);

private static final ExecutorService CLOSE_CLEANUP_EXECUTOR =
IoTDBThreadPoolFactory.newCachedThreadPoolWithDaemon(
ThreadName.SUBSCRIPTION_CONSENSUS_PREFETCH_EXECUTOR_POOL.getName() + "-Close-Cleanup");
Expand Down Expand Up @@ -307,6 +309,8 @@

private final Object prefetchBindingLock = new Object();

private final AtomicLong lastEmptyPollPrefetchWakeupNanos = new AtomicLong(Long.MIN_VALUE);

private volatile ConsensusPrefetchSubtask prefetchSubtask;

private volatile ConsensusSubscriptionPrefetchExecutor prefetchExecutor;
Expand Down Expand Up @@ -943,7 +947,7 @@
}
}

private boolean shouldRecoverPrefetchBindingAfterEmptyPoll() {
private boolean shouldWakePrefetchAfterEmptyPoll() {
if (!prefetchInitialized || isClosed || closeRequested || pendingSeekRequest != null) {
return false;
}
Expand All @@ -962,7 +966,10 @@
|| prefetchExecutor.isShutdown()
|| prefetchExecutor != currentExecutor;
if (!bindingMissing) {
return false;
// Replicated WAL entries do not update the local search index or enter the realtime queue.
// Wake an idle worker on empty polls to discover entries that arrived after its last scan.
return !currentSubtask.isScheduledOrRunning()
&& tryAcquireEmptyPollPrefetchWakeup(lastEmptyPollPrefetchWakeupNanos, System.nanoTime());
}

return hasImmediatePrefetchableWork()
Expand All @@ -973,6 +980,20 @@
|| computeWatermarkDelayMs() > 0L;
}

static boolean tryAcquireEmptyPollPrefetchWakeup(
final AtomicLong lastWakeupNanos, final long currentTimeNanos) {
while (true) {
final long lastWakeupTimeNanos = lastWakeupNanos.get();
if (lastWakeupTimeNanos != Long.MIN_VALUE
&& currentTimeNanos - lastWakeupTimeNanos < EMPTY_POLL_PREFETCH_WAKEUP_INTERVAL_NANOS) {
return false;
}
if (lastWakeupNanos.compareAndSet(lastWakeupTimeNanos, currentTimeNanos)) {
return true;
}
}
}

// ======================== Poll ========================

public SubscriptionEvent poll(final String consumerId) {
Expand Down Expand Up @@ -1025,9 +1046,9 @@
final SubscriptionEvent event = pollInternal(consumerId, timer);
if (Objects.nonNull(event) && !hasReachedPrefetchLimit()) {
requestPrefetch();
} else if (Objects.isNull(event) && shouldRecoverPrefetchBindingAfterEmptyPoll()) {
} else if (Objects.isNull(event) && shouldWakePrefetchAfterEmptyPoll()) {
requestPrefetch();
}

Check warning on line 1051 in iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

This branch's code block is the same as the block for the branch on line 1047. 


See more on https://sonarcloud.io/project/issues?id=apache_iotdb&issues=AaG_9xK0tqj7RNvsA4tK&open=AaG_9xK0tqj7RNvsA4tK&pullRequest=19194
// Deliver reserved events first so ACK can drain memory. An entry that cannot fit even in an
// empty queue is an explicit configuration error, with no replay/commit progress advancement.
final long blockedEntryBytes = memoryBlockedEntryBytes;
Expand Down
130 changes: 130 additions & 0 deletions .../org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueSeekTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@
import org.apache.iotdb.db.subscription.resource.SubscriptionMemoryManager;
import org.apache.iotdb.db.subscription.task.execution.ConsensusSubscriptionPrefetchExecutor;
import org.apache.iotdb.db.subscription.task.execution.ConsensusSubscriptionPrefetchExecutorManager;
import org.apache.iotdb.db.subscription.task.subtask.ConsensusPrefetchSubtask;
import org.apache.iotdb.rpc.subscription.config.TopicConstant;
import org.apache.iotdb.rpc.subscription.payload.poll.RegionProgress;
import org.apache.iotdb.rpc.subscription.payload.poll.SubscriptionPollResponseType;
Expand Down Expand Up @@ -64,14 +65,17 @@
import java.util.Iterator;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;

import static org.awaitility.Awaitility.await;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
Expand Down Expand Up @@ -141,6 +145,132 @@ public void testMultiWriterSeekReplaysAfterResettingOrderedReaders() throws Exce
Arrays.asList(createRequest(1L, 1L, 7), createRequest(-1L, 10L, 8)), 2L, 0L, true);
}

@Test
public void testEmptyPollWakeupRateLimit() {
final AtomicLong lastWakeupNanos = new AtomicLong(Long.MIN_VALUE);
assertTrue(ConsensusPrefetchingQueue.tryAcquireEmptyPollPrefetchWakeup(lastWakeupNanos, 0L));
assertFalse(
ConsensusPrefetchingQueue.tryAcquireEmptyPollPrefetchWakeup(
lastWakeupNanos,
ConsensusPrefetchingQueue.EMPTY_POLL_PREFETCH_WAKEUP_INTERVAL_NANOS - 1L));
assertTrue(
ConsensusPrefetchingQueue.tryAcquireEmptyPollPrefetchWakeup(
lastWakeupNanos, ConsensusPrefetchingQueue.EMPTY_POLL_PREFETCH_WAKEUP_INTERVAL_NANOS));
}

@Test(timeout = 20_000L)
public void testEmptyPollReplaysRemoteEntryAfterPrefetchBecomesDormant() throws Exception {
final int originalBatchMaxDelay = config.getSubscriptionConsensusBatchMaxDelayInMs();
final int originalPrefetchThreads =
config.getSubscriptionConsensusPrefetchExecutorMaxThreadNum();
final long originalWatermarkInterval = config.getSubscriptionConsensusWatermarkIntervalMs();
final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir();
final File systemDir = temporaryFolder.newFolder();
final ConcurrentLinkedQueue<IndexedConsensusRequest> readableEntries =
new ConcurrentLinkedQueue<>();
final AtomicReference<ConsensusPrefetchSubtask> registeredSubtask = new AtomicReference<>();
ConsensusSubscriptionPrefetchExecutor executor = null;
ConsensusPrefetchingQueue queue = null;
try {
config.setSubscriptionConsensusBatchMaxDelayInMs(0);
config.setSubscriptionConsensusPrefetchExecutorMaxThreadNum(1);
config.setSubscriptionConsensusWatermarkIntervalMs(0L);
IoTDBDescriptor.getInstance().getConfig().setSystemDir(systemDir.getAbsolutePath());
executor =
new ConsensusSubscriptionPrefetchExecutor() {
@Override
public synchronized boolean register(final ConsensusPrefetchSubtask subtask) {
if (!super.register(subtask)) {
return false;
}
registeredSubtask.set(subtask);
return true;
}
};
when(manager.getExecutor()).thenReturn(executor);
when(walNode.getLogDirectory()).thenReturn(systemDir);
when(walNode.getCurrentSearchIndex()).thenReturn(0L);
when(server.getWriterSafeFrontierTracker()).thenReturn(new WriterSafeFrontierTracker());
queue =
new ConsensusPrefetchingQueue(
"consumerGroup",
"topic",
TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE,
new DataRegionId(1),
server,
new SubscriptionWalRetentionPolicy("topic", -1L, -1L),
converter,
new ConsensusSubscriptionCommitManager(
(consumerGroupId, topicName, regionId) ->
ConsensusSubscriptionCommitManager.ConfigNodeProgressQueryResult.absent()),
new RegionProgress(Collections.emptyMap()),
1L,
1L,
true) {
@Override
protected ProgressWALIterator createSubscriptionWALIterator(
final long startSearchIndex) {
return new ProgressWALIterator(systemDir, startSearchIndex) {
@Override
boolean hasNext(final ScanBudget scanBudget) {
return !readableEntries.isEmpty();
}

@Override
public IndexedConsensusRequest next() {
return readableEntries.remove();
}
};
}
};
queue.setSubscriptionMemoryManager(new SubscriptionMemoryManager(16L * 1024 * 1024));
queue.seekToBeginning();
await()
.atMost(5L, TimeUnit.SECONDS)
.until(
() ->
registeredSubtask.get() != null
&& !registeredSubtask.get().isScheduledOrRunning());
assertEquals(0L, queue.getRawWalGap());

// Replication can make a WAL entry readable after the worker becomes idle without advancing
// the local search index or admitting a request to the local realtime queue.
readableEntries.add(createRequest(-1L, 10L, 8));
final ConsensusPrefetchingQueue pollingQueue = queue;
final AtomicReference<SubscriptionEvent> received = new AtomicReference<>();
await()
.atMost(5L, TimeUnit.SECONDS)
.until(
() -> {
received.set(pollingQueue.poll("consumer"));
return received.get() != null;
});
final SubscriptionEvent event = received.get();
assertEquals(
SubscriptionPollResponseType.TABLETS.getType(),
event.getCurrentResponse().getResponseType());
final List<Tablet> tablets =
((TabletsPayload) event.getCurrentResponse().getPayload()).getTablets();
assertEquals(1, tablets.size());
assertEquals(1, tablets.get(0).getRowSize());
assertEquals(10L, tablets.get(0).getTimestamps()[0]);
assertTrue(queue.ack("consumer", event.getCommitContext()));
assertEquals(1L, queue.getCurrentReadSearchIndex());
assertEquals(0L, queue.getWalGapSkippedEntries());
} finally {
if (queue != null) {
queue.close();
}
if (executor != null) {
executor.shutdown();
}
config.setSubscriptionConsensusBatchMaxDelayInMs(originalBatchMaxDelay);
config.setSubscriptionConsensusPrefetchExecutorMaxThreadNum(originalPrefetchThreads);
config.setSubscriptionConsensusWatermarkIntervalMs(originalWatermarkInterval);
IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir);
}
}

@Test(timeout = 20_000L)
public void testResolvedSeekDoesNotMaskNewerConfigNodeProgress() throws Exception {
assertActivationProgress(false, false);
Expand Down
Loading