[HUDI-4999] Refactor FlinkOptions#allOptions and CatalogOptions#allOp…

…tions (apache#6901)
satishkotha · Dec 11, 2022 · c5f965b · c5f965b
1 parent 73c1ee2
commit c5f965b
Show file tree

Hide file tree

Showing 3 changed files with 31 additions and 32 deletions.
diff --git a/...flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java b/...flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java
@@ -28,7 +28,6 @@
 import org.apache.hudi.common.model.WriteOperationType;
 import org.apache.hudi.config.HoodieIndexConfig;
 import org.apache.hudi.config.HoodieWriteConfig;
-import org.apache.hudi.exception.HoodieException;
 import org.apache.hudi.hive.MultiPartKeysValueExtractor;
 import org.apache.hudi.hive.ddl.HiveSyncMode;
 import org.apache.hudi.index.HoodieIndex;
@@ -40,8 +39,6 @@
 import org.apache.flink.configuration.ConfigOptions;
 import org.apache.flink.configuration.Configuration;
 
-import java.lang.reflect.Field;
-import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -899,18 +896,6 @@ public static Set<ConfigOption<?>> optionalOptions() {
    * Returns all the config options.
    */
   public static List<ConfigOption<?>> allOptions() {
-    Field[] declaredFields = FlinkOptions.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;
+    return OptionsResolver.allOptions(FlinkOptions.class);
   }
 }
diff --git a/...nk-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/OptionsResolver.java b/...nk-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/OptionsResolver.java
@@ -21,11 +21,16 @@
 import org.apache.hudi.common.model.DefaultHoodieRecordPayload;
 import org.apache.hudi.common.model.WriteOperationType;
 import org.apache.hudi.common.util.StringUtils;
+import org.apache.hudi.exception.HoodieException;
 import org.apache.hudi.index.HoodieIndex;
 import org.apache.hudi.table.format.FilePathUtils;
 
+import org.apache.flink.configuration.ConfigOption;
 import org.apache.flink.configuration.Configuration;
 
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 
@@ -189,4 +194,27 @@ public static boolean isSpecificStartCommit(Configuration conf) {
     return conf.getOptional(FlinkOptions.READ_START_COMMIT).isPresent()
         && !conf.get(FlinkOptions.READ_START_COMMIT).equalsIgnoreCase(FlinkOptions.START_COMMIT_EARLIEST);
   }
+
+  // -------------------------------------------------------------------------
+  //  Utilities
+  // -------------------------------------------------------------------------
+
+  /**
+   * Returns all the config options with the given class {@code clazz}.
+   */
+  public static List<ConfigOption<?>> allOptions(Class<?> clazz) {
+    Field[] declaredFields = clazz.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;
+  }
 }
diff --git a/...ink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/catalog/CatalogOptions.java b/...ink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/catalog/CatalogOptions.java
@@ -18,15 +18,13 @@
 
 package org.apache.hudi.table.catalog;
 
-import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.configuration.OptionsResolver;
 
 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;
 
@@ -69,19 +67,7 @@ public class CatalogOptions {
    * 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;
+    return OptionsResolver.allOptions(CatalogOptions.class);
   }
 
   /**