[FLINK-18426] Remove incompatible deprecated keys from ClusterOptions

ClusterOptions.INITIAL_REGISTRATION_TIMEOUT, MAX_REGISTRATION_TIMEOUT and REFUSED_REGISTRATION_DELAY
have incompatible deprecated options of type Duration associated. This causes the system to fail
if they are specified. Since the deprecated keys have not been used for a very long time, this commit
will remove the deprecated keys from the ClusterOptions.

This closes apache#12763.

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

@@ -34,14 +34,12 @@ public class ClusterOptions {
  public static final ConfigOption<Long> INITIAL_REGISTRATION_TIMEOUT = ConfigOptions
  .key("cluster.registration.initial-timeout")
  .defaultValue(100L)
- .withDeprecatedKeys("taskmanager.initial-registration-pause", "taskmanager.registration.initial-backoff")
  .withDescription("Initial registration timeout between cluster components in milliseconds.");
 
  @Documentation.Section(Documentation.Sections.EXPERT_FAULT_TOLERANCE)
  public static final ConfigOption<Long> MAX_REGISTRATION_TIMEOUT = ConfigOptions
  .key("cluster.registration.max-timeout")
  .defaultValue(30000L)
- .withDeprecatedKeys("taskmanager.max-registration-pause", "taskmanager.registration.max-backoff")
  .withDescription("Maximum registration timeout between cluster components in milliseconds.");
 
  @Documentation.Section(Documentation.Sections.EXPERT_FAULT_TOLERANCE)
@@ -54,7 +52,6 @@ public class ClusterOptions {
  public static final ConfigOption<Long> REFUSED_REGISTRATION_DELAY = ConfigOptions
  .key("cluster.registration.refused-registration-delay")
  .defaultValue(30000L)
- .withDeprecatedKeys("taskmanager.refused-registration-pause", "taskmanager.registration.refused-backoff")
  .withDescription("The pause made after the registration attempt was refused in milliseconds.");
 
  @Documentation.Section(Documentation.Sections.EXPERT_FAULT_TOLERANCE)

flink-runtime/src/test/java/org/apache/flink/runtime/registration/RetryingRegistrationConfigurationTest.java

@@ -20,10 +20,13 @@
 
 import org.apache.flink.configuration.ClusterOptions;
 import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.TaskManagerOptions;
 import org.apache.flink.util.TestLogger;
 
 import org.junit.Test;
 
+import java.time.Duration;
+
 import static org.hamcrest.Matchers.is;
 import static org.junit.Assert.assertThat;
 
@@ -54,4 +57,22 @@ public void testConfigurationParsing() {
  assertThat(retryingRegistrationConfiguration.getErrorDelayMillis(), is(errorRegistrationDelay));
  }
 
+ @Test
+ public void testConfigurationWithDeprecatedOptions() {
+ final Configuration configuration = new Configuration();
+
+ final Duration refusedRegistrationBackoff = Duration.ofMinutes(42L);
+ final Duration registrationMaxBackoff = Duration.ofSeconds(1L);
+ final Duration initialRegistrationBackoff = Duration.ofHours(1337L);
+
+ configuration.set(TaskManagerOptions.REFUSED_REGISTRATION_BACKOFF, refusedRegistrationBackoff);
+ configuration.set(TaskManagerOptions.REGISTRATION_MAX_BACKOFF, registrationMaxBackoff);
+ configuration.set(TaskManagerOptions.INITIAL_REGISTRATION_BACKOFF, initialRegistrationBackoff);
+
+ final RetryingRegistrationConfiguration retryingRegistrationConfiguration = RetryingRegistrationConfiguration.fromConfiguration(configuration);
+
+ assertThat(retryingRegistrationConfiguration.getInitialRegistrationTimeoutMillis(), is(ClusterOptions.INITIAL_REGISTRATION_TIMEOUT.defaultValue()));
+ assertThat(retryingRegistrationConfiguration.getRefusedDelayMillis(), is(ClusterOptions.REFUSED_REGISTRATION_DELAY.defaultValue()));
+ assertThat(retryingRegistrationConfiguration.getMaxRegistrationTimeoutMillis(), is(ClusterOptions.MAX_REGISTRATION_TIMEOUT.defaultValue()));
+ }
 }