From 264b15df87aa01fe06191f1499b422bcefaba4f7 Mon Sep 17 00:00:00 2001
From: HunterXHunter <1356469429@qq.com>
Date: Mon, 13 Jun 2022 22:36:06 +0800
Subject: [PATCH] =?UTF-8?q?[HUDI-4207]=20HoodieFlinkWriteClient.getOrCreat?=
 =?UTF-8?q?eWriteHandle=20throws=20an=20e=E2=80=A6=20(#5788)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Adding more logs to assist in debugging with HoodieFlinkWriteClient.getOrCreateWriteHandle throwing exception
---
 .../org/apache/hudi/io/HoodieMergeHandle.java | 26 ++++++++++++++-----
 1 file changed, 20 insertions(+), 6 deletions(-)

diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandle.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandle.java
index 92fa5c283..4d6427880 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandle.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieMergeHandle.java
@@ -61,6 +61,7 @@ import java.util.Collections;
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
+import java.util.NoSuchElementException;
 import java.util.Map;
 import java.util.Set;
 
@@ -117,16 +118,14 @@ public class HoodieMergeHandle<T extends HoodieRecordPayload, I, K, O> extends H
                            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);
+            getLatestBaseFile(hoodieTable, partitionPath, fileId), 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,8 +138,7 @@ public class HoodieMergeHandle<T extends HoodieRecordPayload, I, K, O> extends H
     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) {
@@ -148,6 +146,22 @@ public class HoodieMergeHandle<T extends HoodieRecordPayload, I, K, O> extends H
     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());
+  }
+
+  public static HoodieBaseFile getLatestBaseFile(HoodieTable<?, ?, ?, ?> hoodieTable, String partitionPath, String fileId) {
+    Option<HoodieBaseFile> baseFileOp = hoodieTable.getBaseFileOnlyView().getLatestBaseFile(partitionPath, fileId);
+    if (!baseFileOp.isPresent()) {
+      throw new NoSuchElementException(String.format("FileID %s of partition path %s does not exist.", fileId, partitionPath));
+    }
+    return baseFileOp.get();
+  }
+
   @Override
   public Schema getWriterSchemaWithMetaFields() {
     return writeSchemaWithMetaFields;