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()));