This is an automated email from the ASF dual-hosted git repository. Caideyipi pushed a commit to branch fix/iot-consensus-batch-byte-accumulation in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit ed9f234cf39e2fa66f2836de5b11b1b10f011613 Author: Caideyipi <[email protected]> AuthorDate: Tue Sep 1 12:01:10 2026 +0800 Fix IoTConsensus batch accumulation by byte size --- .../iotdb/consensus/iot/logdispatcher/Batch.java | 6 +- .../consensus/iot/logdispatcher/LogDispatcher.java | 18 ++++-- .../iot/logdispatcher/LogDispatcherTest.java | 66 ++++++++++++++++++++++ 3 files changed, 83 insertions(+), 7 deletions(-) diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/Batch.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/Batch.java index 72b68ab96ac..8d28743f7ba 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/Batch.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/Batch.java @@ -63,6 +63,10 @@ public class Batch { } public boolean canAccumulate() { + return canAccumulate(config, logEntries.size(), memorySize); + } + + static boolean canAccumulate(IoTConsensusConfig config, int logEntriesSize, long memorySize) { // When reading entries from the WAL, the memory size is calculated based on the serialized // size, which can be significantly smaller than the actual size. // Thus, we add a multiplier to sender's memory size to estimate the receiver's memory cost. @@ -71,7 +75,7 @@ public class Batch { long senderMemSize = LogDispatcher.getSenderMemSizeSum().get(); double multiplier = senderMemSize > 0 ? (double) receiverMemSize / senderMemSize : 1.0; multiplier = Math.max(multiplier, 1.0); - return logEntries.size() < config.getReplication().getMaxLogEntriesNumPerBatch() + return logEntriesSize < config.getReplication().getMaxLogEntriesNumPerBatch() && ((long) (memorySize * multiplier)) < config.getReplication().getMaxSizePerBatch(); } 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 cd9d7eea49f..94d9cb64434 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 @@ -432,12 +432,16 @@ public class LogDispatcher { } 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 IoTConsensusConfig currentConfig = config; + int accumulatedEntries = bufferedEntries.size(); + long accumulatedMemorySize = + bufferedEntries.stream().mapToLong(IndexedConsensusRequest::getMemorySize).sum(); + + // Keep collecting while the batch is below both its entry and memory limits. A plain sleep, + // or checking only the entry limit, makes the dispatcher wait for the full accumulation + // interval after a batch has already reached its memory limit. This unnecessarily throttles + // IoTConsensus when each request contains a large tablet. + while (Batch.canAccumulate(currentConfig, accumulatedEntries, accumulatedMemorySize)) { final long remainingNanos = deadlineNanos - System.nanoTime(); if (remainingNanos <= 0) { return; @@ -449,6 +453,8 @@ public class LogDispatcher { return; } bufferedEntries.add(request); + accumulatedEntries++; + accumulatedMemorySize += request.getMemorySize(); } } 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 fce84147dd5..a6e299748d2 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 @@ -177,6 +177,72 @@ public class LogDispatcherTest { } } + @Test + public void testBatchAccumulationStopsWhenMemoryLimitIsReached() throws Exception { + final Peer localPeer = createPeer(1, 6697); + final Peer remotePeer = createPeer(2, 6698); + final IoTConsensusConfig config = + IoTConsensusConfig.newBuilder() + .setReplication( + IoTConsensusConfig.Replication.newBuilder() + .setMaxLogEntriesNumPerBatch(1024) + .setMaxSizePerBatch(1) + .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, Collections.singletonList(localPeer), 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(1, getPendingEntriesSize()); + assertEquals(1, getBufferRequestSize()); + batchSent.countDown(); + Thread.currentThread().interrupt(); + } + }; + final IndexedConsensusRequest firstRequest = + new IndexedConsensusRequest(1, Collections.singletonList(new TestEntry(1, localPeer))); + firstRequest.buildSerializedRequests(); + final IndexedConsensusRequest secondRequest = + new IndexedConsensusRequest(2, Collections.singletonList(new TestEntry(2, localPeer))); + secondRequest.buildSerializedRequests(); + assertTrue(dispatcherThread.offer(firstRequest)); + assertTrue(dispatcherThread.offer(secondRequest)); + + 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);
