Fix FileId not found exception when FileId is passed to HoodieMergeHandle

Author: voonhous

hudi-common/src/main/java/org/apache/hudi/common/table/view/TableFileSystemView.java

@@ -157,6 +157,8 @@ interface SliceView extends SliceViewWithLatestSlice {
 
   /**
    * Stream all the file groups for a given partition.
+   * <p>
+   * Note: This method will return all file groups in a partition, i.e. uncommitted filegroups will be returned
    */
   Stream<HoodieFileGroup> getAllFileGroups(String partitionPath);
 

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

@@ -157,8 +157,8 @@ private void bootstrapIndexIfNeed(String partition) {
 
     // Load existing fileID belongs to this task
     Map<Integer, String> bucketToFileIDMap = new HashMap<>();
-    this.writeClient.getHoodieTable().getHoodieView().getAllFileGroups(partition).forEach(fileGroup -> {
-      String fileID = fileGroup.getFileGroupId().getFileId();
+    this.writeClient.getHoodieTable().getHoodieView().getLatestBaseFiles(partition).forEach(baseFile -> {
+      String fileID = baseFile.getFileId();
       int bucketNumber = BucketIdentifier.bucketIdFromFileId(fileID);
       if (isBucketToLoad(bucketNumber, partition)) {
         LOG.info(String.format("Should load this partition bucket %s with fileID %s", bucketNumber, fileID));