diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/LogDispatcher.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/LogDispatcher.java index 39caeff33ed5..cd9d7eea49f8 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/LogDispatcher.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/LogDispatcher.java @@ -427,7 +427,29 @@ public void run() { } void waitForBatchAccumulation(long waitingTimeInMs) throws InterruptedException { - Thread.sleep(waitingTimeInMs); + if (waitingTimeInMs <= 0) { + return; + } + + final long deadlineNanos = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(waitingTimeInMs); + final int maxLogEntriesNumPerBatch = config.getReplication().getMaxLogEntriesNumPerBatch(); + + // Keep collecting while the batch is below its entry limit. A plain sleep makes the + // dispatcher wait for the full accumulation interval even when the batch becomes full + // immediately, which unnecessarily throttles IoTConsensus under sustained write load. + while (bufferedEntries.size() < maxLogEntriesNumPerBatch) { + final long remainingNanos = deadlineNanos - System.nanoTime(); + if (remainingNanos <= 0) { + return; + } + + final IndexedConsensusRequest request = + pendingEntries.poll(remainingNanos, TimeUnit.NANOSECONDS); + if (request == null) { + return; + } + bufferedEntries.add(request); + } } public void updateSafelyDeletedSearchIndex() { diff --git a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/logdispatcher/LogDispatcherTest.java b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/logdispatcher/LogDispatcherTest.java index 3f4ea8ab528a..fce84147dd5c 100644 --- a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/logdispatcher/LogDispatcherTest.java +++ b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/logdispatcher/LogDispatcherTest.java @@ -113,6 +113,70 @@ public void sendBatchAsync(Batch sentBatch, DispatchLogHandler handler) { } } + @Test + public void testBatchAccumulationStopsWhenBatchIsFull() throws Exception { + final Peer localPeer = createPeer(1, 6687); + final Peer remotePeer = createPeer(2, 6688); + final IoTConsensusConfig config = + IoTConsensusConfig.newBuilder() + .setReplication( + IoTConsensusConfig.Replication.newBuilder() + .setMaxLogEntriesNumPerBatch(2) + .setMaxWaitingTimeForAccumulatingBatchInMs(10_000) + .build()) + .build(); + final ScheduledExecutorService backgroundTaskService = + Executors.newSingleThreadScheduledExecutor(); + final ExecutorService executorService = Executors.newSingleThreadExecutor(); + LogDispatcher.LogDispatcherThread dispatcherThread = null; + Future dispatcherFuture = null; + try { + final IoTConsensusServerImpl server = + createServer( + localPeer, Arrays.asList(localPeer, remotePeer), config, backgroundTaskService); + final CountDownLatch batchSent = new CountDownLatch(1); + final AtomicInteger getBatchInvocations = new AtomicInteger(); + dispatcherThread = + server.getLogDispatcher().new LogDispatcherThread(remotePeer, config, 0) { + @Override + public Batch getBatch() { + return getBatchInvocations.getAndIncrement() == 0 + ? new Batch(config) + : createBatch(config, 1); + } + + @Override + public void sendBatchAsync(Batch sentBatch, DispatchLogHandler handler) { + assertEquals(0, getPendingEntriesSize()); + batchSent.countDown(); + Thread.currentThread().interrupt(); + } + }; + assertTrue( + dispatcherThread.offer( + new IndexedConsensusRequest( + 1, Collections.singletonList(new TestEntry(1, localPeer))))); + assertTrue( + dispatcherThread.offer( + new IndexedConsensusRequest( + 2, Collections.singletonList(new TestEntry(2, localPeer))))); + + dispatcherFuture = executorService.submit(dispatcherThread); + assertTrue(batchSent.await(2, TimeUnit.SECONDS)); + dispatcherFuture.get(2, TimeUnit.SECONDS); + } finally { + if (dispatcherFuture != null) { + dispatcherFuture.cancel(true); + } + executorService.shutdownNow(); + executorService.awaitTermination(5, TimeUnit.SECONDS); + if (dispatcherThread != null) { + dispatcherThread.stop(); + } + backgroundTaskService.shutdownNow(); + } + } + @Test public void testReloadConfigUpdatesExistingDispatcherPipeline() throws Exception { final Peer localPeer = createPeer(1, 6677);