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 76e6695741025..1a7be7e30e0c1 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 static HoodieRecord generateHoodieRecord(HoodieKey hoodieKey, FileSlice f
     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 258f88403a5b2..1fde4593707b6 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 void processElement(StreamRecord<I> element) throws Exception {
     // 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 e05f09552e0d4..da55e27f0c03b 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 static void createProperties(
       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()));