[HUDI-4408] Reuse old rollover file as base file for flink merge handle

Author: danny0405

hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/FlinkMergeHandle.java

@@ -143,7 +143,13 @@ protected void makeOldAndNewFilePaths(String partitionPath, String oldFileName,
           break;
         }
 
-        rolloverPaths.add(newFilePath);
+        // Override the old file name,
+        // In rare cases, when a checkpoint was aborted and the instant time
+        // is reused, the merge handle generates a new file name
+        // with the reused instant time of last checkpoint, which is duplicate,
+        // use the same name file as new base file in case data loss.
+        oldFilePath = newFilePath;
+        rolloverPaths.add(oldFilePath);
         newFileName = newFileNameWithRollover(rollNumber++);
         newFilePath = makeNewFilePath(partitionPath, newFileName);
         LOG.warn("Duplicate write for MERGE bucket with path: " + oldFilePath + ", rolls over to new path: " + newFilePath);

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

@@ -408,16 +408,6 @@ private boolean hasData() {
         && this.buckets.values().stream().anyMatch(bucket -> bucket.records.size() > 0);
   }
 
-  private void cleanWriteHandles() {
-    if (freshInstant(currentInstant)) {
-      // In rare cases, when a checkpoint was aborted and the instant time
-      // is reused, the merge handle generates a new file name
-      // with the reused instant time of last checkpoint, the write handles
-      // should be kept and reused in case data loss.
-      this.writeClient.cleanHandles();
-    }
-  }
-
   @SuppressWarnings("unchecked, rawtypes")
   private boolean flushBucket(DataBucket bucket) {
     String instant = instantToWrite(true);
@@ -489,7 +479,7 @@ private void flushRemaining(boolean endInput) {
     this.eventGateway.sendEventToCoordinator(event);
     this.buckets.clear();
     this.tracer.reset();
-    cleanWriteHandles();
+    this.writeClient.cleanHandles();
     this.writeStatuses.addAll(writeStatus);
     // blocks flushing until the coordinator starts a new instant
     this.confirming = true;