[HUDI-7213] When using wrong tabe.type value in hudi catalog happends…

… npe (#10300)
apache · Feb 27, 2024 · e1625b1 · e1625b1
1 parent 155a66c
commit e1625b1
Show file tree

Hide file tree

Showing 2 changed files with 21 additions and 1 deletion.
diff --git a/...asource/hudi-flink/src/main/java/org/apache/hudi/table/catalog/TableOptionProperties.java b/...asource/hudi-flink/src/main/java/org/apache/hudi/table/catalog/TableOptionProperties.java
@@ -24,6 +24,7 @@
 import org.apache.hudi.common.table.TableSchemaResolver;
 import org.apache.hudi.configuration.FlinkOptions;
 import org.apache.hudi.exception.HoodieIOException;
+import org.apache.hudi.exception.HoodieValidationException;
 import org.apache.hudi.sync.common.util.SparkDataSourceTableUtils;
 import org.apache.hudi.util.AvroSchemaConverter;
 
@@ -189,7 +190,16 @@ public static Map<String, String> translateFlinkTableProperties2Spark(
     return properties.entrySet().stream()
         .filter(e -> KEY_MAPPING.containsKey(e.getKey()) && !catalogTable.getOptions().containsKey(KEY_MAPPING.get(e.getKey())))
         .collect(Collectors.toMap(e -> KEY_MAPPING.get(e.getKey()),
-            e -> e.getKey().equalsIgnoreCase(FlinkOptions.TABLE_TYPE.key()) ? VALUE_MAPPING.get(e.getValue()) : e.getValue()));
+            e -> {
+              if (e.getKey().equalsIgnoreCase(FlinkOptions.TABLE_TYPE.key())) {
+                  String sparkTableType = VALUE_MAPPING.get(e.getValue());
+                  if (sparkTableType == null) {
+                    throw new HoodieValidationException(String.format("%s's value is invalid", e.getKey()));
+                  }
+                  return sparkTableType;
+              }
+              return e.getValue();
+            }));
   }
 
   private static RowType supplementMetaFields(RowType rowType, boolean withOperationField) {

diff --git a/...asource/hudi-flink/src/test/java/org/apache/hudi/table/catalog/TestHoodieHiveCatalog.java b/...asource/hudi-flink/src/test/java/org/apache/hudi/table/catalog/TestHoodieHiveCatalog.java
@@ -270,6 +270,16 @@ public void testCreateNonHoodieTable() throws TableAlreadyExistException, Databa
     }
   }
 
+  @Test
+  public void testCreateHoodieTableWithWrongTableType() {
+    HashMap<String,String> properties = new HashMap<>();
+    properties.put(FactoryUtil.CONNECTOR.key(), "hudi");
+    properties.put("table.type","wrong type");
+    CatalogTable table =
+            new CatalogTableImpl(schema,  properties, "hudi table");
+    assertThrows(HoodieCatalogException.class, () -> hoodieCatalog.createTable(tablePath, table, false));
+  }
+
   @ParameterizedTest
   @ValueSource(booleans = {true, false})
   public void testDropTable(boolean external) throws TableAlreadyExistException, DatabaseNotExistException, TableNotExistException, IOException {