[modify]解决JobManager由于FlinkHudiTable未定义字段导致的异常

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/mor/MergeOnReadTableState.java

@@ -18,21 +18,24 @@
 
 package org.apache.hudi.table.format.mor;
 
-import org.apache.hudi.common.model.HoodieRecord;
-
-import org.apache.flink.table.types.logical.LogicalType;
-import org.apache.flink.table.types.logical.RowType;
-
 import java.io.Serializable;
 import java.util.Arrays;
 import java.util.List;
+import java.util.stream.Collectors;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.exception.HoodieException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * Statistics for merge on read table source.
  */
 public class MergeOnReadTableState implements Serializable {
 
   private static final long serialVersionUID = 1L;
+  private static final Logger LOG = LoggerFactory.getLogger(MergeOnReadTableState.class);
 
   private final RowType rowType;
   private final RowType requiredRowType;
@@ -56,6 +59,8 @@ public MergeOnReadTableState(
     this.inputSplits = inputSplits;
     this.pkFields = pkFields;
     this.operationPos = rowType.getFieldIndex(HoodieRecord.OPERATION_METADATA_FIELD);
+
+    checkRequiredFieldValidation();
   }
 
   public RowType getRowType() {
@@ -114,4 +119,14 @@ public LogicalType[] getPkTypes(int[] pkOffsets) {
     return Arrays.stream(pkOffsets).mapToObj(offset -> requiredTypes[offset])
         .toArray(LogicalType[]::new);
   }
+
+  private void checkRequiredFieldValidation() {
+    final List<String> fieldNames = rowType.getFieldNames();
+    List<String> invalidFieldNames = requiredRowType.getFieldNames().stream().filter(requiredFieldName -> !fieldNames.contains(requiredFieldName)).collect(Collectors.toList());
+
+    if (!invalidFieldNames.isEmpty()) {
+      LOG.warn("Unknown required fields: {}, all known are: {}", String.join(",", invalidFieldNames), String.join(",", fieldNames));
+      throw new HoodieException("Unknown required fields: " + String.join(",", invalidFieldNames));
+    }
+  }
 }