[flink] fix unstable test: testAsyncCompactionFileDeletedWhenShutdown

liming30 · Nov 6, 2023 · da35cbe · da35cbe
1 parent 4a656d9
commit da35cbe
Show file tree

Hide file tree

Showing 2 changed files with 6 additions and 3 deletions.
diff --git a/...n/src/main/java/org/apache/paimon/flink/sink/AppendOnlyTableCompactionWorkerOperator.java b/...n/src/main/java/org/apache/paimon/flink/sink/AppendOnlyTableCompactionWorkerOperator.java
@@ -109,7 +109,8 @@ public void processElement(StreamRecord<AppendOnlyCompactionTask> element) throw
         result.add(workerExecutor().submit(() -> task.doCompact(write)));
     }
 
-    private ExecutorService workerExecutor() {
+    @VisibleForTesting
+    ExecutorService workerExecutor() {
         if (lazyCompactExecutor == null) {
             lazyCompactExecutor =
                     Executors.newSingleThreadScheduledExecutor(

diff --git a/...c/test/java/org/apache/paimon/flink/sink/AppendOnlyTableCompactionWorkerOperatorTest.java b/...c/test/java/org/apache/paimon/flink/sink/AppendOnlyTableCompactionWorkerOperatorTest.java
@@ -39,6 +39,7 @@
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
@@ -78,7 +79,7 @@ public void testAsyncCompactionWorks() throws Exception {
                                         workerOperator.prepareCommit(false, Long.MAX_VALUE));
 
                                 Long now = System.currentTimeMillis();
-                                if (timeStart - now > timeout && committables.size() != 4) {
+                                if (now - timeStart > timeout && committables.size() != 4) {
                                     throw new RuntimeException(
                                             "Timeout waiting for compaction, maybe some error happens in "
                                                     + AppendOnlyTableCompactionWorkerOperator.class
@@ -105,6 +106,7 @@ public void testAsyncCompactionFileDeletedWhenShutdown() throws Exception {
         AppendOnlyTableCompactionWorkerOperator workerOperator =
                 new AppendOnlyTableCompactionWorkerOperator(
                         (AppendOnlyFileStoreTable) getTableDefault(), "user");
+        ExecutorService workerExecutor = workerOperator.workerExecutor();
 
         // write 200 files
         List<CommitMessage> commitMessages = writeDataDefault(200, 40);
@@ -151,7 +153,7 @@ public void testAsyncCompactionFileDeletedWhenShutdown() throws Exception {
         workerOperator.shutdown();
 
         // wait the last runnable in thread pool to stop
-        Thread.sleep(2_000);
+        workerExecutor.awaitTermination(120, TimeUnit.SECONDS);
 
         for (Future<CommitMessage> f : workerOperator.result()) {
             try {