[FLINK-4764] [core] Introduce Config Options

This is a more concise and maintainable way to define configuration keys, default values,
deprecated keys, etc.

This closes apache#2605

flink-core/src/main/java/org/apache/flink/configuration/ConfigOption.java

@@ -0,0 +1,171 @@
+/*
+ * 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.apache.flink.annotation.PublicEvolving;
+
+import java.util.Arrays;
+import java.util.Collections;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * A {@code ConfigOption} describes a configuration parameter. It encapsulates
+ * the configuration key, deprecated older versions of the key, and an optional
+ * default value for the configuration parameter.
+ * 
+ * <p>{@code ConfigOptions} are built via the {@link ConfigOptions} class.
+ * Once created, a config option is immutable.
+ * 
+ * @param <T> The type of value associated with the configuration option.
+ */
+@PublicEvolving
+public class ConfigOption<T> {
+
+	private static final String[] EMPTY = new String[0];
+
+	// ------------------------------------------------------------------------
+
+	/** The current key for that config option */
+	private final String key;
+
+	/** The list of deprecated keys, in the order to be checked */
+	private final String[] deprecatedKeys;
+
+	/** The default value for this option */
+	private final T defaultValue;
+
+	// ------------------------------------------------------------------------
+
+	/**
+	 * Creates a new config option with no deprecated keys.
+	 *
+	 * @param key             The current key for that config option
+	 * @param defaultValue    The default value for this option
+	 */
+	ConfigOption(String key, T defaultValue) {
+		this.key = checkNotNull(key);
+		this.defaultValue = defaultValue;
+		this.deprecatedKeys = EMPTY;
+	}
+
+	/**
+	 * Creates a new config option with deprecated keys.
+	 *
+	 * @param key             The current key for that config option
+	 * @param defaultValue    The default value for this option
+	 * @param deprecatedKeys  The list of deprecated keys, in the order to be checked
+	 */
+	ConfigOption(String key, T defaultValue, String... deprecatedKeys) {
+		this.key = checkNotNull(key);
+		this.defaultValue = defaultValue;
+		this.deprecatedKeys = deprecatedKeys == null || deprecatedKeys.length == 0 ? EMPTY : deprecatedKeys;
+	}
+
+	// ------------------------------------------------------------------------
+
+	/**
+	 * Creates a new config option, using this option's key and default value, and
+	 * adding the given deprecated keys.
+	 * 
+	 * <p>When obtaining a value from the configuration via {@link Configuration#getValue(ConfigOption)},
+	 * the deprecated keys will be checked in the order provided to this method. The first key for which
+	 * a value is found will be used - that value will be returned.
+	 * 
+	 * @param deprecatedKeys The deprecated keys, in the order in which they should be checked.
+	 * @return A new config options, with the given deprecated keys.
+	 */
+	public ConfigOption<T> withDeprecatedKeys(String... deprecatedKeys) {
+		return new ConfigOption<>(key, defaultValue, deprecatedKeys);
+	}
+
+	// ------------------------------------------------------------------------
+
+	/**
+	 * Gets the configuration key.
+	 * @return The configuration key
+	 */
+	public String key() {
+		return key;
+	}
+
+	/**
+	 * Checks if this option has a default value.
+	 * @return True if it has a default value, false if not.
+	 */
+	public boolean hasDefaultValue() {
+		return defaultValue != null;
+	}
+
+	/**
+	 * Returns the default value, or null, if there is no default value.
+	 * @return The default value, or null.
+	 */
+	public T defaultValue() {
+		return defaultValue;
+	}
+
+	/**
+	 * Checks whether this option has deprecated keys.
+	 * @return True if the option has deprecated keys, false if not.
+	 */
+	public boolean hasDeprecatedKeys() {
+		return deprecatedKeys != EMPTY;
+	}
+
+	/**
+	 * Gets the deprecated keys, in the order to be checked.
+	 * @return The option's deprecated keys.
+	 */
+	public Iterable<String> deprecatedKeys() {
+		return deprecatedKeys == EMPTY ? Collections.<String>emptyList() : Arrays.asList(deprecatedKeys);
+	}
+
+	// ------------------------------------------------------------------------
+
+	@Override
+	public boolean equals(Object o) {
+		if (this == o) {
+			return true;
+		}
+		else if (o != null && o.getClass() == ConfigOption.class) {
+			ConfigOption<?> that = (ConfigOption<?>) o;
+			return this.key.equals(that.key) &&
+					Arrays.equals(this.deprecatedKeys, that.deprecatedKeys) &&
+					(this.defaultValue == null ? that.defaultValue == null :
+							(that.defaultValue != null && this.defaultValue.equals(that.defaultValue)));
+		}
+		else {
+			return false;
+		}
+	}
+
+	@Override
+	public int hashCode() {
+		return 31 * key.hashCode() +
+				17 * Arrays.hashCode(deprecatedKeys) +
+				(defaultValue != null ? defaultValue.hashCode() : 0);
+	}
+
+	@Override
+	public String toString() {
+		return String.format("Key: '%s' , default: %s (deprecated keys: %s)",
+				key, defaultValue, Arrays.toString(deprecatedKeys));
+	}
+}

