This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new c41838e1386 Fix IoTConsensus batch accumulation by byte size (#18560)
c41838e1386 is described below
commit c41838e1386706003125f333ddadb25f6132433d
Author: Caideyipi <[email protected]>
AuthorDate: Mon Sep 7 10:11:12 2026 +0800
Fix IoTConsensus batch accumulation by byte size (#18560)
---
.../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 55569b8a34f..26fcddb5d2a 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
@@ -69,6 +69,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.
@@ -77,7 +81,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 1ff6579e0de..d9b34b9f79d 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);
