[hotfix][core] Add Configuration.fromMap

deepfield · Nov 3, 2020 · 4fa58f3 · 4fa58f3
1 parent e90f2b1
commit 4fa58f3
Show file tree

Hide file tree

Showing 3 changed files with 15 additions and 11 deletions.
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/Configuration.java b/flink-core/src/main/java/org/apache/flink/configuration/Configuration.java
@@ -84,6 +84,17 @@ public Configuration(Configuration other) {
 
  // --------------------------------------------------------------------------------------------
 
+ /**
+ * Creates a new configuration that is initialized with the options of the given map.
+ */
+ public static Configuration fromMap(Map<String, String> map) {
+ final Configuration configuration = new Configuration();
+ map.forEach(configuration::setString);
+ return configuration;
+ }
+
+ // --------------------------------------------------------------------------------------------
+
  /**
  * Returns the class associated with the given key as a string.
  *

diff --git a/flink-core/src/test/java/org/apache/flink/configuration/DelegatingConfigurationTest.java b/flink-core/src/test/java/org/apache/flink/configuration/DelegatingConfigurationTest.java
@@ -20,7 +20,6 @@
 
 import org.junit.Test;
 
-import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.lang.reflect.Modifier;
 import java.util.Map;
@@ -36,14 +35,15 @@
 public class DelegatingConfigurationTest {
 
  @Test
- public void testIfDelegatesImplementAllMethods() throws IllegalAccessException, IllegalArgumentException, InvocationTargetException {
+ public void testIfDelegatesImplementAllMethods() throws IllegalArgumentException {
 
  // For each method in the Configuration class...
  Method[] confMethods = Configuration.class.getDeclaredMethods();
  Method[] delegateMethods = DelegatingConfiguration.class.getDeclaredMethods();
 
  for (Method configurationMethod : confMethods) {
- if (!Modifier.isPublic(configurationMethod.getModifiers())) {
+ final int mod = configurationMethod.getModifiers();
+ if (!Modifier.isPublic(mod) || Modifier.isStatic(mod)) {
  continue;
  }
 

diff --git a/...-table/flink-table-common/src/main/java/org/apache/flink/table/factories/FactoryUtil.java b/...-table/flink-table-common/src/main/java/org/apache/flink/table/factories/FactoryUtil.java
@@ -45,7 +45,6 @@
 import java.util.HashSet;
 import java.util.LinkedList;
 import java.util.List;
-import java.util.Map;
 import java.util.Optional;
 import java.util.ServiceConfigurationError;
 import java.util.ServiceLoader;
@@ -404,12 +403,6 @@ private static String stringifyOption(String key, String value) {
  EncodingUtils.escapeSingleQuotes(value));
  }
 
- private static Configuration asConfiguration(Map<String, String> options) {
- final Configuration configuration = new Configuration();
- options.forEach(configuration::setString);
- return configuration;
- }
-
  private static <T> T readOption(ReadableConfig options, ConfigOption<T> option) {
  try {
  return options.get(option);
@@ -440,7 +433,7 @@ public static class TableFactoryHelper {
  private TableFactoryHelper(DynamicTableFactory tableFactory, DynamicTableFactory.Context context) {
  this.tableFactory = tableFactory;
  this.context = context;
- this.allOptions = asConfiguration(context.getCatalogTable().getOptions());
+ this.allOptions = Configuration.fromMap(context.getCatalogTable().getOptions());
  this.consumedOptionKeys = new HashSet<>();
  this.consumedOptionKeys.add(PROPERTY_VERSION.key());
  this.consumedOptionKeys.add(CONNECTOR.key());