[HUDI-4880] Fix corrupted parquet file issue left over by cancelled compaction task

Author: TengHuo

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieWriteHandle.java

@@ -181,13 +181,13 @@ protected Path makeNewFilePath(String partitionPath, String fileName) {
   }
 
   /**
-   * Creates an empty marker file corresponding to storage writer path.
+   * Creates an empty marker file corresponding to storage writer path. Do nothing if marker file already exists.
    *
    * @param partitionPath Partition path
    */
   protected void createMarkerFile(String partitionPath, String dataFileName) {
     WriteMarkersFactory.get(config.getMarkersType(), hoodieTable, instantTime)
-        .create(partitionPath, dataFileName, getIOType());
+        .createIfNotExists(partitionPath, dataFileName, getIOType());
   }
 
   public Schema getWriterSchemaWithMetaFields() {

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

@@ -130,4 +130,14 @@ private HoodieWriteConfig reloadWriteConfig() throws Exception {
   public void setExecutor(NonThrownExecutor executor) {
     this.executor = executor;
   }
+
+  @Override
+  public void close() throws Exception {
+    if (this.asyncCompaction) {
+      this.executor.close();
+    }
+    if (null != this.writeClient) {
+      this.writeClient.close();
+    }
+  }
 }

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

@@ -129,9 +129,6 @@ private void scheduleCompaction(HoodieFlinkTable<?> table, long checkpointId) th
       List<CompactionOperation> operations = compactionPlan.getOperations().stream()
           .map(CompactionOperation::convertFromAvroRecordInstance).collect(toList());
       LOG.info("Execute compaction plan for instant {} as {} file groups", compactionInstantTime, operations.size());
-      WriteMarkersFactory
-          .get(table.getConfig().getMarkersType(), table, compactionInstantTime)
-          .deleteMarkerDir(table.getContext(), table.getConfig().getMarkersDeleteParallelism());
       for (CompactionOperation operation : operations) {
         output.collect(new StreamRecord<>(new CompactionPlanEvent(compactionInstantTime, operation)));
       }