From 32a939f69ea1cac4e7310289340f66f154159f47 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Thu, 5 Nov 2026 16:47:59 +0800 Subject: [PATCH 1/2] Fix WAL roll file test flush wait (#17904) (cherry picked from commit ad94d2b68cd3cc2ef4d23df18aebe600a1caad24) --- .../wal/node/WALNodeWaitForRollFileTest.java | 58 ++++++++++--------- 1 file changed, 31 insertions(+), 27 deletions(-) diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNodeWaitForRollFileTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNodeWaitForRollFileTest.java index eec2e33e81cee..5bbb277eeef76 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNodeWaitForRollFileTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNodeWaitForRollFileTest.java @@ -31,6 +31,8 @@ import org.apache.iotdb.db.storageengine.dataregion.memtable.IMemTable; import org.apache.iotdb.db.storageengine.dataregion.memtable.PrimitiveMemTable; import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALMode; +import org.apache.iotdb.db.storageengine.dataregion.wal.utils.listener.AbstractResultListener.Status; +import org.apache.iotdb.db.storageengine.dataregion.wal.utils.listener.WALFlushListener; import org.apache.iotdb.db.utils.EnvironmentUtils; import org.apache.iotdb.db.utils.constant.TestConstant; @@ -40,7 +42,6 @@ import org.apache.tsfile.utils.Binary; import org.apache.tsfile.utils.BitMap; import org.apache.tsfile.write.schema.MeasurementSchema; -import org.awaitility.Awaitility; import org.junit.After; import org.junit.Before; import org.junit.Test; @@ -53,6 +54,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; @@ -122,9 +124,8 @@ public void testWaitForNextReadyNotWokenByFlushWithoutRoll() throws Exception { // write a small amount of data (not enough to trigger roll) InsertTabletNode insertTabletNode = getInsertTabletNode(devicePath, new long[] {1}); insertTabletNode.setSearchIndex(1); - walNode.log(memTable.getMemTableId(), insertTabletNode, 0, insertTabletNode.getRowCount()); - - Awaitility.await().atMost(10, TimeUnit.SECONDS).until(() -> walNode.isAllWALEntriesConsumed()); + waitForFlush( + walNode.log(memTable.getMemTableId(), insertTabletNode, 0, insertTabletNode.getRowCount())); // data is flushed to buffer but no WAL file roll happened yet, iterator at search index 1 // should not find data (because the current-writing WAL file is not readable by the iterator) @@ -155,14 +156,12 @@ public void testWaitForNextReadySucceedsAfterRollFile() throws Exception { for (int i = 1; i <= 5; i++) { InsertTabletNode insertTabletNode = getInsertTabletNode(devicePath, new long[] {i}); insertTabletNode.setSearchIndex(i); - walNode.log(memTable.getMemTableId(), insertTabletNode, 0, insertTabletNode.getRowCount()); + waitForFlush( + walNode.log(memTable.getMemTableId(), insertTabletNode, 0, insertTabletNode.getRowCount())); } - Awaitility.await().atMost(10, TimeUnit.SECONDS).until(() -> walNode.isAllWALEntriesConsumed()); - // roll the WAL file so the data is in a closed file readable by the iterator walNode.rollWALFile(); - Awaitility.await().atMost(10, TimeUnit.SECONDS).until(() -> walNode.isAllWALEntriesConsumed()); // iterator at search index 1 should find the data after roll ConsensusReqReader.ReqIterator iterator = walNode.getReqIterator(1); @@ -182,11 +181,11 @@ public void testWaitForNextReadyWakesUpOnConcurrentRoll() throws Exception { // write data with search index InsertTabletNode insertTabletNode = getInsertTabletNode(devicePath, new long[] {1}); insertTabletNode.setSearchIndex(1); - walNode.log(memTable.getMemTableId(), insertTabletNode, 0, insertTabletNode.getRowCount()); - walNode.log( - memTable.getMemTableId(), new ContinuousSameSearchIndexSeparatorNode(new PlanNodeId(""))); - - Awaitility.await().atMost(10, TimeUnit.SECONDS).until(() -> walNode.isAllWALEntriesConsumed()); + waitForFlush( + walNode.log(memTable.getMemTableId(), insertTabletNode, 0, insertTabletNode.getRowCount())); + waitForFlush( + walNode.log( + memTable.getMemTableId(), new ContinuousSameSearchIndexSeparatorNode(new PlanNodeId("")))); ConsensusReqReader.ReqIterator iterator = walNode.getReqIterator(1); @@ -206,9 +205,6 @@ public void testWaitForNextReadyWakesUpOnConcurrentRoll() throws Exception { // trigger WAL file roll — this should signal rollLogWriterCondition and wake up the iterator walNode.rollWALFile(); - Awaitility.await() - .atMost(10, TimeUnit.SECONDS) - .until(() -> walNode.isAllWALEntriesConsumed()); assertTrue( "Iterator should have found data after WAL file roll", @@ -237,11 +233,7 @@ public void testWaitForNextReadyWithAutoRollOnSizeThreshold() throws Exception { // write initial data with search index InsertTabletNode first = getInsertTabletNode(devicePath, new long[] {1}); first.setSearchIndex(1); - walNode.log(memTable.getMemTableId(), first, 0, first.getRowCount()); - - Awaitility.await() - .atMost(10, TimeUnit.SECONDS) - .until(() -> walNode.isAllWALEntriesConsumed()); + waitForFlush(walNode.log(memTable.getMemTableId(), first, 0, first.getRowCount())); ConsensusReqReader.ReqIterator iterator = walNode.getReqIterator(1); @@ -256,11 +248,13 @@ public void testWaitForNextReadyWithAutoRollOnSizeThreshold() throws Exception { assertTrue(waiterStarted.await(10, TimeUnit.SECONDS)); // write more data to exceed the small threshold and trigger auto-roll + WALFlushListener lastFlushListener = null; for (int i = 2; i <= 50; i++) { InsertTabletNode node = getInsertTabletNode(devicePath, new long[] {i}); node.setSearchIndex(i); - walNode.log(memTable.getMemTableId(), node, 0, node.getRowCount()); + lastFlushListener = walNode.log(memTable.getMemTableId(), node, 0, node.getRowCount()); } + waitForFlush(lastFlushListener); assertTrue( "Iterator should have found data after auto WAL file roll", @@ -288,11 +282,11 @@ public void testWaitForNextReadyAutoTriggersRollOnTimeout() throws Exception { // write data with search index — stays in the current (active) WAL file InsertTabletNode insertTabletNode = getInsertTabletNode(devicePath, new long[] {1}); insertTabletNode.setSearchIndex(1); - walNode.log(memTable.getMemTableId(), insertTabletNode, 0, insertTabletNode.getRowCount()); - walNode.log( - memTable.getMemTableId(), new ContinuousSameSearchIndexSeparatorNode(new PlanNodeId(""))); - - Awaitility.await().atMost(10, TimeUnit.SECONDS).until(() -> walNode.isAllWALEntriesConsumed()); + waitForFlush( + walNode.log(memTable.getMemTableId(), insertTabletNode, 0, insertTabletNode.getRowCount())); + waitForFlush( + walNode.log( + memTable.getMemTableId(), new ContinuousSameSearchIndexSeparatorNode(new PlanNodeId("")))); // iterator cannot read the active WAL file, so hasNext() should be false ConsensusReqReader.ReqIterator iterator = walNode.getReqIterator(1); @@ -334,6 +328,16 @@ private static void shutdownExecutor(final ExecutorService executor) throws Inte assertTrue(executor.awaitTermination(10, TimeUnit.SECONDS)); } + private void waitForFlush(WALFlushListener flushListener) throws Exception { + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + Future waitFuture = executor.submit(flushListener::waitForResult); + assertEquals(Status.SUCCESS, waitFuture.get(10, TimeUnit.SECONDS)); + } finally { + executor.shutdownNow(); + } + } + private InsertTabletNode getInsertTabletNode(String devicePath, long[] times) throws IllegalPathException { String[] measurements = new String[] {"s1", "s2", "s3", "s4", "s5", "s6"}; From bf4bad4e25e86285bc57df2cf154e080ddf1bd4c Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 6 Nov 2026 15:03:14 +0800 Subject: [PATCH 2/2] Update WALNodeWaitForRollFileTest.java --- .../dataregion/wal/node/WALNodeWaitForRollFileTest.java | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNodeWaitForRollFileTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNodeWaitForRollFileTest.java index 5bbb277eeef76..3e2e1e76d4887 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNodeWaitForRollFileTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNodeWaitForRollFileTest.java @@ -157,7 +157,8 @@ public void testWaitForNextReadySucceedsAfterRollFile() throws Exception { InsertTabletNode insertTabletNode = getInsertTabletNode(devicePath, new long[] {i}); insertTabletNode.setSearchIndex(i); waitForFlush( - walNode.log(memTable.getMemTableId(), insertTabletNode, 0, insertTabletNode.getRowCount())); + walNode.log( + memTable.getMemTableId(), insertTabletNode, 0, insertTabletNode.getRowCount())); } // roll the WAL file so the data is in a closed file readable by the iterator @@ -185,7 +186,8 @@ public void testWaitForNextReadyWakesUpOnConcurrentRoll() throws Exception { walNode.log(memTable.getMemTableId(), insertTabletNode, 0, insertTabletNode.getRowCount())); waitForFlush( walNode.log( - memTable.getMemTableId(), new ContinuousSameSearchIndexSeparatorNode(new PlanNodeId("")))); + memTable.getMemTableId(), + new ContinuousSameSearchIndexSeparatorNode(new PlanNodeId("")))); ConsensusReqReader.ReqIterator iterator = walNode.getReqIterator(1); @@ -286,7 +288,8 @@ public void testWaitForNextReadyAutoTriggersRollOnTimeout() throws Exception { walNode.log(memTable.getMemTableId(), insertTabletNode, 0, insertTabletNode.getRowCount())); waitForFlush( walNode.log( - memTable.getMemTableId(), new ContinuousSameSearchIndexSeparatorNode(new PlanNodeId("")))); + memTable.getMemTableId(), + new ContinuousSameSearchIndexSeparatorNode(new PlanNodeId("")))); // iterator cannot read the active WAL file, so hasNext() should be false ConsensusReqReader.ReqIterator iterator = walNode.getReqIterator(1);