[flink] create table validate options

apache · Aug 6, 2024 · 5eee069 · 5eee069
1 parent 8aac48e
commit 5eee069
Showing 1 changed file with 9 additions and 3 deletions.
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalog.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalog.java
@@ -335,9 +335,13 @@ public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ig
         }
     }
 
-    private void validateOptions(Map<String, String> options) {
+    private static void validateOptions(Map<String, String> options) {
         Set<String> validOptions =
-                Stream.of(CoreOptions.class.getFields())
+                Stream.of(
+                                CoreOptions.class.getFields(),
+                                FlinkConnectorOptions.class.getFields(),
+                                FlinkCatalogOptions.class.getFields())
+                        .flatMap(Stream::of)
                         .filter(field -> java.lang.reflect.Modifier.isStatic(field.getModifiers()))
                         .filter(field -> ConfigOption.class.isAssignableFrom(field.getType()))
                         .map(
@@ -358,7 +362,9 @@ private void validateOptions(Map<String, String> options) {
                                 "log.system",
                                 "kafka.bootstrap.servers",
                                 "kafka.topic",
-                                "kafka.transaction.timeout.ms"));
+                                "kafka.transaction.timeout.ms",
+                                "fields.cnt.aggregate-function",
+                                "kafka.batch.size"));
 
         options.keySet()
                 .forEach(