[To rel/1.2][IOTDB-6022] Modify the default parameters of iotconsensus (#10325)

iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/config/IoTConsensusConfig.java

@@ -256,6 +256,7 @@ public class IoTConsensusConfig {
 
     private final int maxQueueLength;
     private final long maxWaitingTimeForWaitBatchInMs;
+    private final int maxWaitingTimeForAccumulatingBatchInMs;
     private final long basicRetryWaitTimeMs;
     private final long maxRetryWaitTimeMs;
     private final long walThrottleThreshold;
@@ -270,6 +271,7 @@ public class IoTConsensusConfig {
         int maxPendingBatchesNum,
         int maxQueueLength,
         long maxWaitingTimeForWaitBatchInMs,
+        int maxWaitingTimeForAccumulatingBatchInMs,
         long basicRetryWaitTimeMs,
         long maxRetryWaitTimeMs,
         long walThrottleThreshold,
@@ -282,6 +284,7 @@ public class IoTConsensusConfig {
       this.maxPendingBatchesNum = maxPendingBatchesNum;
       this.maxQueueLength = maxQueueLength;
       this.maxWaitingTimeForWaitBatchInMs = maxWaitingTimeForWaitBatchInMs;
+      this.maxWaitingTimeForAccumulatingBatchInMs = maxWaitingTimeForAccumulatingBatchInMs;
       this.basicRetryWaitTimeMs = basicRetryWaitTimeMs;
       this.maxRetryWaitTimeMs = maxRetryWaitTimeMs;
       this.walThrottleThreshold = walThrottleThreshold;
@@ -311,6 +314,10 @@ public class IoTConsensusConfig {
       return maxWaitingTimeForWaitBatchInMs;
     }
 
+    public int getMaxWaitingTimeForAccumulatingBatchInMs() {
+      return maxWaitingTimeForAccumulatingBatchInMs;
+    }
+
     public long getBasicRetryWaitTimeMs() {
       return basicRetryWaitTimeMs;
     }
@@ -347,9 +354,11 @@ public class IoTConsensusConfig {
 
       private int maxLogEntriesNumPerBatch = 1024;
       private int maxSizePerBatch = 16 * 1024 * 1024;
-      private int maxPendingBatchesNum = 12;
+      private int maxPendingBatchesNum = 5;
       private int maxQueueLength = 4096;
       private long maxWaitingTimeForWaitBatchInMs = 10 * 1000L;
+
+      private int maxWaitingTimeForAccumulatingBatchInMs = 500;
       private long basicRetryWaitTimeMs = TimeUnit.MILLISECONDS.toMillis(100);
       private long maxRetryWaitTimeMs = TimeUnit.SECONDS.toMillis(20);
       private long walThrottleThreshold = 50 * 1024 * 1024 * 1024L;
@@ -384,6 +393,12 @@ public class IoTConsensusConfig {
         return this;
       }
 
+      public Builder setMaxWaitingTimeForAccumulatingBatchInMs(
+          int maxWaitingTimeForAccumulatingBatchInMs) {
+        this.maxWaitingTimeForAccumulatingBatchInMs = maxWaitingTimeForAccumulatingBatchInMs;
+        return this;
+      }
+
       public Replication.Builder setBasicRetryWaitTimeMs(long basicRetryWaitTimeMs) {
         this.basicRetryWaitTimeMs = basicRetryWaitTimeMs;
         return this;
@@ -426,6 +441,7 @@ public class IoTConsensusConfig {
             maxPendingBatchesNum,
             maxQueueLength,
             maxWaitingTimeForWaitBatchInMs,
+            maxWaitingTimeForAccumulatingBatchInMs,
             basicRetryWaitTimeMs,
             maxRetryWaitTimeMs,
             walThrottleThreshold,

iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/logdispatcher/LogDispatcher.java

@@ -314,6 +314,11 @@ public class LogDispatcher {
                 pendingEntries.poll(PENDING_REQUEST_TAKING_TIME_OUT_IN_SEC, TimeUnit.SECONDS);
             if (request != null) {
               bufferedEntries.add(request);
+              // If write pressure is low, we simply sleep a little to reduce the number of RPC
+              if (pendingEntries.size() <= config.getReplication().getMaxLogEntriesNumPerBatch()
+                  && bufferedEntries.isEmpty()) {
+                Thread.sleep(config.getReplication().getMaxWaitingTimeForAccumulatingBatchInMs());
+              }
             }
           }
           logDispatcherThreadMetrics.recordConstructBatchTime(

iotdb-core/node-commons/src/assembly/resources/conf/iotdb-common.properties

@@ -765,7 +765,7 @@ cluster_name=defaultCluster
 
 # The maximum pending batches num in IoTConsensus
 # Datatype: int
-# data_region_iot_max_pending_batches_num = 12
+# data_region_iot_max_pending_batches_num = 5
 
 # The maximum memory ratio for queue in IoTConsensus
 # Datatype: double

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/ThreadName.java

@@ -18,6 +18,9 @@
  */
 package org.apache.iotdb.commons.concurrent;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import java.util.Arrays;
 import java.util.HashSet;
 import java.util.Set;
@@ -133,6 +136,7 @@ public enum ThreadName {
   SYSTEM_SCHEDULE_METRICS("SystemScheduleMetrics"),
   RESOURCE_CONTROL_DISK_STATISTIC("ResourceControl-DataRegionDiskStatistics"),
   PROMETHEUS_REACTOR_HTTP_NIO("reactor-http-nio"),
+  PROMETHEUS_REACTOR_HTTP_EPOLL("reactor-http-epoll"),
   PROMETHEUS_BOUNDED_ELASTIC("boundedElastic-evictor"),
   // -------------------------- Other --------------------------
   TTL_CHECK("TTL-CHECK"),
@@ -147,6 +151,7 @@ public enum ThreadName {
   UNKOWN("UNKNOWN");
 
   private final String name;
+  private static final Logger log = LoggerFactory.getLogger(ThreadName.class);
   private static Set<ThreadName> queryThreadNames =
       new HashSet<>(
           Arrays.asList(
@@ -258,6 +263,7 @@ public enum ThreadName {
               SYSTEM_SCHEDULE_METRICS,
               RESOURCE_CONTROL_DISK_STATISTIC,
               PROMETHEUS_REACTOR_HTTP_NIO,
+              PROMETHEUS_REACTOR_HTTP_EPOLL,
               PROMETHEUS_BOUNDED_ELASTIC));
   private static Set<ThreadName> otherThreadNames =
       new HashSet<>(
@@ -352,6 +358,7 @@ public enum ThreadName {
         }
       }
     }
+    log.debug("Unknown thread name: {}", givenThreadName);
     return ThreadName.UNKOWN;
   }
 }