You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by dw...@apache.org on 2020/04/07 11:46:18 UTC
[flink] branch release-1.10 updated: [FLINK-16913] Support more
getters in the ReadableConfigToConfigurationAdapter
This is an automated email from the ASF dual-hosted git repository.
dwysakowicz pushed a commit to branch release-1.10
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.10 by this push:
new 23e984d [FLINK-16913] Support more getters in the ReadableConfigToConfigurationAdapter
23e984d is described below
commit 23e984d6e7ae5983762d2be5b9884b4c81311f75
Author: Dawid Wysakowicz <dw...@apache.org>
AuthorDate: Thu Apr 2 10:07:26 2020 +0200
[FLINK-16913] Support more getters in the ReadableConfigToConfigurationAdapter
---
.../ReadableConfigToConfigurationAdapter.java | 114 +++++++++++++--------
.../ReadableConfigToConfigurationAdapterTest.java | 110 ++++++++++++++++++++
2 files changed, 179 insertions(+), 45 deletions(-)
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/ReadableConfigToConfigurationAdapter.java b/flink-core/src/main/java/org/apache/flink/configuration/ReadableConfigToConfigurationAdapter.java
index c2e9292..ab5adf3 100644
--- a/flink-core/src/main/java/org/apache/flink/configuration/ReadableConfigToConfigurationAdapter.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/ReadableConfigToConfigurationAdapter.java
@@ -23,6 +23,8 @@ import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.DataOutputView;
import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Properties;
@@ -104,6 +106,73 @@ public class ReadableConfigToConfigurationAdapter extends Configuration {
}
@Override
+ public <T extends Enum<T>> T getEnum(
+ Class<T> enumClass,
+ ConfigOption<String> configOption) {
+
+ List<String> deprecatedKeys = new ArrayList<>();
+ List<String> fallbackKeys = new ArrayList<>();
+ for (FallbackKey fallbackKey : configOption.fallbackKeys()) {
+ if (fallbackKey.isDeprecated()) {
+ deprecatedKeys.add(fallbackKey.getKey());
+ } else {
+ fallbackKeys.add(fallbackKey.getKey());
+ }
+ }
+
+ ConfigOption<T> key = ConfigOptions.key(configOption.key())
+ .enumType(enumClass)
+ .noDefaultValue()
+ .withFallbackKeys(fallbackKeys.toArray(new String[0]))
+ .withDeprecatedKeys(deprecatedKeys.toArray(new String[0]));
+
+ return get(key);
+ }
+
+ @Override
+ public String getString(String key, String defaultValue) {
+ return get(ConfigOptions.key(key).stringType().defaultValue(defaultValue));
+ }
+
+ @Override
+ public int getInteger(String key, int defaultValue) {
+ return get(ConfigOptions.key(key).intType().defaultValue(defaultValue));
+ }
+
+ @Override
+ public long getLong(String key, long defaultValue) {
+ return get(ConfigOptions.key(key).longType().defaultValue(defaultValue));
+ }
+
+ @Override
+ public boolean getBoolean(String key, boolean defaultValue) {
+ return get(ConfigOptions.key(key).booleanType().defaultValue(defaultValue));
+ }
+
+ @Override
+ public float getFloat(String key, float defaultValue) {
+ return get(ConfigOptions.key(key).floatType().defaultValue(defaultValue));
+ }
+
+ @Override
+ public double getDouble(String key, double defaultValue) {
+ return get(ConfigOptions.key(key).doubleType().defaultValue(defaultValue));
+ }
+
+ @Override
+ public <T> Class<T> getClass(
+ String key,
+ Class<? extends T> defaultValue,
+ ClassLoader classLoader) throws ClassNotFoundException {
+ Optional<String> option = getOptional(ConfigOptions.key(key).stringType().noDefaultValue());
+ if (option.isPresent()) {
+ return (Class<T>) Class.forName(option.get(), true, classLoader);
+ } else {
+ return (Class<T>) defaultValue;
+ }
+ }
+
+ @Override
public <T> T get(ConfigOption<T> option) {
return backingConfig.get(option);
}
@@ -252,13 +321,6 @@ public class ReadableConfigToConfigurationAdapter extends Configuration {
}
@Override
- public <T extends Enum<T>> T getEnum(
- Class<T> enumClass,
- ConfigOption<String> configOption) {
- throw new UnsupportedOperationException("The adapter does not support this method");
- }
-
- @Override
public Configuration clone() {
throw new UnsupportedOperationException("The adapter does not support this method");
}
@@ -269,44 +331,6 @@ public class ReadableConfigToConfigurationAdapter extends Configuration {
}
@Override
- public double getDouble(String key, double defaultValue) {
- throw new UnsupportedOperationException("The adapter does not support this method");
- }
-
- @Override
- public <T> Class<T> getClass(
- String key,
- Class<? extends T> defaultValue,
- ClassLoader classLoader) throws ClassNotFoundException {
- throw new UnsupportedOperationException("The adapter does not support this method");
- }
-
- @Override
- public String getString(String key, String defaultValue) {
- throw new UnsupportedOperationException("The adapter does not support this method");
- }
-
- @Override
- public int getInteger(String key, int defaultValue) {
- throw new UnsupportedOperationException("The adapter does not support this method");
- }
-
- @Override
- public long getLong(String key, long defaultValue) {
- throw new UnsupportedOperationException("The adapter does not support this method");
- }
-
- @Override
- public boolean getBoolean(String key, boolean defaultValue) {
- throw new UnsupportedOperationException("The adapter does not support this method");
- }
-
- @Override
- public float getFloat(String key, float defaultValue) {
- throw new UnsupportedOperationException("The adapter does not support this method");
- }
-
- @Override
public Set<String> keySet() {
throw new UnsupportedOperationException("The adapter does not support this method");
}
diff --git a/flink-core/src/test/java/org/apache/flink/configuration/ReadableConfigToConfigurationAdapterTest.java b/flink-core/src/test/java/org/apache/flink/configuration/ReadableConfigToConfigurationAdapterTest.java
new file mode 100644
index 0000000..8f50d0b
--- /dev/null
+++ b/flink-core/src/test/java/org/apache/flink/configuration/ReadableConfigToConfigurationAdapterTest.java
@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.configuration;
+
+import org.junit.Test;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.junit.Assert.assertThat;
+
+/**
+ * Tests for {@link ReadableConfigToConfigurationAdapter}.
+ */
+public class ReadableConfigToConfigurationAdapterTest {
+ /**
+ * Simple enum for testing.
+ */
+ public enum TestEnum {
+ A,
+ B;
+ }
+
+ @Test
+ public void testGettingEnum() {
+ String enumKey = "enumKey";
+ Configuration configuration = new Configuration();
+ configuration.setString(enumKey, "A");
+
+ ReadableConfigToConfigurationAdapter readableConfig = new ReadableConfigToConfigurationAdapter(configuration);
+ TestEnum actual = readableConfig.getEnum(TestEnum.class, ConfigOptions.key(enumKey).stringType().defaultValue("B"));
+ assertThat(actual, equalTo(TestEnum.A));
+ }
+
+ @Test
+ public void testGettingClass() throws ClassNotFoundException {
+ String classKey = "classKey";
+ Configuration configuration = new Configuration();
+ configuration.setString(classKey, TestClassB.class.getCanonicalName());
+
+ ReadableConfigToConfigurationAdapter readableConfig = new ReadableConfigToConfigurationAdapter(configuration);
+ Class<TestInterface> actual = readableConfig.getClass(
+ classKey,
+ TestClassA.class,
+ this.getClass().getClassLoader());
+ assertThat(actual, equalTo(TestClassB.class));
+ }
+
+ @Test
+ public void testPrimitiveGetters() {
+ Configuration configuration = new Configuration();
+ configuration.setString("int", String.valueOf(Integer.MAX_VALUE));
+ configuration.setString("long", String.valueOf(Long.MAX_VALUE));
+ configuration.setString("boolean", String.valueOf(false));
+ configuration.setString("string", "XYZ");
+ configuration.setString("double", String.valueOf(Double.MAX_VALUE));
+ configuration.setString("float", String.valueOf(Float.MAX_VALUE));
+
+ ReadableConfigToConfigurationAdapter readableConfig = new ReadableConfigToConfigurationAdapter(configuration);
+ assertThat(readableConfig.getInteger("int", 0), equalTo(Integer.MAX_VALUE));
+ assertThat(readableConfig.getLong("long", 0L), equalTo(Long.MAX_VALUE));
+ assertThat(readableConfig.getBoolean("boolean", true), equalTo(false));
+ assertThat(readableConfig.getString("string", "ABC"), equalTo("XYZ"));
+ assertThat(readableConfig.getDouble("double", 0D), equalTo(Double.MAX_VALUE));
+ assertThat(readableConfig.getFloat("float", 0F), equalTo(Float.MAX_VALUE));
+ }
+
+ @Test
+ public void testPrimitiveOptionGetters() {
+ Configuration configuration = new Configuration();
+ configuration.setString("int", String.valueOf(Integer.MAX_VALUE));
+ configuration.setString("long", String.valueOf(Long.MAX_VALUE));
+ configuration.setString("boolean", String.valueOf(false));
+ configuration.setString("string", "XYZ");
+ configuration.setString("double", String.valueOf(Double.MAX_VALUE));
+ configuration.setString("float", String.valueOf(Float.MAX_VALUE));
+
+ ReadableConfigToConfigurationAdapter readableConfig = new ReadableConfigToConfigurationAdapter(configuration);
+ assertThat(readableConfig.get(ConfigOptions.key("int").intType().defaultValue(0)), equalTo(Integer.MAX_VALUE));
+ assertThat(readableConfig.get(ConfigOptions.key("long").longType().defaultValue(0L)), equalTo(Long.MAX_VALUE));
+ assertThat(readableConfig.get(ConfigOptions.key("boolean").booleanType().defaultValue(true)), equalTo(false));
+ assertThat(readableConfig.get(ConfigOptions.key("string").stringType().defaultValue("ABC")), equalTo("XYZ"));
+ assertThat(
+ readableConfig.get(ConfigOptions.key("double").doubleType().defaultValue(0D)),
+ equalTo(Double.MAX_VALUE));
+ assertThat(
+ readableConfig.get(ConfigOptions.key("float").floatType().defaultValue(0F)),
+ equalTo(Float.MAX_VALUE));
+ }
+}
+
+interface TestInterface {}
+
+class TestClassA implements TestInterface {}
+
+class TestClassB implements TestInterface {}