From 2c5ac4798c354bc4d9bed57e76486d2d58e16744 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Mon, 9 Nov 2026 16:47:04 +0800 Subject: [PATCH 1/2] Fix subscription prefetch wakeup for late replicated WAL entries Wake an existing idle prefetch task after an empty poll so replicated WAL entries can be discovered without another local write. Preserve initialization, seek, lifecycle and executor-binding guards. Add a regression covering delivery and acknowledgement without advancing the local WAL cursor. --- .../consensus/ConsensusPrefetchingQueue.java | 8 +- .../ConsensusPrefetchingQueueSeekTest.java | 116 ++++++++++++++++++ 2 files changed, 121 insertions(+), 3 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 e98d2a1adf04..0f621637e3ad 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 @@ -943,7 +943,7 @@ private ConsensusPrefetchSubtask ensurePrefetchSubtaskBound() { } } - private boolean shouldRecoverPrefetchBindingAfterEmptyPoll() { + private boolean shouldWakePrefetchAfterEmptyPoll() { if (!prefetchInitialized || isClosed || closeRequested || pendingSeekRequest != null) { return false; } @@ -962,7 +962,9 @@ private boolean shouldRecoverPrefetchBindingAfterEmptyPoll() { || 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(); } return hasImmediatePrefetchableWork() @@ -1025,7 +1027,7 @@ public SubscriptionEvent poll( 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(); } // Deliver reserved events first so ACK can drain memory. An entry that cannot fit even in an diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueSeekTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueSeekTest.java index 00f7c921f2a0..6a9ba5e9cf5a 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueSeekTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueSeekTest.java @@ -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; @@ -64,6 +65,7 @@ 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.AtomicReference; import java.util.function.Consumer; @@ -72,6 +74,7 @@ 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; @@ -141,6 +144,119 @@ public void testMultiWriterSeekReplaysAfterResettingOrderedReaders() throws Exce Arrays.asList(createRequest(1L, 1L, 7), createRequest(-1L, 10L, 8)), 2L, 0L, true); } + @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 readableEntries = + new ConcurrentLinkedQueue<>(); + final AtomicReference 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 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 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); From 3a343faaff65424068f357819ecfe97c8a41ca9d Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Mon, 9 Nov 2026 17:21:08 +0800 Subject: [PATCH 2/2] Throttle subscription prefetch empty-poll wakeups --- .../consensus/ConsensusPrefetchingQueue.java | 21 ++++++++++++++++++- .../ConsensusPrefetchingQueueSeekTest.java | 14 +++++++++++++ 2 files changed, 34 insertions(+), 1 deletion(-) 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 0f621637e3ad..a93ac4172abc 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 @@ -115,6 +115,8 @@ public class ConsensusPrefetchingQueue { 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"); @@ -307,6 +309,8 @@ public class ConsensusPrefetchingQueue { private final Object prefetchBindingLock = new Object(); + private final AtomicLong lastEmptyPollPrefetchWakeupNanos = new AtomicLong(Long.MIN_VALUE); + private volatile ConsensusPrefetchSubtask prefetchSubtask; private volatile ConsensusSubscriptionPrefetchExecutor prefetchExecutor; @@ -964,7 +968,8 @@ private boolean shouldWakePrefetchAfterEmptyPoll() { if (!bindingMissing) { // 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(); + return !currentSubtask.isScheduledOrRunning() + && tryAcquireEmptyPollPrefetchWakeup(lastEmptyPollPrefetchWakeupNanos, System.nanoTime()); } return hasImmediatePrefetchableWork() @@ -975,6 +980,20 @@ private boolean shouldWakePrefetchAfterEmptyPoll() { || 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) { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueSeekTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueSeekTest.java index 6a9ba5e9cf5a..d96d8536a770 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueSeekTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueSeekTest.java @@ -67,6 +67,7 @@ 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; @@ -144,6 +145,19 @@ 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();