From 7fc7e9b2bc6c5aeabd6f490376e9e0ae76e07874 Mon Sep 17 00:00:00 2001
From: Danny Chan <yuzhao.cyz@gmail.com>
Date: Wed, 3 Nov 2021 14:30:20 +0800
Subject: [PATCH] [HUDI-2660] Delete the view storage properties first before
 creation (#3899)

---
 .../org/apache/hudi/sink/bootstrap/BootstrapOperator.java   | 2 +-
 .../hudi/sink/bootstrap/batch/BatchBootstrapOperator.java   | 6 ++++++
 .../java/org/apache/hudi/util/ViewStorageProperties.java    | 1 +
 3 files changed, 8 insertions(+), 1 deletion(-)

diff --git a/hudi-flink/src/main/java/org/apache/hudi/sink/bootstrap/BootstrapOperator.java b/hudi-flink/src/main/java/org/apache/hudi/sink/bootstrap/BootstrapOperator.java
index 76e669574..1a7be7e30 100644
--- a/hudi-flink/src/main/java/org/apache/hudi/sink/bootstrap/BootstrapOperator.java
+++ b/hudi-flink/src/main/java/org/apache/hudi/sink/bootstrap/BootstrapOperator.java
@@ -231,7 +231,7 @@ public class BootstrapOperator<I, O extends HoodieRecord>
     return hoodieRecord;
   }
 
-  private static boolean shouldLoadFile(String fileId,
+  protected boolean shouldLoadFile(String fileId,
                                         int maxParallelism,
                                         int parallelism,
                                         int taskID) {
diff --git a/hudi-flink/src/main/java/org/apache/hudi/sink/bootstrap/batch/BatchBootstrapOperator.java b/hudi-flink/src/main/java/org/apache/hudi/sink/bootstrap/batch/BatchBootstrapOperator.java
index 258f88403..1fde45937 100644
--- a/hudi-flink/src/main/java/org/apache/hudi/sink/bootstrap/batch/BatchBootstrapOperator.java
+++ b/hudi-flink/src/main/java/org/apache/hudi/sink/bootstrap/batch/BatchBootstrapOperator.java
@@ -75,4 +75,10 @@ public class BatchBootstrapOperator<I, O extends HoodieRecord>
     // send the trigger record
     output.collect((StreamRecord<O>) element);
   }
+
+  @Override
+  protected boolean shouldLoadFile(String fileId, int maxParallelism, int parallelism, int taskID) {
+    // load all the file groups in the partition
+    return true;
+  }
 }
diff --git a/hudi-flink/src/main/java/org/apache/hudi/util/ViewStorageProperties.java b/hudi-flink/src/main/java/org/apache/hudi/util/ViewStorageProperties.java
index e05f09552..da55e27f0 100644
--- a/hudi-flink/src/main/java/org/apache/hudi/util/ViewStorageProperties.java
+++ b/hudi-flink/src/main/java/org/apache/hudi/util/ViewStorageProperties.java
@@ -51,6 +51,7 @@ public class ViewStorageProperties {
       FileSystemViewStorageConfig config) throws IOException {
     Path propertyPath = getPropertiesFilePath(basePath);
     FileSystem fs = FSUtils.getFs(basePath, StreamerUtil.getHadoopConf());
+    fs.delete(propertyPath, false);
     try (FSDataOutputStream outputStream = fs.create(propertyPath)) {
       config.getProps().store(outputStream,
           "Filesystem view storage properties saved on " + new Date(System.currentTimeMillis()));