[FLINK-13965] Keep hasDeprecatedKeys and deprecatedKeys methods in Co…

…nfigOption and mark it with @deprecated annotation

This closes apache#9691.

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

@@ -23,6 +23,7 @@
 
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
 import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -195,6 +196,31 @@ public T defaultValue() {
 		return defaultValue;
 	}
 
+	/**
+	 * Checks whether this option has deprecated keys.
+	 * @return True if the option has deprecated keys, false if not.
+	 * @deprecated Replaced by {@link #hasFallbackKeys()}
+	 */
+	@Deprecated
+	public boolean hasDeprecatedKeys() {
+		return fallbackKeys == EMPTY ? false :
+			Arrays.stream(fallbackKeys).anyMatch(FallbackKey::isDeprecated);
+	}
+
+	/**
+	 * Gets the deprecated keys, in the order to be checked.
+	 * @return The option's deprecated keys.
+	 * @deprecated Replaced by {@link #fallbackKeys()}
+	 */
+	@Deprecated
+	public Iterable<String> deprecatedKeys() {
+		return fallbackKeys == EMPTY ? Collections.<String>emptyList() :
+			Arrays.stream(fallbackKeys)
+				.filter(FallbackKey::isDeprecated)
+				.map(FallbackKey::getKey)
+				.collect(Collectors.toList());
+	}
+
 	/**
 	 * Checks whether this option has fallback keys.
 	 * @return True if the option has fallback keys, false if not.

flink-core/src/test/java/org/apache/flink/configuration/ConfigOptionTest.java

@@ -20,10 +20,15 @@
 
 import org.apache.flink.util.TestLogger;
 
+import org.apache.flink.shaded.guava18.com.google.common.collect.Sets;
+
 import org.junit.Test;
 
 import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Set;
 
 import static org.hamcrest.Matchers.containsInAnyOrder;
 import static org.junit.Assert.assertEquals;
@@ -86,4 +91,30 @@ public void testDeprecationFlagForMixedAlternativeKeys() {
 		assertThat(fallbackKeys, containsInAnyOrder("fallback1", "fallback2"));
 		assertThat(deprecatedKeys, containsInAnyOrder("deprecated1", "deprecated2"));
 	}
+
+	@Test
+	public void testDeprecationForDeprecatedKeys() {
+		String[] deprecatedKeys = new String[] { "deprecated1", "deprecated2" };
+		final Set<String> expectedDeprecatedKeys = new HashSet<>(Arrays.asList(deprecatedKeys));
+
+		final ConfigOption<Integer> optionWithDeprecatedKeys = ConfigOptions
+			.key("key")
+			.defaultValue(0)
+			.withDeprecatedKeys(deprecatedKeys)
+			.withFallbackKeys("fallback1");
+
+		assertTrue(optionWithDeprecatedKeys.hasDeprecatedKeys());
+		assertEquals(expectedDeprecatedKeys, Sets.newHashSet(optionWithDeprecatedKeys.deprecatedKeys()));
+	}
+
+	@Test
+	public void testNoDeprecationForFallbackKeysWithoutDeprecated() {
+		final ConfigOption<Integer> optionWithFallbackKeys = ConfigOptions
+			.key("key")
+			.defaultValue(0)
+			.withFallbackKeys("fallback1");
+
+		assertFalse(optionWithFallbackKeys.hasDeprecatedKeys());
+	}
+
 }