From 0dde7f9185a350599adaab5af5738b73a15243d7 Mon Sep 17 00:00:00 2001
From: ZhangChaoMing <72908278+ZhangChaoming@users.noreply.github.com>
Date: Tue, 2 Mar 2021 17:52:21 +0800
Subject: [PATCH] [HUDI-1584] Modify maker file path, which should start with
 the target base path. (#2539)

---
 .../hudi/operator/InstantGenerateOperator.java      | 13 +++++++++++--
 1 file changed, 11 insertions(+), 2 deletions(-)

diff --git a/hudi-flink/src/main/java/org/apache/hudi/operator/InstantGenerateOperator.java b/hudi-flink/src/main/java/org/apache/hudi/operator/InstantGenerateOperator.java
index 8ae8d8524..d4e9e4d62 100644
--- a/hudi-flink/src/main/java/org/apache/hudi/operator/InstantGenerateOperator.java
+++ b/hudi-flink/src/main/java/org/apache/hudi/operator/InstantGenerateOperator.java
@@ -125,7 +125,7 @@ public class InstantGenerateOperator extends AbstractStreamOperator<HoodieRecord
   public void prepareSnapshotPreBarrier(long checkpointId) throws Exception {
     super.prepareSnapshotPreBarrier(checkpointId);
     String instantMarkerFileName = String.format("%d%s%d%s%d", indexOfThisSubtask, DELIMITER, checkpointId, DELIMITER, recordCounter.get());
-    Path path = new Path(new Path(HoodieTableMetaClient.AUXILIARYFOLDER_NAME, INSTANT_MARKER_FOLDER_NAME), instantMarkerFileName);
+    Path path = generateCurrentMakerFilePath(instantMarkerFileName);
     // create marker file
     fs.create(path, true);
     LOG.info("Subtask [{}] at checkpoint [{}] created marker file [{}]", indexOfThisSubtask, checkpointId, instantMarkerFileName);
@@ -231,7 +231,7 @@ public class InstantGenerateOperator extends AbstractStreamOperator<HoodieRecord
   private boolean checkReceivedData(long checkpointId) throws InterruptedException, IOException {
     int numberOfParallelSubtasks = runtimeContext.getNumberOfParallelSubtasks();
     FileStatus[] fileStatuses;
-    Path instantMarkerPath = new Path(HoodieTableMetaClient.AUXILIARYFOLDER_NAME, INSTANT_MARKER_FOLDER_NAME);
+    Path instantMarkerPath = generateCurrentMakerDirPath();
     // waiting all subtask create marker file ready
     while (true) {
       Thread.sleep(500L);
@@ -283,4 +283,13 @@ public class InstantGenerateOperator extends AbstractStreamOperator<HoodieRecord
       fs.delete(fileStatus.getPath(), true);
     }
   }
+
+  private Path generateCurrentMakerDirPath() {
+    Path auxPath = new Path(cfg.targetBasePath, HoodieTableMetaClient.AUXILIARYFOLDER_NAME);
+    return new Path(auxPath, INSTANT_MARKER_FOLDER_NAME);
+  }
+
+  private Path generateCurrentMakerFilePath(String instantMarkerFileName) {
+    return new Path(generateCurrentMakerDirPath(), instantMarkerFileName);
+  }
 }