From 03920bec5100ff238142c51bd311e76110a270b1 Mon Sep 17 00:00:00 2001 From: shuwenwei Date: Thu, 20 Aug 2026 10:28:04 +0800 Subject: [PATCH] fix: make CompactionWorkerTest UT deterministic by using CountDownLatch instead of fixed 2s sleep to wait for task drop status reset The two failing UTs (testFailedToAllocateFileNumInCrossTask and testFailedToCheckValidInCrossTask) were flaky because they used thread.join(2s) to wait for the task to be dropped. However, in these cases queue.take() never returns - the task is dropped by dropCompactionTask which calls resetCompactionCandidateStatusForAllSourceFiles(). The 2s sleep was not reliable under load. Change to CountDownLatch attached to the resetCompactionCandidateStatusForAllSourceFiles() hook, with 5s timeout. This makes the test wait for the exact condition (status back to NORMAL) instead of arbitrary time. --- .../compaction/CompactionWorkerTest.java | 58 +++++++++++++++++-- 1 file changed, 52 insertions(+), 6 deletions(-) diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/CompactionWorkerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/CompactionWorkerTest.java index 11121d5bc015c..ec1ec976e2b29 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/CompactionWorkerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/CompactionWorkerTest.java @@ -95,6 +95,15 @@ public void testFailedToAllocateMemoryInCrossTask() throws Exception { 0); CrossSpaceCompactionTask taskMock = Mockito.spy(task); Mockito.doReturn(true).when(taskMock).start(); + CountDownLatch statusResetLatch = new CountDownLatch(1); + Mockito.doAnswer( + invocation -> { + invocation.callRealMethod(); + statusResetLatch.countDown(); + return null; + }) + .when(taskMock) + .resetCompactionCandidateStatusForAllSourceFiles(); FixedPriorityBlockingQueue queue = new CompactionTaskQueue(50, new DefaultCompactionTaskComparatorImpl()); queue.put(taskMock); @@ -108,7 +117,9 @@ public void testFailedToAllocateMemoryInCrossTask() throws Exception { } }); thread.start(); - thread.join(TimeUnit.SECONDS.toMillis(2)); + Assert.assertTrue( + "source files should be reset to NORMAL after the cross-space task is dropped", + statusResetLatch.await(5, TimeUnit.SECONDS)); Assert.assertEquals( 0, SystemInfo.getInstance().getCompactionMemoryBlock().getUsedMemoryInBytes()); Assert.assertEquals(0, SystemInfo.getInstance().getCompactionFileNumCost().get()); @@ -153,6 +164,15 @@ public void testFailedToAllocateFileNumInCrossTask() throws InterruptedException 0L, tsFileManager, sequenceFiles, unsequenceFiles, null, 1000, 0); CrossSpaceCompactionTask taskMock = Mockito.spy(task); Mockito.doReturn(true).when(taskMock).start(); + CountDownLatch statusResetLatch = new CountDownLatch(1); + Mockito.doAnswer( + invocation -> { + invocation.callRealMethod(); + statusResetLatch.countDown(); + return null; + }) + .when(taskMock) + .resetCompactionCandidateStatusForAllSourceFiles(); FixedPriorityBlockingQueue queue = new CompactionTaskQueue(50, new DefaultCompactionTaskComparatorImpl()); queue.put(taskMock); @@ -165,7 +185,9 @@ public void testFailedToAllocateFileNumInCrossTask() throws InterruptedException } }); thread.start(); - thread.join(TimeUnit.SECONDS.toMillis(2)); + Assert.assertTrue( + "source files should be reset to NORMAL after the cross-space task is dropped", + statusResetLatch.await(5, TimeUnit.SECONDS)); Assert.assertEquals( 0, SystemInfo.getInstance().getCompactionMemoryBlock().getUsedMemoryInBytes()); Assert.assertEquals(0, SystemInfo.getInstance().getCompactionFileNumCost().get()); @@ -208,9 +230,19 @@ public void testFailedToCheckValidInCrossTask() throws InterruptedException { CrossSpaceCompactionTask task = new CrossSpaceCompactionTask( 0L, tsFileManager, sequenceFiles, unsequenceFiles, null, 1000, 0); + CrossSpaceCompactionTask taskMock = Mockito.spy(task); + CountDownLatch statusResetLatch = new CountDownLatch(1); + Mockito.doAnswer( + invocation -> { + invocation.callRealMethod(); + statusResetLatch.countDown(); + return null; + }) + .when(taskMock) + .resetCompactionCandidateStatusForAllSourceFiles(); FixedPriorityBlockingQueue queue = new CompactionTaskQueue(50, new DefaultCompactionTaskComparatorImpl()); - queue.put(task); + queue.put(taskMock); Thread thread = new Thread( () -> { @@ -220,7 +252,9 @@ public void testFailedToCheckValidInCrossTask() throws InterruptedException { } }); thread.start(); - thread.join(TimeUnit.SECONDS.toMillis(2)); + Assert.assertTrue( + "source files should be reset to NORMAL after the cross-space task is dropped", + statusResetLatch.await(5, TimeUnit.SECONDS)); Assert.assertEquals( 0, SystemInfo.getInstance().getCompactionMemoryBlock().getUsedMemoryInBytes()); Assert.assertEquals(0, SystemInfo.getInstance().getCompactionFileNumCost().get()); @@ -248,9 +282,19 @@ public void testFailedToCheckValidInInnerTask() throws InterruptedException { // fail to check valid when tsfile manager is not allowed to compaction in inner task InnerSpaceCompactionTask innerTask = new InnerSpaceCompactionTask(0L, tsFileManager, sequenceFiles, true, null, 0L); + InnerSpaceCompactionTask innerTaskMock = Mockito.spy(innerTask); + CountDownLatch statusResetLatch = new CountDownLatch(1); + Mockito.doAnswer( + invocation -> { + invocation.callRealMethod(); + statusResetLatch.countDown(); + return null; + }) + .when(innerTaskMock) + .resetCompactionCandidateStatusForAllSourceFiles(); FixedPriorityBlockingQueue queue = new CompactionTaskQueue(50, new DefaultCompactionTaskComparatorImpl()); - queue.put(innerTask); + queue.put(innerTaskMock); Thread thread = new Thread( () -> { @@ -260,7 +304,9 @@ public void testFailedToCheckValidInInnerTask() throws InterruptedException { } }); thread.start(); - thread.join(TimeUnit.SECONDS.toMillis(2)); + Assert.assertTrue( + "source files should be reset to NORMAL after the inner-space task is dropped", + statusResetLatch.await(5, TimeUnit.SECONDS)); Assert.assertEquals( 0, SystemInfo.getInstance().getCompactionMemoryBlock().getUsedMemoryInBytes()); Assert.assertEquals(0, SystemInfo.getInstance().getCompactionFileNumCost().get());