[HUDI-4880] Fix corrupted parquet file issue left over by cancelled c…

…ompaction task (#6733)
apache · Nov 3, 2022 · 0d7d3aa · 0d7d3aa
1 parent 14ec963
commit 0d7d3aa
Show file tree

Hide file tree

Showing 2 changed files with 16 additions and 1 deletion.
diff --git a/...tasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/ClusteringOperator.java b/...tasource/hudi-flink/src/main/java/org/apache/hudi/sink/clustering/ClusteringOperator.java
@@ -180,7 +180,10 @@ public void processElement(StreamRecord<ClusteringPlanEvent> element) throws Exc
   }
 
   @Override
-  public void close() {
+  public void close() throws Exception {
+    if (null != this.executor) {
+      this.executor.close();
+    }
     if (this.writeClient != null) {
       this.writeClient.cleanHandlesGracefully();
       this.writeClient.close();

diff --git a/...ink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactFunction.java b/...ink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactFunction.java
@@ -133,4 +133,16 @@ private HoodieWriteConfig reloadWriteConfig() throws Exception {
   public void setExecutor(NonThrownExecutor executor) {
     this.executor = executor;
   }
+
+  @Override
+  public void close() throws Exception {
+    if (null != this.executor) {
+      this.executor.close();
+    }
+    if (null != this.writeClient) {
+      this.writeClient.cleanHandlesGracefully();
+      this.writeClient.close();
+      this.writeClient = null;
+    }
+  }
 }