[FLINK-22475][table-common] Exclude options with '#' placeholder from validation of required options

Author: Airblader

flink-table/flink-table-common/src/main/java/org/apache/flink/table/factories/FactoryUtil.java

@@ -107,6 +107,12 @@ public final class FactoryUtil {
      */
     public static final String FORMAT_SUFFIX = ".format";
 
+    /**
+     * The placeholder symbol to be used for keys of options which can be templated. See {@link
+     * Factory} for details.
+     */
+    public static final String PLACEHOLDER_SYMBOL = "#";
+
     /**
      * Creates a {@link DynamicTableSource} from a {@link CatalogTable}.
      *
@@ -362,6 +368,9 @@ public static void validateFactoryOptions(
 
         final List<String> missingRequiredOptions =
                 requiredOptions.stream()
+                        // Templated options will never appear with their template key, so we need
+                        // to ignore them as required properties here
+                        .filter(option -> !option.key().contains(PLACEHOLDER_SYMBOL))
                         .filter(option -> readOption(options, option) == null)
                         .map(ConfigOption::key)
                         .sorted()
@@ -384,7 +393,7 @@ public static void validateUnconsumedKeys(
             String factoryIdentifier, Set<String> allOptionKeys, Set<String> consumedOptionKeys) {
         final Set<String> remainingOptionKeys = new HashSet<>(allOptionKeys);
         remainingOptionKeys.removeAll(consumedOptionKeys);
-        if (remainingOptionKeys.size() > 0) {
+        if (!remainingOptionKeys.isEmpty()) {
             throw new ValidationException(
                     String.format(
                             "Unsupported options found for '%s'.\n\n"

flink-table/flink-table-common/src/test/java/org/apache/flink/table/factories/FactoryUtilTest.java

@@ -18,6 +18,9 @@
 
 package org.apache.flink.table.factories;
 
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
 import org.apache.flink.table.api.TableSchema;
 import org.apache.flink.table.api.ValidationException;
 import org.apache.flink.table.catalog.Catalog;
@@ -37,9 +40,11 @@
 
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
 import java.util.function.Consumer;
 
 import static org.apache.flink.core.testutils.FlinkMatchers.containsCause;
@@ -277,6 +282,14 @@ public void testConnectorErrorHint() {
         }
     }
 
+    @Test
+    public void testRequiredPlaceholderOption() {
+        final Set<ConfigOption<?>> requiredOptions = new HashSet<>();
+        requiredOptions.add(ConfigOptions.key("fields.#.min").intType().noDefaultValue());
+
+        FactoryUtil.validateFactoryOptions(requiredOptions, new HashSet<>(), new Configuration());
+    }
+
     @Test
     public void testCreateCatalog() {
         final Map<String, String> options = new HashMap<>();