Fix that WALBuffer waits for flush instead of file-roll (#17628)

Author: jt2594838

.idea/icon.png

@@ -51,7 +51,7 @@ public interface IWALBuffer extends AutoCloseable {
    *
    * @throws InterruptedException when interrupted by the flush thread
    */
-  void waitForFlush() throws InterruptedException;
+  void waitForRollFile() throws InterruptedException;
 
   /**
    * Wait for next flush operation done, if the predicate == true after entering a locked
@@ -60,14 +60,14 @@ public interface IWALBuffer extends AutoCloseable {
    * @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;
+  public void waitForRollFile(Predicate<WALBuffer> waitPredicate) throws InterruptedException;
 
   /**
    * Wait for next flush operation done.
    *
    * @throws InterruptedException when interrupted by the flush thread
    */
-  boolean waitForFlush(long time, TimeUnit unit) throws InterruptedException;
+  boolean waitForRollFile(long time, TimeUnit unit) throws InterruptedException;
 
   /** Return true when all wal entries all consumed and flushed. */
   boolean isAllWALEntriesConsumed();

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

@@ -90,6 +90,7 @@ public class WALBuffer extends AbstractWALBuffer {
   private final Lock buffersLock = new ReentrantLock();
   // condition to guarantee correctness of switching buffers
   private final Condition idleBufferReadyCondition = buffersLock.newCondition();
+  private final Condition rollLogWriterCondition = buffersLock.newCondition();
   // last writer position when fsync is called, help record each entry's position
   private long lastFsyncPosition;
 
@@ -170,6 +171,13 @@ private int getCompressedByteBufferSize(int size) {
   protected File rollLogWriter(long searchIndex, WALFileStatus fileStatus) throws IOException {
     File file = super.rollLogWriter(searchIndex, fileStatus);
     currentWALFileWriter.setCompressedByteBuffer(compressedByteBuffer);
+    buffersLock.lock();
+    try {
+      // notify WALReader that new file is generated, and it can read new file
+      rollLogWriterCondition.signalAll();
+    } finally {
+      buffersLock.unlock();
+    }
     return file;
   }
 
@@ -656,7 +664,7 @@ private void switchSyncingBufferToIdle() {
   }
 
   @Override
-  public void waitForFlush() throws InterruptedException {
+  public void waitForRollFile() throws InterruptedException {
     buffersLock.lock();
     try {
       idleBufferReadyCondition.await();
@@ -666,22 +674,22 @@ public void waitForFlush() throws InterruptedException {
   }
 
   @Override
-  public void waitForFlush(Predicate<WALBuffer> waitPredicate) throws InterruptedException {
+  public void waitForRollFile(Predicate<WALBuffer> waitPredicate) throws InterruptedException {
     buffersLock.lock();
     try {
       if (waitPredicate.test(this)) {
-        idleBufferReadyCondition.await();
+        rollLogWriterCondition.await();
       }
     } finally {
       buffersLock.unlock();
     }
   }
 
   @Override
-  public boolean waitForFlush(long time, TimeUnit unit) throws InterruptedException {
+  public boolean waitForRollFile(long time, TimeUnit unit) throws InterruptedException {
     buffersLock.lock();
     try {
-      return idleBufferReadyCondition.await(time, unit);
+      return rollLogWriterCondition.await(time, unit);
     } finally {
       buffersLock.unlock();
     }

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

@@ -96,7 +96,7 @@ public class WALNode implements IWALNode {
   // no iot consensus, all insert nodes can be safely deleted
   public static final long DEFAULT_SAFELY_DELETED_SEARCH_INDEX = Long.MAX_VALUE;
   // timeout threshold when waiting for next wal entry
-  private static final long WAIT_FOR_NEXT_WAL_ENTRY_TIMEOUT_IN_SEC = 30;
+  public static final long WAIT_FOR_NEXT_WAL_ENTRY_TIMEOUT_IN_SEC = 30;
   private static final WritingMetrics WRITING_METRICS = WritingMetrics.getInstance();
 
   // unique identifier of this WALNode
@@ -792,7 +792,7 @@ public void waitForNextReady() throws InterruptedException {
       while (!hasNext()) {
         if (!walFileRolled) {
           boolean timeout =
-              !buffer.waitForFlush(WAIT_FOR_NEXT_WAL_ENTRY_TIMEOUT_IN_SEC, TimeUnit.SECONDS);
+              !buffer.waitForRollFile(WAIT_FOR_NEXT_WAL_ENTRY_TIMEOUT_IN_SEC, TimeUnit.SECONDS);
           if (timeout) {
             bufferLastSearchIndex = buffer.getCurrentSearchIndex();
             logger.info(
@@ -805,7 +805,7 @@ public void waitForNextReady() throws InterruptedException {
         } else {
           // 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);
+          buffer.waitForRollFile(buf -> buf.getCurrentSearchIndex() == finalBufferLastSearchIndex);
         }
       }
     }
@@ -814,8 +814,8 @@ public void waitForNextReady() throws InterruptedException {
     public void waitForNextReady(long time, TimeUnit unit)
         throws InterruptedException, TimeoutException {
       if (!hasNext()) {
-        boolean timeout = !buffer.waitForFlush(time, unit);
-        if (timeout || !hasNext()) {
+        boolean timeout = !buffer.waitForRollFile(time, unit);
+        if (timeout && !hasNext()) {
           throw new TimeoutException();
         }
       }