Avoid unnecessary IoTConsensus batch accumulation delay

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() {

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);