[HUDI-4207] HoodieFlinkWriteClient.getOrCreateWriteHandle throws an e…

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandle.java

@@ -61,6 +61,7 @@
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
+import java.util.NoSuchElementException;
 import java.util.Map;
 import java.util.Set;
 
@@ -116,17 +117,19 @@ public class HoodieMergeHandle<T extends HoodieRecordPayload, I, K, O> extends H
   public HoodieMergeHandle(HoodieWriteConfig config, String instantTime, HoodieTable<T, I, K, O> hoodieTable,
                            Iterator<HoodieRecord<T>> recordItr, String partitionPath, String fileId,
                            TaskContextSupplier taskContextSupplier, Option<BaseKeyGenerator> keyGeneratorOpt) {
-    this(config, instantTime, hoodieTable, recordItr, partitionPath, fileId, taskContextSupplier,
-        hoodieTable.getBaseFileOnlyView().getLatestBaseFile(partitionPath, fileId).get(), keyGeneratorOpt);
+    super(config, instantTime, partitionPath, fileId, hoodieTable, taskContextSupplier);
+    Option<HoodieBaseFile> baseFileOp = hoodieTable.getBaseFileOnlyView().getLatestBaseFile(partitionPath, fileId);
+    if (!baseFileOp.isPresent()) {
+      throw new NoSuchElementException(String.format("FileID %s of partitionPath %s is not exist, But record loc is 'U' ", fileId, partitionPath));
+    }
+    init(recordItr, baseFileOp.get(), keyGeneratorOpt);
   }
 
   public HoodieMergeHandle(HoodieWriteConfig config, String instantTime, HoodieTable<T, I, K, O> hoodieTable,
                            Iterator<HoodieRecord<T>> recordItr, String partitionPath, String fileId,
                            TaskContextSupplier taskContextSupplier, HoodieBaseFile baseFile, Option<BaseKeyGenerator> keyGeneratorOpt) {
     super(config, instantTime, partitionPath, fileId, hoodieTable, taskContextSupplier);
-    init(fileId, recordItr);
-    init(fileId, partitionPath, baseFile);
-    validateAndSetAndKeyGenProps(keyGeneratorOpt, config.populateMetaFields());
+    init(recordItr, baseFile, keyGeneratorOpt);
   }
 
   /**
@@ -139,15 +142,22 @@ public HoodieMergeHandle(HoodieWriteConfig config, String instantTime, HoodieTab
     this.keyToNewRecords = keyToNewRecords;
     this.useWriterSchemaForCompaction = true;
     this.preserveMetadata = config.isPreserveHoodieCommitMetadataForCompaction();
-    init(fileId, this.partitionPath, dataFileToBeMerged);
-    validateAndSetAndKeyGenProps(keyGeneratorOpt, config.populateMetaFields());
+    init(null, dataFileToBeMerged, keyGeneratorOpt);
   }
 
   private void validateAndSetAndKeyGenProps(Option<BaseKeyGenerator> keyGeneratorOpt, boolean populateMetaFields) {
     ValidationUtils.checkArgument(populateMetaFields == !keyGeneratorOpt.isPresent());
     this.keyGeneratorOpt = keyGeneratorOpt;
   }
 
+  private void init(Iterator<HoodieRecord<T>> recordItr, HoodieBaseFile baseFile, Option<BaseKeyGenerator> keyGeneratorOpt) {
+    if (recordItr != null) {
+      init(this.fileId, recordItr);
+    }
+    init(this.fileId, this.partitionPath, baseFile);
+    validateAndSetAndKeyGenProps(keyGeneratorOpt, this.config.populateMetaFields());
+  }
+
   @Override
   public Schema getWriterSchemaWithMetaFields() {
     return writeSchemaWithMetaFields;