[To dev/1.3] Fix IoTConsensus batch accumulation and config reload (#18501)

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

@@ -97,7 +97,7 @@ public class IoTConsensus implements IConsensus {
       new ConcurrentHashMap<>();
   private final IoTConsensusRPCService service;
   private final RegisterManager registerManager = new RegisterManager();
-  private IoTConsensusConfig config;
+  private volatile IoTConsensusConfig config;
   private final IClientManager<TEndPoint, AsyncIoTConsensusServiceClient> clientManager;
   private final IClientManager<TEndPoint, SyncIoTConsensusServiceClient> syncClientManager;
   private final ScheduledExecutorService backgroundTaskService;
@@ -472,6 +472,11 @@ public String getRegionDirFromConsensusGroupId(ConsensusGroupId groupId) {
   public void reloadConsensusConfig(ConsensusConfig consensusConfig) {
     config = consensusConfig.getIotConsensusConfig();
 
+    IoTConsensusMemoryManager.getInstance()
+        .init(
+            config.getReplication().getAllocateMemoryForConsensus(),
+            config.getReplication().getAllocateMemoryForQueue());
+
     for (IoTConsensusServerImpl impl : stateMachineMap.values()) {
       impl.reloadConsensusConfig(config);
     }

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

@@ -116,7 +116,7 @@ public class IoTConsensusServerImpl {
   private final TreeSet<Peer> configuration;
   private final AtomicLong searchIndex;
   private final LogDispatcher logDispatcher;
-  private IoTConsensusConfig config;
+  private volatile IoTConsensusConfig config;
   private final ConsensusReqReader consensusReqReader;
   private volatile boolean active;
   private String newSnapshotDirName;
@@ -911,6 +911,7 @@ public String getConsensusGroupId() {
   /** This method is used for hot reload of IoTConsensusConfig. */
   public void reloadConsensusConfig(IoTConsensusConfig config) {
     this.config = config;
+    logDispatcher.reloadConfig(config);
   }
 
   /**

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

@@ -33,8 +33,8 @@ public class IoTConsensusMemoryManager {
   private final AtomicLong memorySizeInByte = new AtomicLong(0);
   private final AtomicLong queueMemorySizeInByte = new AtomicLong(0);
   private final AtomicLong syncMemorySizeInByte = new AtomicLong(0);
-  private Long maxMemorySizeInByte = Runtime.getRuntime().maxMemory() / 10;
-  private Long maxMemorySizeForQueueInByte = Runtime.getRuntime().maxMemory() / 100 * 6;
+  private volatile long maxMemorySizeInByte = Runtime.getRuntime().maxMemory() / 10;
+  private volatile long maxMemorySizeForQueueInByte = Runtime.getRuntime().maxMemory() / 100 * 6;
 
   private IoTConsensusMemoryManager() {
     MetricService.getInstance().addMetricSet(new IoTConsensusMemoryManagerMetrics(this));

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

@@ -179,6 +179,10 @@ public void checkAndFlushIndex() {
     }
   }
 
+  public synchronized void reloadConfig(IoTConsensusConfig config) {
+    threads.forEach(thread -> thread.reloadConfig(config));
+  }
+
   public void offer(IndexedConsensusRequest request) {
     // we don't need to serialize and offer request when replicaNum is 1.
     if (!threads.isEmpty()) {
@@ -215,7 +219,7 @@ public class LogDispatcherThread implements Runnable {
 
     private static final long PENDING_REQUEST_TAKING_TIME_OUT_IN_SEC = 10;
     private static final long START_INDEX = 1;
-    private final IoTConsensusConfig config;
+    private volatile IoTConsensusConfig config;
     private final Peer peer;
     private final IndexController controller;
     // A sliding window class that manages asynchronous pendingBatches
@@ -273,6 +277,11 @@ public IoTConsensusConfig getConfig() {
       return config;
     }
 
+    private void reloadConfig(IoTConsensusConfig config) {
+      this.config = config;
+      syncStatus.reloadConfig(config);
+    }
+
     public int getPendingEntriesSize() {
       return pendingEntries.size();
     }
@@ -358,11 +367,16 @@ public void run() {
             IndexedConsensusRequest request =
                 pendingEntries.poll(PENDING_REQUEST_TAKING_TIME_OUT_IN_SEC, TimeUnit.SECONDS);
             if (request != null) {
+              final IoTConsensusConfig currentConfig = config;
+              final boolean shouldWaitForBatchAccumulation =
+                  pendingEntries.size()
+                          <= currentConfig.getReplication().getMaxLogEntriesNumPerBatch()
+                      && bufferedEntries.isEmpty();
               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());
+              if (shouldWaitForBatchAccumulation) {
+                waitForBatchAccumulation(
+                    currentConfig.getReplication().getMaxWaitingTimeForAccumulatingBatchInMs());
               }
             }
             // Immediately check for interrupts after poll and sleep
@@ -392,6 +406,10 @@ public void run() {
       logger.info("{}: Dispatcher for {} exits", impl.getThisNode(), peer);
     }
 
+    void waitForBatchAccumulation(long waitingTimeInMs) throws InterruptedException {
+      Thread.sleep(waitingTimeInMs);
+    }
+
     public void updateSafelyDeletedSearchIndex() {
       // update safely deleted search index to delete outdated info,
       // indicating that insert nodes whose search index are before this value can be deleted
@@ -406,6 +424,7 @@ public void updateSafelyDeletedSearchIndex() {
     }
 
     public Batch getBatch() {
+      final IoTConsensusConfig currentConfig = config;
       long startIndex = syncStatus.getNextSendingIndex();
       long maxIndex;
       synchronized (impl.getIndexObject()) {
@@ -420,7 +439,7 @@ public Batch getBatch() {
         // Use drainTo instead of poll to reduce lock overhead
         pendingEntries.drainTo(
             bufferedEntries,
-            config.getReplication().getMaxLogEntriesNumPerBatch() - bufferedEntries.size());
+            currentConfig.getReplication().getMaxLogEntriesNumPerBatch() - bufferedEntries.size());
       }
       // remove all request that searchIndex < startIndex
       Iterator<IndexedConsensusRequest> iterator = bufferedEntries.iterator();
@@ -434,7 +453,7 @@ public Batch getBatch() {
         }
       }
 
-      Batch batches = new Batch(config);
+      Batch batches = new Batch(currentConfig);
       // This condition will be executed in several scenarios:
       // 1. restart
       // 2. The getBatch() is invoked immediately at the moment the PendingEntries are consumed

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

@@ -31,7 +31,7 @@
 public class SyncStatus {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(SyncStatus.class);
-  private final IoTConsensusConfig config;
+  private IoTConsensusConfig config;
   private final IndexController controller;
   private final LinkedList<Batch> pendingBatches = new LinkedList<>();
   private final IoTConsensusMemoryManager iotConsensusMemoryManager =
@@ -42,6 +42,11 @@ public SyncStatus(IndexController controller, IoTConsensusConfig config) {
     this.config = config;
   }
 
+  public synchronized void reloadConfig(IoTConsensusConfig config) {
+    this.config = config;
+    notifyAll();
+  }
+
   /**
    * we may block here if the synchronization pipeline is full.
    *

iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/logdispatcher/IoTConsensusMemoryManagerTest.java

@@ -37,6 +37,20 @@
 
 public class IoTConsensusMemoryManagerTest {
 
+  @Test
+  public void testInitUpdatesMemoryLimits() {
+    IoTConsensusMemoryManager memoryManager = IoTConsensusMemoryManager.getInstance();
+    long previousMaxMemory = memoryManager.getMaxMemorySizeInByte();
+    long previousMaxQueueMemory = memoryManager.getMaxMemorySizeForQueueInByte();
+    try {
+      memoryManager.init(1024, 512);
+      assertEquals(1024L, memoryManager.getMaxMemorySizeInByte().longValue());
+      assertEquals(512L, memoryManager.getMaxMemorySizeForQueueInByte().longValue());
+    } finally {
+      memoryManager.init(previousMaxMemory, previousMaxQueueMemory);
+    }
+  }
+
   @Test
   public void testAllocateQueue() {
     IoTConsensusMemoryManager memoryManager = IoTConsensusMemoryManager.getInstance();