[HUDI-4856] Missing option for HoodieCatalogFactory

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/catalog/CatalogOptions.java

@@ -18,11 +18,16 @@
 
 package org.apache.hudi.table.catalog;
 
+import org.apache.hudi.exception.HoodieException;
+
 import org.apache.flink.configuration.ConfigOption;
 import org.apache.flink.configuration.ConfigOptions;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.table.catalog.CommonCatalogOptions;
 
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Map;
 
 /**
@@ -60,6 +65,25 @@ public class CatalogOptions {
       .defaultValue(false)
       .withDescription("Whether the table is external, default false");
 
+  /**
+   * Returns all the config options.
+   */
+  public static List<ConfigOption<?>> allOptions() {
+    Field[] declaredFields = CatalogOptions.class.getDeclaredFields();
+    List<ConfigOption<?>> options = new ArrayList<>();
+    for (Field field : declaredFields) {
+      if (java.lang.reflect.Modifier.isStatic(field.getModifiers())
+          && field.getType().equals(ConfigOption.class)) {
+        try {
+          options.add((ConfigOption<?>) field.get(ConfigOption.class));
+        } catch (IllegalAccessException e) {
+          throw new HoodieException("Error while fetching static config option", e);
+        }
+      }
+    }
+    return options;
+  }
+
   /**
    * Returns all the common table options that can be shared.
    *

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/catalog/HoodieCatalogFactory.java

@@ -33,8 +33,6 @@
 import java.util.Locale;
 import java.util.Set;
 
-import static org.apache.flink.table.factories.FactoryUtil.PROPERTY_VERSION;
-
 /**
  * A catalog factory impl that creates {@link HoodieCatalog}.
  */
@@ -77,12 +75,6 @@ public Set<ConfigOption<?>> requiredOptions() {
 
   @Override
   public Set<ConfigOption<?>> optionalOptions() {
-    final Set<ConfigOption<?>> options = new HashSet<>();
-    options.add(CatalogOptions.DEFAULT_DATABASE);
-    options.add(PROPERTY_VERSION);
-    options.add(CatalogOptions.HIVE_CONF_DIR);
-    options.add(CatalogOptions.MODE);
-    options.add(CatalogOptions.CATALOG_PATH);
-    return options;
+    return new HashSet<>(CatalogOptions.allOptions());
   }
 }

hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/catalog/TestHoodieCatalogFactory.java

@@ -18,7 +18,10 @@
 
 package org.apache.hudi.table.catalog;
 
+import org.apache.hudi.adapter.TestTableEnvs;
+
 import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.api.TableEnvironment;
 import org.apache.flink.table.catalog.AbstractCatalog;
 import org.apache.flink.table.catalog.Catalog;
 import org.apache.flink.table.catalog.CommonCatalogOptions;
@@ -34,7 +37,10 @@
 
 import static org.apache.hudi.table.catalog.CatalogOptions.CATALOG_PATH;
 import static org.apache.hudi.table.catalog.CatalogOptions.DEFAULT_DATABASE;
+import static org.hamcrest.CoreMatchers.containsString;
+import static org.hamcrest.MatcherAssert.assertThat;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 
 /**
  * Test cases for {@link HoodieCatalogFactory}.
@@ -47,7 +53,23 @@ public class TestHoodieCatalogFactory {
   File tempFile;
 
   @Test
-  public void testCreateHMSCatalog() {
+  void testCreateCatalogThroughSQL() {
+    TableEnvironment tableEnv = TestTableEnvs.getBatchTableEnv();
+    String catalogDDL = ""
+        + "create catalog hudi_catalog\n"
+        + "  with(\n"
+        + "    'type' = 'hudi',\n"
+        + "    'catalog.path' = '" + tempFile.getAbsolutePath() + "/warehouse',\n"
+        + "    'mode' = 'hms',\n"
+        + "    'hive.conf.dir' = '" + CONF_DIR.getPath() + "',\n"
+        + "    'table.external' = 'true'\n"
+        + "  )\n";
+    RuntimeException exception = assertThrows(RuntimeException.class, () -> tableEnv.executeSql(catalogDDL));
+    assertThat(exception.getMessage(), containsString("hive metastore"));
+  }
+
+  @Test
+  void testCreateHMSCatalog() {
     final String catalogName = "mycatalog";
 
     final HoodieHiveCatalog expectedCatalog = HoodieCatalogTestUtils.createHiveCatalog(catalogName);
@@ -69,7 +91,7 @@ public void testCreateHMSCatalog() {
   }
 
   @Test
-  public void testCreateDFSCatalog() {
+  void testCreateDFSCatalog() {
     final String catalogName = "mycatalog";
 
     Map<String, String> catalogOptions = new HashMap<>();