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..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; @@ -943,7 +947,7 @@ private ConsensusPrefetchSubtask ensurePrefetchSubtaskBound() { } } - private boolean shouldRecoverPrefetchBindingAfterEmptyPoll() { + private boolean shouldWakePrefetchAfterEmptyPoll() { if (!prefetchInitialized || isClosed || closeRequested || pendingSeekRequest != null) { return false; } @@ -962,7 +966,10 @@ 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() + && tryAcquireEmptyPollPrefetchWakeup(lastEmptyPollPrefetchWakeupNanos, System.nanoTime()); } return hasImmediatePrefetchableWork() @@ -973,6 +980,20 @@ private boolean shouldRecoverPrefetchBindingAfterEmptyPoll() { || 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) { @@ -1025,7 +1046,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..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 @@ -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,7 +65,9 @@ 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; @@ -72,6 +75,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 +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 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);