Fix frozen IoTConsensus sync lag (apache#15023)

* Fix frozen IoTConsensus sync lag

* spotless

Author: jt2594838

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/IWALBuffer.java

@@ -20,6 +20,7 @@
 package org.apache.iotdb.db.storageengine.dataregion.wal.buffer;
 
 import java.util.concurrent.TimeUnit;
+import java.util.function.Predicate;
 
 /**
  * This class serializes and flushes {@link WALEntry}. If search is enabled, the order of search
@@ -52,6 +53,15 @@ public interface IWALBuffer extends AutoCloseable {
    */
   void waitForFlush() throws InterruptedException;
 
+  /**
+   * Wait for next flush operation done, if the predicate == true after entering a locked
+   * environment. Otherwise, return directly.
+   *
+   * @param waitPredicate the condition which should be satisfied before waiting.
+   * @throws InterruptedException when interrupted by the flush thread
+   */
+  public void waitForFlush(Predicate<WALBuffer> waitPredicate) throws InterruptedException;
+
   /**
    * Wait for next flush operation done.
    *

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java

@@ -64,6 +64,7 @@
 import java.util.concurrent.locks.Condition;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Predicate;
 
 import static org.apache.iotdb.db.storageengine.dataregion.wal.node.WALNode.DEFAULT_SEARCH_INDEX;
 
@@ -666,6 +667,18 @@ public void waitForFlush() throws InterruptedException {
     }
   }
 
+  @Override
+  public void waitForFlush(Predicate<WALBuffer> waitPredicate) throws InterruptedException {
+    buffersLock.lock();
+    try {
+      if (waitPredicate.test(this)) {
+        idleBufferReadyCondition.await();
+      }
+    } finally {
+      buffersLock.unlock();
+    }
+  }
+
   @Override
   public boolean waitForFlush(long time, TimeUnit unit) throws InterruptedException {
     buffersLock.lock();

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNode.java

@@ -826,20 +826,24 @@ public IndexedConsensusRequest next() {
     @Override
     public void waitForNextReady() throws InterruptedException {
       boolean walFileRolled = false;
+      long bufferLastSearchIndex = 0;
       while (!hasNext()) {
         if (!walFileRolled) {
           boolean timeout =
               !buffer.waitForFlush(WAIT_FOR_NEXT_WAL_ENTRY_TIMEOUT_IN_SEC, TimeUnit.SECONDS);
           if (timeout) {
+            bufferLastSearchIndex = buffer.getCurrentSearchIndex();
             logger.info(
                 "timeout when waiting for next WAL entry ready, execute rollWALFile. Current search index in wal buffer is {}, and next target index is {}",
-                buffer.getCurrentSearchIndex(),
+                bufferLastSearchIndex,
                 nextSearchIndex);
             rollWALFile();
             walFileRolled = true;
           }
         } else {
-          buffer.waitForFlush();
+          // only wait when the search index of the buffer remains the same as the previous check
+          long finalBufferLastSearchIndex = bufferLastSearchIndex;
+          buffer.waitForFlush(buf -> buf.getCurrentSearchIndex() == finalBufferLastSearchIndex);
         }
       }
     }