Skip to content

[To dev/1.3] Fix flaky concurrent queue and WAL tests (#18246) #18255

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jt2594838 merged 1 commit into from
Jul 21, 2026
+203 −176
Merged

[To dev/1.3] Fix flaky concurrent queue and WAL tests (#18246) #18255

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
141 changes: 62 additions & 79 deletions ...ava/org/apache/iotdb/db/storageengine/dataregion/wal/node/WALNodeWaitForRollFileTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,13 +46,12 @@
import org.junit.Test;

import java.io.File;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
Expand Down Expand Up @@ -175,7 +174,7 @@ public void testWaitForNextReadySucceedsAfterRollFile() throws Exception {
* Verifies that waitForNextReady wakes up when a WAL file roll is triggered concurrently. A
* background thread rolls the WAL file while the main thread waits on the iterator.
*/
@Test(timeout = 30000)
@Test
public void testWaitForNextReadyWakesUpOnConcurrentRoll() throws Exception {
IMemTable memTable = new PrimitiveMemTable(databasePath, dataRegionId);
walNode.onMemTableCreated(memTable, logDirectory + File.separator + "test.tsfile");
Expand All @@ -191,49 +190,45 @@ public void testWaitForNextReadyWakesUpOnConcurrentRoll() throws Exception {

ConsensusReqReader.ReqIterator iterator = walNode.getReqIterator(1);

AtomicBoolean found = new AtomicBoolean(false);
AtomicReference<Exception> error = new AtomicReference<>();
ExecutorService executor = Executors.newSingleThreadExecutor();

// background: wait for data to become available via waitForNextReady
Future<?> waitFuture =
executor.submit(
() -> {
try {
CountDownLatch waiterStarted = new CountDownLatch(1);
try {
// background: wait for data to become available via waitForNextReady
Future<Boolean> waitFuture =
executor.submit(
() -> {
waiterStarted.countDown();
iterator.waitForNextReady(15, TimeUnit.SECONDS);
if (iterator.hasNext()) {
found.set(true);
}
} catch (Exception e) {
error.set(e);
}
});

// give the waiter thread time to start blocking
Thread.sleep(500);

// trigger WAL file roll — this should signal rollLogWriterCondition and wake up the iterator
walNode.rollWALFile();
Awaitility.await().atMost(10, TimeUnit.SECONDS).until(() -> walNode.isAllWALEntriesConsumed());
return iterator.hasNext();
});

waitFuture.get(20, TimeUnit.SECONDS);
executor.shutdown();
assertTrue(waiterStarted.await(10, TimeUnit.SECONDS));

if (error.get() != null) {
throw error.get();
// 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",
waitFuture.get(20, TimeUnit.SECONDS));
} finally {
shutdownExecutor(executor);
}
assertTrue("Iterator should have found data after WAL file roll", found.get());
}

/**
* Verifies that the no-arg waitForNextReady eventually proceeds when enough data is written to
* trigger an automatic WAL file roll (file size exceeds threshold). Uses a small WAL file size
* threshold to trigger the roll quickly.
*/
@Test(timeout = 60000)
@Test
public void testWaitForNextReadyWithAutoRollOnSizeThreshold() throws Exception {
// use small WAL file size to trigger auto-roll
config.setWalFileSizeThresholdInByte(1024);
ExecutorService executor = Executors.newSingleThreadExecutor();
CountDownLatch waiterStarted = new CountDownLatch(1);

try {
IMemTable memTable = new PrimitiveMemTable(databasePath, dataRegionId);
Expand All @@ -250,24 +245,15 @@ public void testWaitForNextReadyWithAutoRollOnSizeThreshold() throws Exception {

ConsensusReqReader.ReqIterator iterator = walNode.getReqIterator(1);

AtomicBoolean found = new AtomicBoolean(false);
AtomicReference<Exception> error = new AtomicReference<>();
ExecutorService executor = Executors.newSingleThreadExecutor();

Future<?> waitFuture =
Future<Boolean> waitFuture =
executor.submit(
() -> {
try {
iterator.waitForNextReady(30, TimeUnit.SECONDS);
if (iterator.hasNext()) {
found.set(true);
}
} catch (Exception e) {
error.set(e);
}
waiterStarted.countDown();
iterator.waitForNextReady(30, TimeUnit.SECONDS);
return iterator.hasNext();
});

Thread.sleep(500);
assertTrue(waiterStarted.await(10, TimeUnit.SECONDS));

// write more data to exceed the small threshold and trigger auto-roll
for (int i = 2; i <= 50; i++) {
Expand All @@ -276,15 +262,15 @@ public void testWaitForNextReadyWithAutoRollOnSizeThreshold() throws Exception {
walNode.log(memTable.getMemTableId(), node, 0, node.getRowCount());
}

waitFuture.get(40, TimeUnit.SECONDS);
executor.shutdown();

if (error.get() != null) {
fail("waitForNextReady threw unexpected exception: " + error.get().getMessage());
}
assertTrue("Iterator should have found data after auto WAL file roll", found.get());
assertTrue(
"Iterator should have found data after auto WAL file roll",
waitFuture.get(40, TimeUnit.SECONDS));
} finally {
config.setWalFileSizeThresholdInByte(2 * 1024 * 1024);
try {
shutdownExecutor(executor);
} finally {
config.setWalFileSizeThresholdInByte(2 * 1024 * 1024);
}
}
}

Expand All @@ -294,7 +280,7 @@ public void testWaitForNextReadyWithAutoRollOnSizeThreshold() throws Exception {
* buffer → waitForRollFile(30s) times out → rollWALFile() called → data moves to closed file →
* hasNext() returns true → method returns.
*/
@Test(timeout = 120000)
@Test
public void testWaitForNextReadyAutoTriggersRollOnTimeout() throws Exception {
IMemTable memTable = new PrimitiveMemTable(databasePath, dataRegionId);
walNode.onMemTableCreated(memTable, logDirectory + File.separator + "test.tsfile");
Expand All @@ -312,8 +298,6 @@ public void testWaitForNextReadyAutoTriggersRollOnTimeout() throws Exception {
ConsensusReqReader.ReqIterator iterator = walNode.getReqIterator(1);
assertFalse("Data should not be visible before WAL file roll", iterator.hasNext());

AtomicBoolean found = new AtomicBoolean(false);
AtomicReference<Exception> error = new AtomicReference<>();
ExecutorService executor = Executors.newSingleThreadExecutor();

long startTime = System.currentTimeMillis();
Expand All @@ -322,33 +306,32 @@ public void testWaitForNextReadyAutoTriggersRollOnTimeout() throws Exception {
// 1) wait 30s for rollLogWriterCondition (timeout)
// 2) auto-call rollWALFile()
// 3) data becomes readable, hasNext() returns true, method returns
Future<?> waitFuture =
executor.submit(
() -> {
try {
try {
Future<Boolean> waitFuture =
executor.submit(
() -> {
iterator.waitForNextReady();
if (iterator.hasNext()) {
found.set(true);
}
} catch (Exception e) {
error.set(e);
}
});

waitFuture.get(90, TimeUnit.SECONDS);
executor.shutdown();
return iterator.hasNext();
});

long elapsed = System.currentTimeMillis() - startTime;
assertTrue(
"Iterator should have found data after auto-triggered WAL file roll",
waitFuture.get(90, TimeUnit.SECONDS));

if (error.get() != null) {
fail("waitForNextReady() threw unexpected exception: " + error.get().getMessage());
long elapsed = System.currentTimeMillis() - startTime;
assertTrue(
"Should have waited at least 30s for the timeout to trigger auto-roll, but only waited "
+ elapsed
+ "ms",
elapsed >= TimeUnit.SECONDS.toMillis(WALNode.WAIT_FOR_NEXT_WAL_ENTRY_TIMEOUT_IN_SEC - 1));
} finally {
shutdownExecutor(executor);
}
assertTrue("Iterator should have found data after auto-triggered WAL file roll", found.get());
assertTrue(
"Should have waited at least 30s for the timeout to trigger auto-roll, but only waited "
+ elapsed
+ "ms",
elapsed >= TimeUnit.SECONDS.toMillis(WALNode.WAIT_FOR_NEXT_WAL_ENTRY_TIMEOUT_IN_SEC - 1));
}

private static void shutdownExecutor(final ExecutorService executor) throws InterruptedException {
executor.shutdownNow();
assertTrue(executor.awaitTermination(10, TimeUnit.SECONDS));
}

private InsertTabletNode getInsertTabletNode(String devicePath, long[] times)
Expand Down
Loading
Loading