(incubator-paimon) branch master updated: [flink] wait executors shutdown to avoid test break (#2245)

[lzljs3620320]

This is an automated email from the ASF dual-hosted git repository.

lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git


The following commit(s) were added to refs/heads/master by this push:
     new 56bb71475 [flink] wait executors shutdown to avoid test break (#2245)
56bb71475 is described below

commit 56bb71475d662854bd22d482f9ff5f00aaf13f1b
Author: YeJunHao <41894543+leaves12...@users.noreply.github.com>
AuthorDate: Thu Nov 2 21:55:35 2023 +0800

    [flink] wait executors shutdown to avoid test break (#2245)
---
 .../paimon/flink/sink/AppendOnlyTableCompactionWorkerOperator.java   | 5 +++++
 1 file changed, 5 insertions(+)

diff --git 
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AppendOnlyTableCompactionWorkerOperator.java
 
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AppendOnlyTableCompactionWorkerOperator.java
index f2fe92234..4ed939fdc 100644
--- 
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AppendOnlyTableCompactionWorkerOperator.java
+++ 
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AppendOnlyTableCompactionWorkerOperator.java
@@ -40,6 +40,7 @@ import java.util.Queue;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
 /**
@@ -129,6 +130,10 @@ public class AppendOnlyTableCompactionWorkerOperator
         if (lazyCompactExecutor != null) {
             // ignore runnable tasks in queue
             lazyCompactExecutor.shutdownNow();
+            if (!lazyCompactExecutor.awaitTermination(120, TimeUnit.SECONDS)) {
+                LOG.warn(
+                        "Executors shutdown timeout, there may be some files 
aren't deleted correctly");
+            }
             List<CommitMessage> messages = new ArrayList<>();
             for (Future<CommitMessage> resultFuture : result) {
                 if (!resultFuture.isDone()) {