flink-core/src/main/java/org/apache/flink/configuration/ConfigOptions.java

@@ -0,0 +1,116 @@
+/*
+ * 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.apache.flink.annotation.PublicEvolving;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * {@code ConfigOptions} are used to build a {@link ConfigOption}.
+ * The option is typically built in one of the following pattern:
+ * 
+ * <pre>{@code
+ * // simple string-valued option with a default value
+ * ConfigOption<String> tempDirs = ConfigOptions
+ *     .key("tmp.dir")
+ *     .defaultValue("/tmp");
+ * 
+ * // simple integer-valued option with a default value
+ * ConfigOption<Integer> parallelism = ConfigOptions
+ *     .key("application.parallelism")
+ *     .defaultValue(100);
+ * 
+ * // option with no default value
+ * ConfigOption<String> userName = ConfigOptions
+ *     .key("user.name")
+ *     .noDefaultValue();
+ * 
+ * // option with deprecated keys to check
+ * ConfigOption<Double> threshold = ConfigOptions
+ *     .key("cpu.utilization.threshold")
+ *     .defaultValue(0.9).
+ *     .withDeprecatedKeys("cpu.threshold");
+ * }</pre>
+ */
+@PublicEvolving
+public class ConfigOptions {
+
+	/**
+	 * Starts building a new {@link ConfigOption}.
+	 * 
+	 * @param key The key for the config option.
+	 * @return The builder for the config option with the given key.
+	 */
+	public static OptionBuilder key(String key) {
+		checkNotNull(key);
+		return new OptionBuilder(key);
+	}
+
+	// ------------------------------------------------------------------------
+
+	/**
+	 * The option builder is used to create a {@link ConfigOption}.
+	 * It is instantiated via {@link ConfigOptions#key(String)}.
+	 */
+	public static final class OptionBuilder {
+
+		/** The key for the config option */
+		private final String key;
+
+		/**
+		 * Creates a new OptionBuilder.
+		 * @param key The key for the config option
+		 */
+		OptionBuilder(String key) {
+			this.key = key;
+		}
+
+		/**
+		 * Creates a ConfigOption with the given default value.
+		 * 
+		 * <p>This method does not accept "null". For options with no default value, choose
+		 * one of the {@code noDefaultValue} methods.
+		 * 
+		 * @param value The default value for the config option
+		 * @param <T> The type of the default value.
+		 * @return The config option with the default value.
+		 */
+		public <T> ConfigOption<T> defaultValue(T value) {
+			checkNotNull(value);
+			return new ConfigOption<T>(key, value);
+		}
+
+		/**
+		 * Creates a string-valued option with no default value.
+		 * String-valued options are the only ones that can have no
+		 * default value.
+		 * 
+		 * @return The created ConfigOption.
+		 */
+		public ConfigOption<String> noDefaultValue() {
+			return new ConfigOption<>(key, null);
+		}
+	}
+
+	// ------------------------------------------------------------------------
+
+	/** Not intended to be instantiated */
+	private ConfigOptions() {}
+}