[HUDI-4880] Fix corrupted parquet file issue left over by cancelled compaction task (apache#6733)

Author: TengHuo

hudi-flink-datasource/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();

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/compact/CompactFunction.java

@@ -130,4 +130,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;
+    }
+  }
 }