From 99555c897acf9bdd576e7ab233dc448d537e7aea Mon Sep 17 00:00:00 2001
From: BruceLin <brucekellan@gmail.com>
Date: Tue, 17 May 2022 21:09:27 +0800
Subject: [PATCH] [HUDI-4110] Clean the marker files for flink compaction
 (#5604)

---
 .../java/org/apache/hudi/client/HoodieFlinkWriteClient.java    | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkWriteClient.java b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkWriteClient.java
index 524758a67..f62592a49 100644
--- a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkWriteClient.java
+++ b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkWriteClient.java
@@ -372,6 +372,9 @@ public class HoodieFlinkWriteClient<T extends HoodieRecordPayload> extends
     } finally {
       this.txnManager.endTransaction(Option.of(compactionInstant));
     }
+    WriteMarkersFactory
+        .get(config.getMarkersType(), table, compactionCommitTime)
+        .quietDeleteMarkerDir(context, config.getMarkersDeleteParallelism());
     if (compactionTimer != null) {
       long durationInMs = metrics.getDurationInMs(compactionTimer.stop());
       try {