[FLINK-6495] Migrate Akka configuration options

This closes #3935.

flink-clients/src/test/java/org/apache/flink/client/program/ClientConnectionTest.java

@@ -21,6 +21,7 @@ package org.apache.flink.client.program;
 import akka.actor.ActorRef;
 import akka.actor.ActorSystem;
 import akka.actor.Props;
+import org.apache.flink.configuration.AkkaOptions;
 import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.core.testutils.CommonTestUtils;
@@ -92,8 +93,8 @@ public class ClientConnectionTest extends TestLogger {
 	private static void testFailureBehavior(final InetSocketAddress unreachableEndpoint) throws Exception {
 
 		final Configuration config = new Configuration();
-		config.setString(ConfigConstants.AKKA_ASK_TIMEOUT, (ASK_STARTUP_TIMEOUT) + " ms");
-		config.setString(ConfigConstants.AKKA_LOOKUP_TIMEOUT, (CONNECT_TIMEOUT) + " ms");
+		config.setString(AkkaOptions.ASK_TIMEOUT, ASK_STARTUP_TIMEOUT + " ms");
+		config.setString(AkkaOptions.LOOKUP_TIMEOUT, CONNECT_TIMEOUT + " ms");
 		config.setString(ConfigConstants.JOB_MANAGER_IPC_ADDRESS_KEY, unreachableEndpoint.getHostName());
 		config.setInteger(ConfigConstants.JOB_MANAGER_IPC_PORT_KEY, unreachableEndpoint.getPort());
 

flink-clients/src/test/java/org/apache/flink/client/program/ClientTest.java

@@ -32,6 +32,7 @@ import org.apache.flink.api.java.ExecutionEnvironment;
 import org.apache.flink.api.java.io.DiscardingOutputFormat;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.client.program.DetachedEnvironment.DetachedJobExecutionResult;
+import org.apache.flink.configuration.AkkaOptions;
 import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.optimizer.DataStatistics;
@@ -97,7 +98,7 @@ public class ClientTest extends TestLogger {
 		config = new Configuration();
 		config.setString(ConfigConstants.JOB_MANAGER_IPC_ADDRESS_KEY, "localhost");
 		config.setInteger(ConfigConstants.JOB_MANAGER_IPC_PORT_KEY, freePort);
-		config.setString(ConfigConstants.AKKA_ASK_TIMEOUT, ConfigConstants.DEFAULT_AKKA_ASK_TIMEOUT);
+		config.setString(AkkaOptions.ASK_TIMEOUT, AkkaOptions.ASK_TIMEOUT.defaultValue());
 
 		try {
 			scala.Tuple2<String, Object> address = new scala.Tuple2<String, Object>("localhost", freePort);

flink-contrib/flink-storm/src/main/java/org/apache/flink/storm/api/FlinkClient.java

@@ -39,6 +39,7 @@ import org.apache.flink.client.program.ClusterClient;
 import org.apache.flink.client.program.JobWithJars;
 import org.apache.flink.client.program.ProgramInvocationException;
 import org.apache.flink.client.program.StandaloneClusterClient;
+import org.apache.flink.configuration.AkkaOptions;
 import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.GlobalConfiguration;
@@ -269,7 +270,7 @@ public class FlinkClient {
 	JobID getTopologyJobId(final String id) {
 		final Configuration configuration = GlobalConfiguration.loadConfiguration();
 		if (this.timeout != null) {
-			configuration.setString(ConfigConstants.AKKA_ASK_TIMEOUT, this.timeout);
+			configuration.setString(AkkaOptions.ASK_TIMEOUT, this.timeout);
 		}
 
 		try {
@@ -309,7 +310,7 @@ public class FlinkClient {
 	private FiniteDuration getTimeout() {
 		final Configuration configuration = GlobalConfiguration.loadConfiguration();
 		if (this.timeout != null) {
-			configuration.setString(ConfigConstants.AKKA_ASK_TIMEOUT, this.timeout);
+			configuration.setString(AkkaOptions.ASK_TIMEOUT, this.timeout);
 		}
 
 		return AkkaUtils.getClientTimeout(configuration);

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

@@ -29,30 +29,114 @@ import org.apache.flink.annotation.PublicEvolving;
 public class AkkaOptions {
 
 	/**
-	 * Timeout for akka ask calls
+	 * Timeout for akka ask calls.
 	 */
-	public static final ConfigOption<String> AKKA_ASK_TIMEOUT = ConfigOptions
+	public static final ConfigOption<String> ASK_TIMEOUT = ConfigOptions
 		.key("akka.ask.timeout")
 		.defaultValue("10 s");
 
+	/**
+	 * The Akka death watch heartbeat interval.
+	 */
+	public static final ConfigOption<String> WATCH_HEARTBEAT_INTERVAL = ConfigOptions
+		.key("akka.watch.heartbeat.interval")
+		.defaultValue(ASK_TIMEOUT.defaultValue());
+
+	/**
+	 * The maximum acceptable Akka death watch heartbeat pause.
+	 */
+	public static final ConfigOption<String> WATCH_HEARTBEAT_PAUSE = ConfigOptions
+		.key("akka.watch.heartbeat.pause")
+		.defaultValue(ASK_TIMEOUT.defaultValue());
+
 	/**
 	 * The Akka tcp connection timeout.
 	 */
-	public static final ConfigOption<String> AKKA_TCP_TIMEOUT = ConfigOptions
+	public static final ConfigOption<String> TCP_TIMEOUT = ConfigOptions
 		.key("akka.tcp.timeout")
 		.defaultValue("20 s");
 
 	/**
-	 * The Akka death watch heartbeat interval.
+	 * Timeout for the startup of the actor system.
 	 */
-	public static final ConfigOption<String> AKKA_WATCH_HEARTBEAT_INTERVAL = ConfigOptions
-		.key("akka.watch.heartbeat.interval")
+	public static final ConfigOption<String> STARTUP_TIMEOUT = ConfigOptions
+		.key("akka.startup-timeout")
+		.noDefaultValue();
+
+	/**
+	 * Heartbeat interval of the transport failure detector.
+	 */
+	public static final ConfigOption<String> TRANSPORT_HEARTBEAT_INTERVAL = ConfigOptions
+		.key("akka.transport.heartbeat.interval")
+		.defaultValue("1000 s");
+
+	/**
+	 * Allowed heartbeat pause for the transport failure detector.
+	 */
+	public static final ConfigOption<String> TRANSPORT_HEARTBEAT_PAUSE = ConfigOptions
+		.key("akka.transport.heartbeat.pause")
+		.defaultValue("6000 s");
+
+	/**
+	 * Detection threshold of transport failure detector.
+	 */
+	public static final ConfigOption<Double> TRANSPORT_THRESHOLD = ConfigOptions
+		.key("akka.transport.threshold")
+		.defaultValue(300.0);
+
+	/**
+	 * Detection threshold for the phi accrual watch failure detector.
+	 */
+	public static final ConfigOption<Integer> WATCH_THRESHOLD = ConfigOptions
+		.key("akka.watch.threshold")
+		.defaultValue(12);
+
+	/**
+	 * Override SSL support for the Akka transport.
+	 */
+	public static final ConfigOption<Boolean> SSL_ENABLED = ConfigOptions
+		.key("akka.ssl.enabled")
+		.defaultValue(true);
+
+	/**
+	 * Maximum framesize of akka messages.
+	 */
+	public static final ConfigOption<String> FRAMESIZE = ConfigOptions
+		.key("akka.framesize")
+		.defaultValue("10485760b");
+
+	/**
+	 * Maximum number of messages until another actor is executed by the same thread.
+	 */
+	public static final ConfigOption<Integer> DISPATCHER_THROUGHPUT = ConfigOptions
+		.key("akka.throughput")
+		.defaultValue(15);
+
+	/**
+	 * Log lifecycle events.
+	 */
+	public static final ConfigOption<Boolean> LOG_LIFECYCLE_EVENTS = ConfigOptions
+		.key("akka.log.lifecycle.events")
+		.defaultValue(false);
+
+	/**
+	 * Timeout for all blocking calls that look up remote actors.
+	 */
+	public static final ConfigOption<String> LOOKUP_TIMEOUT = ConfigOptions
+		.key("akka.lookup.timeout")
 		.defaultValue("10 s");
 
 	/**
-	 * The maximum acceptable Akka death watch heartbeat pause.
+	 * Timeout for all blocking calls on the client side.
 	 */
-	public static final ConfigOption<String> AKKA_WATCH_HEARTBEAT_PAUSE = ConfigOptions
-		.key("akka.watch.heartbeat.pause")
+	public static final ConfigOption<String> CLIENT_TIMEOUT = ConfigOptions
+		.key("akka.client.timeout")
 		.defaultValue("60 s");
+
+	/**
+	 * Exit JVM on fatal Akka errors.
+	 */
+	public static final ConfigOption<Boolean> JVM_EXIT_ON_FATAL_ERROR = ConfigOptions
+		.key("akka.jvm-exit-on-fatal-error")
+		.defaultValue(true);
 }

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

@@ -708,82 +708,130 @@ public final class ConfigConstants {
 
 	/**
 	 * Timeout for the startup of the actor system
+	 *
+	 * @deprecated Use {@link AkkaOptions#STARTUP_TIMEOUT} instead.
 	 */
+	@Deprecated
 	public static final String AKKA_STARTUP_TIMEOUT = "akka.startup-timeout";
 
 	/**
 	 * Heartbeat interval of the transport failure detector
+	 *
+	 * @deprecated Use {@link AkkaOptions#TRANSPORT_HEARTBEAT_INTERVAL} instead.
 	 */
+	@Deprecated
 	public static final String AKKA_TRANSPORT_HEARTBEAT_INTERVAL = "akka.transport.heartbeat.interval";
 
 	/**
 	 * Allowed heartbeat pause for the transport failure detector
+	 *
+	 * @deprecated Use {@link AkkaOptions#TRANSPORT_HEARTBEAT_PAUSE} instead.
 	 */
+	@Deprecated
 	public static final String AKKA_TRANSPORT_HEARTBEAT_PAUSE = "akka.transport.heartbeat.pause";
 
 	/**
 	 * Detection threshold of transport failure detector
+	 *
+	 * @deprecated Use {@link AkkaOptions#TRANSPORT_THRESHOLD} instead.
 	 */
+	@Deprecated
 	public static final String AKKA_TRANSPORT_THRESHOLD = "akka.transport.threshold";
 
 	/**
 	 * Heartbeat interval of watch failure detector
+	 *
+	 * @deprecated Use {@link AkkaOptions#WATCH_HEARTBEAT_INTERVAL} instead.
 	 */
+	@Deprecated
 	public static final String AKKA_WATCH_HEARTBEAT_INTERVAL = "akka.watch.heartbeat.interval";
 
 	/**
 	 * Allowed heartbeat pause for the watch failure detector
+	 *
+	 * @deprecated Use {@link AkkaOptions#WATCH_HEARTBEAT_PAUSE} instead.
 	 */
+	@Deprecated
 	public static final String AKKA_WATCH_HEARTBEAT_PAUSE = "akka.watch.heartbeat.pause";
 
 	/**
 	 * Detection threshold for the phi accrual watch failure detector
+	 *
+	 * @deprecated Use {@link AkkaOptions#WATCH_THRESHOLD} instead.
 	 */
+	@Deprecated
 	public static final String AKKA_WATCH_THRESHOLD = "akka.watch.threshold";
 
 	/**
 	 * Akka TCP timeout
+	 *
+	 * @deprecated Use {@link AkkaOptions#TCP_TIMEOUT} instead.
 	 */
+	@Deprecated
 	public static final String AKKA_TCP_TIMEOUT = "akka.tcp.timeout";
 
 	/**
 	 * Override SSL support for the Akka transport
+	 *
+	 * @deprecated Use {@link AkkaOptions#SSL_ENABLED} instead.
 	 */
+	@Deprecated
 	public static final String AKKA_SSL_ENABLED = "akka.ssl.enabled";
 
 	/**
 	 * Maximum framesize of akka messages
+	 *
+	 * @deprecated Use {@link AkkaOptions#FRAMESIZE} instead.
 	 */
+	@Deprecated
 	public static final String AKKA_FRAMESIZE = "akka.framesize";
 
 	/**
 	 * Maximum number of messages until another actor is executed by the same thread
+	 *
+	 * @deprecated Use {@link AkkaOptions#DISPATCHER_THROUGHPUT} instead.
 	 */
+	@Deprecated
 	public static final String AKKA_DISPATCHER_THROUGHPUT = "akka.throughput";
 
 	/**
 	 * Log lifecycle events
+	 *
+	 * @deprecated Use {@link AkkaOptions#LOG_LIFECYCLE_EVENTS} instead.
 	 */
+	@Deprecated
 	public static final String AKKA_LOG_LIFECYCLE_EVENTS = "akka.log.lifecycle.events";
 
 	/**
 	 * Timeout for all blocking calls on the cluster side
+	 *
+	 * @deprecated Use {@link AkkaOptions#ASK_TIMEOUT} instead.
 	 */
+	@Deprecated
 	public static final String AKKA_ASK_TIMEOUT = "akka.ask.timeout";
 
 	/**
 	 * Timeout for all blocking calls that look up remote actors
+	 *
+	 * @deprecated Use {@link AkkaOptions#LOOKUP_TIMEOUT} instead.
 	 */
+	@Deprecated
 	public static final String AKKA_LOOKUP_TIMEOUT = "akka.lookup.timeout";
 
 	/**
 	 * Timeout for all blocking calls on the client side
+	 *
+	 * @deprecated Use {@link AkkaOptions#CLIENT_TIMEOUT} instead.
 	 */
+	@Deprecated
 	public static final String AKKA_CLIENT_TIMEOUT = "akka.client.timeout";
 
 	/**
 	 * Exit JVM on fatal Akka errors
+	 *
+	 * @deprecated Use {@link AkkaOptions#JVM_EXIT_ON_FATAL_ERROR} instead.
 	 */
+	@Deprecated
 	public static final String AKKA_JVM_EXIT_ON_FATAL_ERROR = "akka.jvm-exit-on-fatal-error";
 	
 	// ----------------------------- Transport SSL Settings--------------------
@@ -1425,26 +1473,70 @@ public final class ConfigConstants {
 
 	// ------------------------------ Akka Values ------------------------------
 
+	/**
+	 * @deprecated Use {@link AkkaOptions#TRANSPORT_HEARTBEAT_INTERVAL} instead.
+	 */
+	@Deprecated
 	public static String DEFAULT_AKKA_TRANSPORT_HEARTBEAT_INTERVAL = "1000 s";
 
+	/**
+	 * @deprecated Use {@link AkkaOptions#TRANSPORT_HEARTBEAT_PAUSE} instead.
+	 */
+	@Deprecated
 	public static String DEFAULT_AKKA_TRANSPORT_HEARTBEAT_PAUSE = "6000 s";
 
+	/**
+	 * @deprecated Use {@link AkkaOptions#TRANSPORT_THRESHOLD} instead.
+	 */
+	@Deprecated
 	public static double DEFAULT_AKKA_TRANSPORT_THRESHOLD = 300.0;
 
+	/**
+	 * @deprecated Use {@link AkkaOptions#WATCH_THRESHOLD} instead.
+	 */
+	@Deprecated
 	public static double DEFAULT_AKKA_WATCH_THRESHOLD = 12;
 
+	/**
+	 * @deprecated Use {@link AkkaOptions#DISPATCHER_THROUGHPUT} instead.
+	 */
+	@Deprecated
 	public static int DEFAULT_AKKA_DISPATCHER_THROUGHPUT = 15;
 
+	/**
+	 * @deprecated Use {@link AkkaOptions#LOG_LIFECYCLE_EVENTS} instead.
+	 */
+	@Deprecated
 	public static boolean DEFAULT_AKKA_LOG_LIFECYCLE_EVENTS = false;
 
+	/**
+	 * @deprecated Use {@link AkkaOptions#FRAMESIZE} instead.
+	 */
+	@Deprecated
 	public static String DEFAULT_AKKA_FRAMESIZE = "10485760b";
 
+	/**
+	 * @deprecated Use {@link AkkaOptions#ASK_TIMEOUT} instead.
+	 */
+	@Deprecated
 	public static String DEFAULT_AKKA_ASK_TIMEOUT = "10 s";
 
+	/**
+	 * @deprecated Use {@link AkkaOptions#LOOKUP_TIMEOUT} instead.
+	 */
+	@Deprecated
 	public static String DEFAULT_AKKA_LOOKUP_TIMEOUT = "10 s";
 
+	/**
+	 * @deprecated Use {@link AkkaOptions#CLIENT_TIMEOUT} instead.
+	 */
+	@Deprecated
 	public static String DEFAULT_AKKA_CLIENT_TIMEOUT = "60 s";
 
+	/**
+	 * @deprecated Use {@link AkkaOptions#SSL_ENABLED} instead.
+	 */
+	@Deprecated
 	public static boolean DEFAULT_AKKA_SSL_ENABLED = true;
 
 	// ----------------------------- SSL Values --------------------------------

flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosTaskManagerRunner.java

@@ -26,6 +26,7 @@ import org.apache.commons.cli.CommandLine;
 import org.apache.commons.cli.CommandLineParser;
 import org.apache.commons.cli.Options;
 import org.apache.commons.cli.PosixParser;
+import org.apache.flink.configuration.AkkaOptions;
 import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.GlobalConfiguration;
@@ -108,7 +109,7 @@ public class MesosTaskManagerRunner {
 		}
 
 		// tell akka to die in case of an error
-		configuration.setBoolean(ConfigConstants.AKKA_JVM_EXIT_ON_FATAL_ERROR, true);
+		configuration.setBoolean(AkkaOptions.JVM_EXIT_ON_FATAL_ERROR, true);
 
 		// Infer the resource identifier from the environment variable
 		String containerID = Preconditions.checkNotNull(envs.get(MesosConfigKeys.ENV_FLINK_CONTAINER_ID));

flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/metrics/MetricFetcher.java

@@ -23,7 +23,7 @@ import akka.dispatch.OnFailure;
 import akka.dispatch.OnSuccess;
 import akka.pattern.Patterns;
 import akka.util.Timeout;
-import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.AkkaOptions;
 import org.apache.flink.runtime.instance.ActorGateway;
 import org.apache.flink.runtime.instance.Instance;
 import org.apache.flink.runtime.messages.JobManagerMessages;
@@ -61,7 +61,7 @@ public class MetricFetcher {
 	private final ActorSystem actorSystem;
 	private final JobManagerRetriever retriever;
 	private final ExecutionContext ctx;
-	private final FiniteDuration timeout = new FiniteDuration(Duration.create(ConfigConstants.DEFAULT_AKKA_ASK_TIMEOUT).toMillis(), TimeUnit.MILLISECONDS);
+	private final FiniteDuration timeout = new FiniteDuration(Duration.create(AkkaOptions.ASK_TIMEOUT.defaultValue()).toMillis(), TimeUnit.MILLISECONDS);
 
 	private MetricStore metrics = new MetricStore();
 	private MetricDumpDeserializer deserializer = new MetricDumpDeserializer();

flink-runtime/src/main/java/org/apache/flink/runtime/client/JobAttachmentClientActor.java

@@ -23,7 +23,7 @@ import akka.actor.Props;
 import akka.actor.Status;
 import akka.dispatch.Futures;
 import org.apache.flink.api.common.JobID;
-import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.AkkaOptions;
 import org.apache.flink.runtime.akka.ListeningBehaviour;
 import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;
 import org.apache.flink.runtime.messages.JobClientMessages;
@@ -114,7 +114,7 @@ public class JobAttachmentClientActor extends JobClientActor {
 					client.tell(
 						decorateMessage(new Status.Failure(
 							new JobClientActorRegistrationTimeoutException("Registration for Job at the JobManager " +
-								"timed out. " +	"You may increase '" + ConfigConstants.AKKA_CLIENT_TIMEOUT +
+								"timed out. " +	"You may increase '" + AkkaOptions.CLIENT_TIMEOUT.key() +
 								"' in case the JobManager needs more time to confirm the job client registration."))),
 						getSelf());
 				}

flink-runtime/src/main/java/org/apache/flink/runtime/client/JobSubmissionClientActor.java

@@ -22,7 +22,7 @@ import akka.actor.ActorRef;
 import akka.actor.Props;
 import akka.actor.Status;
 import akka.dispatch.Futures;
-import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.AkkaOptions;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.runtime.akka.ListeningBehaviour;
 import org.apache.flink.runtime.instance.ActorGateway;
@@ -119,7 +119,7 @@ public class JobSubmissionClientActor extends JobClientActor {
 					client.tell(
 						decorateMessage(new Status.Failure(
 							new JobClientActorSubmissionTimeoutException("Job submission to the JobManager timed out. " +
-								"You may increase '" + ConfigConstants.AKKA_CLIENT_TIMEOUT + "' in case the JobManager " +
+								"You may increase '" + AkkaOptions.CLIENT_TIMEOUT.key() + "' in case the JobManager " +
 								"needs more time to configure and confirm the job submission."))),
 						getSelf());
 				}

flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/FlinkResourceManager.java

@@ -26,7 +26,7 @@ import akka.dispatch.OnComplete;
 import akka.pattern.Patterns;
 import akka.util.Timeout;
 
-import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.AkkaOptions;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.runtime.akka.AkkaUtils;
 import org.apache.flink.runtime.akka.FlinkUntypedActor;
@@ -148,7 +148,7 @@ public abstract class FlinkResourceManager<WorkerType extends ResourceIDRetrieva
 		}
 		catch (Exception e) {
 			lt = new FiniteDuration(
-				Duration.apply(ConfigConstants.DEFAULT_AKKA_LOOKUP_TIMEOUT).toMillis(),
+				Duration.apply(AkkaOptions.LOOKUP_TIMEOUT.defaultValue()).toMillis(),
 				TimeUnit.MILLISECONDS);
 		}
 		this.messageTimeout = lt;

flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/restart/FailureRateRestartStrategy.java

@@ -19,6 +19,7 @@
 package org.apache.flink.runtime.executiongraph.restart;
 
 import org.apache.flink.api.common.time.Time;
+import org.apache.flink.configuration.AkkaOptions;
 import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.runtime.concurrent.impl.FlinkFuture;
@@ -92,7 +93,7 @@ public class FailureRateRestartStrategy implements RestartStrategy {
 		String failuresIntervalString = configuration.getString(
 				ConfigConstants.RESTART_STRATEGY_FAILURE_RATE_FAILURE_RATE_INTERVAL, Duration.apply(1, TimeUnit.MINUTES).toString()
 		);
-		String timeoutString = configuration.getString(ConfigConstants.AKKA_WATCH_HEARTBEAT_INTERVAL, ConfigConstants.DEFAULT_AKKA_ASK_TIMEOUT);
+		String timeoutString = configuration.getString(AkkaOptions.WATCH_HEARTBEAT_INTERVAL);
 		String delayString = configuration.getString(ConfigConstants.RESTART_STRATEGY_FAILURE_RATE_DELAY, timeoutString);
 
 		Duration failuresInterval = Duration.apply(failuresIntervalString);

flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/restart/FixedDelayRestartStrategy.java

@@ -18,6 +18,7 @@
 
 package org.apache.flink.runtime.executiongraph.restart;
 
+import org.apache.flink.configuration.AkkaOptions;
 import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.runtime.concurrent.impl.FlinkFuture;
@@ -72,8 +73,7 @@ public class FixedDelayRestartStrategy implements RestartStrategy {
 		int maxAttempts = configuration.getInteger(ConfigConstants.RESTART_STRATEGY_FIXED_DELAY_ATTEMPTS, 1);
 
 		String timeoutString = configuration.getString(
-			ConfigConstants.AKKA_WATCH_HEARTBEAT_INTERVAL,
-			ConfigConstants.DEFAULT_AKKA_ASK_TIMEOUT);
+			AkkaOptions.WATCH_HEARTBEAT_INTERVAL);
 
 		String delayString = configuration.getString(
 			ConfigConstants.RESTART_STRATEGY_FIXED_DELAY_DELAY,
@@ -87,7 +87,7 @@ public class FixedDelayRestartStrategy implements RestartStrategy {
 		} catch (NumberFormatException nfe) {
 			if (delayString.equals(timeoutString)) {
 				throw new Exception("Invalid config value for " +
-						ConfigConstants.AKKA_WATCH_HEARTBEAT_PAUSE + ": " + timeoutString +
+						AkkaOptions.WATCH_HEARTBEAT_PAUSE.key() + ": " + timeoutString +
 						". Value must be a valid duration (such as '10 s' or '1 min')");
 			} else {
 				throw new Exception("Invalid config value for " +

flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/restart/RestartStrategyFactory.java

@@ -19,6 +19,7 @@
 package org.apache.flink.runtime.executiongraph.restart;
 
 import org.apache.flink.api.common.restartstrategy.RestartStrategies;
+import org.apache.flink.configuration.AkkaOptions;
 import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
 import org.slf4j.Logger;
@@ -88,8 +89,7 @@ public abstract class RestartStrategyFactory implements Serializable {
 				// support deprecated ConfigConstants values
 				final int numberExecutionRetries = configuration.getInteger(ConfigConstants.EXECUTION_RETRIES_KEY,
 					ConfigConstants.DEFAULT_EXECUTION_RETRIES);
-				String pauseString = configuration.getString(ConfigConstants.AKKA_WATCH_HEARTBEAT_PAUSE,
-					ConfigConstants.DEFAULT_AKKA_ASK_TIMEOUT);
+				String pauseString = configuration.getString(AkkaOptions.WATCH_HEARTBEAT_PAUSE);
 				String delayString = configuration.getString(ConfigConstants.EXECUTION_RETRY_DELAY_KEY,
 					pauseString);
 
@@ -100,7 +100,7 @@ public abstract class RestartStrategyFactory implements Serializable {
 				} catch (NumberFormatException nfe) {
 					if (delayString.equals(pauseString)) {
 						throw new Exception("Invalid config value for " +
-							ConfigConstants.AKKA_WATCH_HEARTBEAT_PAUSE + ": " + pauseString +
+							AkkaOptions.WATCH_HEARTBEAT_PAUSE.key() + ": " + pauseString +
 							". Value must be a valid duration (such as '10 s' or '1 min')");
 					} else {
 						throw new Exception("Invalid config value for " +

flink-runtime/src/main/java/org/apache/flink/runtime/query/QueryableStateClient.java

@@ -24,7 +24,7 @@ import akka.dispatch.Mapper;
 import akka.dispatch.Recover;
 import org.apache.flink.api.common.JobID;
 import org.apache.flink.api.common.state.StateDescriptor;
-import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.AkkaOptions;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.IllegalConfigurationException;
 import org.apache.flink.configuration.QueryableStateOptions;
@@ -114,13 +114,11 @@ public class QueryableStateClient {
 		LeaderRetrievalService leaderRetrievalService = highAvailabilityServices.getJobManagerLeaderRetriever(HighAvailabilityServices.DEFAULT_JOB_ID);
 
 		// Get the ask timeout
-		String askTimeoutString = config.getString(
-				ConfigConstants.AKKA_ASK_TIMEOUT,
-				ConfigConstants.DEFAULT_AKKA_ASK_TIMEOUT);
+		String askTimeoutString = config.getString(AkkaOptions.ASK_TIMEOUT);
 
 		Duration timeout = FiniteDuration.apply(askTimeoutString);
 		if (!timeout.isFinite()) {
-			throw new IllegalConfigurationException(ConfigConstants.AKKA_ASK_TIMEOUT
+			throw new IllegalConfigurationException(AkkaOptions.ASK_TIMEOUT.key()
 					+ " is not a finite timeout ('" + askTimeoutString + "')");
 		}
 

flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerConfiguration.java

@@ -53,24 +53,24 @@ public class ResourceManagerConfiguration {
 	// --------------------------------------------------------------------------
 
 	public static ResourceManagerConfiguration fromConfiguration(Configuration configuration) throws ConfigurationException {
-		final String strTimeout = configuration.getString(AkkaOptions.AKKA_ASK_TIMEOUT);
+		final String strTimeout = configuration.getString(AkkaOptions.ASK_TIMEOUT);
 		final Time timeout;
 
 		try {
 			timeout = Time.milliseconds(Duration.apply(strTimeout).toMillis());
 		} catch (NumberFormatException e) {
 			throw new ConfigurationException("Could not parse the resource manager's timeout " +
-				"value " + AkkaOptions.AKKA_ASK_TIMEOUT + '.', e);
+				"value " + AkkaOptions.ASK_TIMEOUT + '.', e);
 		}
 
-		final String strHeartbeatInterval = configuration.getString(AkkaOptions.AKKA_WATCH_HEARTBEAT_INTERVAL);
+		final String strHeartbeatInterval = configuration.getString(AkkaOptions.WATCH_HEARTBEAT_INTERVAL);
 		final Time heartbeatInterval;
 
 		try {
 			heartbeatInterval = Time.milliseconds(Duration.apply(strHeartbeatInterval).toMillis());
 		} catch (NumberFormatException e) {
 			throw new ConfigurationException("Could not parse the resource manager's heartbeat interval " +
-				"value " + AkkaOptions.AKKA_WATCH_HEARTBEAT_INTERVAL + '.', e);
+				"value " + AkkaOptions.WATCH_HEARTBEAT_INTERVAL + '.', e);
 		}
 
 		return new ResourceManagerConfiguration(timeout, heartbeatInterval);

flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotManagerConfiguration.java

@@ -19,9 +19,7 @@
 package org.apache.flink.runtime.resourcemanager.slotmanager;
 
 import org.apache.flink.api.common.time.Time;
-import org.apache.flink.configuration.ConfigConstants;
-import org.apache.flink.configuration.ConfigOption;
-import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.AkkaOptions;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.util.ConfigurationException;
 import org.apache.flink.util.Preconditions;
@@ -55,18 +53,14 @@ public class SlotManagerConfiguration {
 	}
 
 	public static SlotManagerConfiguration fromConfiguration(Configuration configuration) throws ConfigurationException {
-		ConfigOption<String> timeoutOption = ConfigOptions
-			.key(ConfigConstants.AKKA_ASK_TIMEOUT)
-			.defaultValue(ConfigConstants.DEFAULT_AKKA_ASK_TIMEOUT);
-
-		final String strTimeout = configuration.getString(timeoutOption);
+		final String strTimeout = configuration.getString(AkkaOptions.ASK_TIMEOUT);
 		final Time timeout;
 
 		try {
 			timeout = Time.milliseconds(Duration.apply(strTimeout).toMillis());
 		} catch (NumberFormatException e) {
 			throw new ConfigurationException("Could not parse the resource manager's timeout " +
-				"value " + timeoutOption + '.', e);
+				"value " + AkkaOptions.ASK_TIMEOUT + '.', e);
 		}
 
 		return new SlotManagerConfiguration(timeout, timeout, timeout);

flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcServiceUtils.java

@@ -21,7 +21,7 @@ package org.apache.flink.runtime.rpc.akka;
 import akka.actor.ActorSystem;
 import com.typesafe.config.Config;
 import org.apache.flink.api.common.time.Time;
-import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.AkkaOptions;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.runtime.akka.AkkaUtils;
 import org.apache.flink.runtime.highavailability.HighAvailabilityServicesUtils;
@@ -130,9 +130,7 @@ public class AkkaRpcServiceUtils {
 
 		checkNotNull(config, "config is null");
 
-		final boolean sslEnabled = config.getBoolean(
-					ConfigConstants.AKKA_SSL_ENABLED,
-					ConfigConstants.DEFAULT_AKKA_SSL_ENABLED) &&
+		final boolean sslEnabled = config.getBoolean(AkkaOptions.SSL_ENABLED) &&
 				SSLUtils.getSSLEnabled(config);
 
 		return getRpcUrl(

flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerConfiguration.java

@@ -19,6 +19,7 @@
 package org.apache.flink.runtime.taskexecutor;
 
 import org.apache.flink.api.common.time.Time;
+import org.apache.flink.configuration.AkkaOptions;
 import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.TaskManagerOptions;
@@ -146,7 +147,7 @@ public class TaskManagerConfiguration implements TaskManagerRuntimeInfo {
 			timeout = Time.milliseconds(AkkaUtils.getTimeout(configuration).toMillis());
 		} catch (Exception e) {
 			throw new IllegalArgumentException(
-				"Invalid format for '" + ConfigConstants.AKKA_ASK_TIMEOUT +
+				"Invalid format for '" + AkkaOptions.ASK_TIMEOUT.key() +
 					"'.Use formats like '50 s' or '1 min' to specify the timeout.");
 		}
 

flink-runtime/src/main/scala/org/apache/flink/runtime/akka/AkkaUtils.scala

@@ -184,13 +184,11 @@ object AkkaUtils {
    * @return Flink's basic Akka config
    */
   private def getBasicAkkaConfig(configuration: Configuration): Config = {
-    val akkaThroughput = configuration.getInteger(ConfigConstants.AKKA_DISPATCHER_THROUGHPUT,
-      ConfigConstants.DEFAULT_AKKA_DISPATCHER_THROUGHPUT)
-    val lifecycleEvents = configuration.getBoolean(ConfigConstants.AKKA_LOG_LIFECYCLE_EVENTS,
-      ConfigConstants.DEFAULT_AKKA_LOG_LIFECYCLE_EVENTS)
+    val akkaThroughput = configuration.getInteger(AkkaOptions.DISPATCHER_THROUGHPUT)
+    val lifecycleEvents = configuration.getBoolean(AkkaOptions.LOG_LIFECYCLE_EVENTS)
 
     val jvmExitOnFatalError = if (
-      configuration.getBoolean(ConfigConstants.AKKA_JVM_EXIT_ON_FATAL_ERROR, true)){
+      configuration.getBoolean(AkkaOptions.JVM_EXIT_ON_FATAL_ERROR)){
       "on"
     } else {
       "off"
@@ -269,48 +267,36 @@ object AkkaUtils {
                                   bindAddress: String, port: Int,
                                   externalHostname: String, externalPort: Int): Config = {
 
-    val akkaAskTimeout = Duration(configuration.getString(
-      ConfigConstants.AKKA_ASK_TIMEOUT,
-      ConfigConstants.DEFAULT_AKKA_ASK_TIMEOUT))
+    val akkaAskTimeout = Duration(configuration.getString(AkkaOptions.ASK_TIMEOUT))
 
     val startupTimeout = configuration.getString(
-      ConfigConstants.AKKA_STARTUP_TIMEOUT,
+      AkkaOptions.STARTUP_TIMEOUT,
       (akkaAskTimeout * 10).toString)
 
     val transportHeartbeatInterval = configuration.getString(
-      ConfigConstants.AKKA_TRANSPORT_HEARTBEAT_INTERVAL,
-      ConfigConstants.DEFAULT_AKKA_TRANSPORT_HEARTBEAT_INTERVAL)
+      AkkaOptions.TRANSPORT_HEARTBEAT_INTERVAL)
 
     val transportHeartbeatPause = configuration.getString(
-      ConfigConstants.AKKA_TRANSPORT_HEARTBEAT_PAUSE,
-      ConfigConstants.DEFAULT_AKKA_TRANSPORT_HEARTBEAT_PAUSE)
+      AkkaOptions.TRANSPORT_HEARTBEAT_PAUSE)
 
-    val transportThreshold = configuration.getDouble(
-      ConfigConstants.AKKA_TRANSPORT_THRESHOLD,
-      ConfigConstants.DEFAULT_AKKA_TRANSPORT_THRESHOLD)
+    val transportThreshold = configuration.getDouble(AkkaOptions.TRANSPORT_THRESHOLD)
 
-    val watchHeartbeatInterval = configuration.getString(AkkaOptions.AKKA_WATCH_HEARTBEAT_INTERVAL);
+    val watchHeartbeatInterval = configuration.getString(
+      AkkaOptions.WATCH_HEARTBEAT_INTERVAL)
 
-    val watchHeartbeatPause = configuration.getString(AkkaOptions.AKKA_WATCH_HEARTBEAT_PAUSE);
+    val watchHeartbeatPause = configuration.getString(AkkaOptions.WATCH_HEARTBEAT_PAUSE)
 
-    val watchThreshold = configuration.getDouble(
-      ConfigConstants.AKKA_WATCH_THRESHOLD,
-      ConfigConstants.DEFAULT_AKKA_WATCH_THRESHOLD)
+    val watchThreshold = configuration.getInteger(AkkaOptions.WATCH_THRESHOLD)
 
-    val akkaTCPTimeout = configuration.getString(AkkaOptions.AKKA_TCP_TIMEOUT);
+    val akkaTCPTimeout = configuration.getString(AkkaOptions.TCP_TIMEOUT)
 
-    val akkaFramesize = configuration.getString(
-      ConfigConstants.AKKA_FRAMESIZE,
-      ConfigConstants.DEFAULT_AKKA_FRAMESIZE)
+    val akkaFramesize = configuration.getString(AkkaOptions.FRAMESIZE)
 
-    val lifecycleEvents = configuration.getBoolean(
-      ConfigConstants.AKKA_LOG_LIFECYCLE_EVENTS,
-      ConfigConstants.DEFAULT_AKKA_LOG_LIFECYCLE_EVENTS)
+    val lifecycleEvents = configuration.getBoolean(AkkaOptions.LOG_LIFECYCLE_EVENTS)
 
     val logLifecycleEvents = if (lifecycleEvents) "on" else "off"
 
-    val akkaEnableSSLConfig = configuration.getBoolean(ConfigConstants.AKKA_SSL_ENABLED,
-        ConfigConstants.DEFAULT_AKKA_SSL_ENABLED) &&
+    val akkaEnableSSLConfig = configuration.getBoolean(AkkaOptions.SSL_ENABLED) &&
           SSLUtils.getSSLEnabled(configuration)
 
     val akkaEnableSSL = if (akkaEnableSSLConfig) "on" else "off"
@@ -588,14 +574,13 @@ object AkkaUtils {
   }
 
   def getTimeout(config: Configuration): FiniteDuration = {
-    val duration = Duration(config.getString(ConfigConstants.AKKA_ASK_TIMEOUT,
-      ConfigConstants.DEFAULT_AKKA_ASK_TIMEOUT))
+    val duration = Duration(config.getString(AkkaOptions.ASK_TIMEOUT))
 
     new FiniteDuration(duration.toMillis, TimeUnit.MILLISECONDS)
   }
 
   def getDefaultTimeout: Time = {
-    val duration = Duration(ConfigConstants.DEFAULT_AKKA_ASK_TIMEOUT)
+    val duration = Duration(AkkaOptions.ASK_TIMEOUT.defaultValue())
 
     Time.milliseconds(duration.toMillis)
   }
@@ -607,30 +592,24 @@ object AkkaUtils {
   }
 
   def getLookupTimeout(config: Configuration): FiniteDuration = {
-    val duration = Duration(config.getString(
-      ConfigConstants.AKKA_LOOKUP_TIMEOUT,
-      ConfigConstants.DEFAULT_AKKA_LOOKUP_TIMEOUT))
+    val duration = Duration(config.getString(AkkaOptions.LOOKUP_TIMEOUT))
 
     new FiniteDuration(duration.toMillis, TimeUnit.MILLISECONDS)
   }
 
   def getDefaultLookupTimeout: FiniteDuration = {
-    val duration = Duration(ConfigConstants.DEFAULT_AKKA_LOOKUP_TIMEOUT)
+    val duration = Duration(AkkaOptions.LOOKUP_TIMEOUT.defaultValue())
     new FiniteDuration(duration.toMillis, TimeUnit.MILLISECONDS)
   }
 
   def getClientTimeout(config: Configuration): FiniteDuration = {
-    val duration = Duration(
-      config.getString(
-        ConfigConstants.AKKA_CLIENT_TIMEOUT,
-        ConfigConstants.DEFAULT_AKKA_CLIENT_TIMEOUT
-      ))
+    val duration = Duration(config.getString(AkkaOptions.CLIENT_TIMEOUT))
 
     new FiniteDuration(duration.toMillis, TimeUnit.MILLISECONDS)
   }
 
   def getDefaultClientTimeout: FiniteDuration = {
-    val duration = Duration(ConfigConstants.DEFAULT_AKKA_CLIENT_TIMEOUT)
+    val duration = Duration(AkkaOptions.CLIENT_TIMEOUT.defaultValue())
 
     new FiniteDuration(duration.toMillis, TimeUnit.MILLISECONDS)
   }

flink-runtime/src/main/scala/org/apache/flink/runtime/minicluster/FlinkMiniCluster.scala

@@ -27,7 +27,7 @@ import akka.pattern.ask
 import akka.actor.{ActorRef, ActorSystem}
 import com.typesafe.config.Config
 import org.apache.flink.api.common.{JobExecutionResult, JobID, JobSubmissionResult}
-import org.apache.flink.configuration.{ConfigConstants, Configuration}
+import org.apache.flink.configuration.{AkkaOptions, ConfigConstants, Configuration}
 import org.apache.flink.core.fs.Path
 import org.apache.flink.runtime.akka.AkkaUtils
 import org.apache.flink.runtime.client.{JobClient, JobExecutionException}
@@ -265,9 +265,9 @@ abstract class FlinkMiniCluster(
     // https://docs.travis-ci.com/user/environment-variables#Default-Environment-Variables
     if (sys.env.contains("CI")) {
       // Only set if nothing specified in config
-      if (config.getString(ConfigConstants.AKKA_ASK_TIMEOUT, null) == null) {
-        val duration = Duration(ConfigConstants.DEFAULT_AKKA_ASK_TIMEOUT) * 10
-        config.setString(ConfigConstants.AKKA_ASK_TIMEOUT, s"${duration.toSeconds}s")
+      if (!config.contains(AkkaOptions.ASK_TIMEOUT)) {
+        val duration = Duration(AkkaOptions.ASK_TIMEOUT.defaultValue()) * 10
+        config.setString(AkkaOptions.ASK_TIMEOUT, s"${duration.toSeconds}s")
 
         LOG.info(s"Akka ask timeout set to ${duration.toSeconds}s")
       }

flink-runtime/src/test/java/org/apache/flink/runtime/clusterframework/ResourceManagerTest.java

@@ -22,7 +22,7 @@ import akka.actor.ActorSystem;
 import akka.testkit.JavaTestKit;
 import org.apache.flink.api.common.JobID;
 import org.apache.flink.api.common.time.Time;
-import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.AkkaOptions;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.runtime.akka.AkkaUtils;
 import org.apache.flink.runtime.clusterframework.messages.NotifyResourceStarted;
@@ -192,7 +192,7 @@ public class ResourceManagerTest extends TestLogger {
 
 			// set a short timeout for lookups
 			Configuration shortTimeoutConfig = config.clone();
-			shortTimeoutConfig.setString(ConfigConstants.AKKA_LOOKUP_TIMEOUT, "1 s");
+			shortTimeoutConfig.setString(AkkaOptions.LOOKUP_TIMEOUT, "1 s");
 
 			fakeJobManager = TestingUtils.createForwardingActor(
 				system,
@@ -234,7 +234,7 @@ public class ResourceManagerTest extends TestLogger {
 
 			// set a long timeout for lookups such that the test fails in case of timeouts
 			Configuration shortTimeoutConfig = config.clone();
-			shortTimeoutConfig.setString(ConfigConstants.AKKA_LOOKUP_TIMEOUT, "99999 s");
+			shortTimeoutConfig.setString(AkkaOptions.LOOKUP_TIMEOUT, "99999 s");
 
 			fakeJobManager = TestingUtils.createForwardingActor(
 				system,

flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PartialConsumePipelinedResultTest.java

@@ -18,6 +18,7 @@
 
 package org.apache.flink.runtime.io.network.partition;
 
+import org.apache.flink.configuration.AkkaOptions;
 import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.TaskManagerOptions;
@@ -51,7 +52,7 @@ public class PartialConsumePipelinedResultTest {
 		final Configuration config = new Configuration();
 		config.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, NUMBER_OF_TMS);
 		config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, NUMBER_OF_SLOTS_PER_TM);
-		config.setString(ConfigConstants.AKKA_ASK_TIMEOUT, TestingUtils.DEFAULT_AKKA_ASK_TIMEOUT());
+		config.setString(AkkaOptions.ASK_TIMEOUT, TestingUtils.DEFAULT_AKKA_ASK_TIMEOUT());
 		config.setInteger(TaskManagerOptions.NETWORK_NUM_BUFFERS, NUMBER_OF_NETWORK_BUFFERS);
 
 		flink = new TestingCluster(config, true);

flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/JobManagerTest.java

@@ -26,6 +26,7 @@ import akka.testkit.JavaTestKit;
 import akka.testkit.TestProbe;
 import com.typesafe.config.Config;
 import org.apache.flink.api.common.JobID;
+import org.apache.flink.configuration.AkkaOptions;
 import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.JobManagerOptions;
@@ -601,7 +602,7 @@ public class JobManagerTest extends TestLogger {
 		Deadline deadline = new FiniteDuration(100, TimeUnit.SECONDS).fromNow();
 
 		Configuration config = new Configuration();
-		config.setString(ConfigConstants.AKKA_ASK_TIMEOUT, "100ms");
+		config.setString(AkkaOptions.ASK_TIMEOUT, "100ms");
 
 		ActorRef jobManagerActor = JobManager.startJobManagerActors(
 			config,

flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerComponentsStartupShutdownTest.java

@@ -27,6 +27,7 @@ import akka.actor.Props;
 import akka.testkit.JavaTestKit;
 
 import org.apache.flink.api.common.time.Time;
+import org.apache.flink.configuration.AkkaOptions;
 import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.core.memory.MemoryType;
@@ -77,9 +78,9 @@ public class TaskManagerComponentsStartupShutdownTest extends TestLogger {
 		final int BUFFER_SIZE = 32 * 1024;
 
 		Configuration config = new Configuration();
-		config.setString(ConfigConstants.AKKA_WATCH_HEARTBEAT_INTERVAL, "200 ms");
-		config.setString(ConfigConstants.AKKA_WATCH_HEARTBEAT_PAUSE, "1 s");
-		config.setInteger(ConfigConstants.AKKA_WATCH_THRESHOLD, 1);
+		config.setString(AkkaOptions.WATCH_HEARTBEAT_INTERVAL, "200 ms");
+		config.setString(AkkaOptions.WATCH_HEARTBEAT_PAUSE, "1 s");
+		config.setInteger(AkkaOptions.WATCH_THRESHOLD, 1);
 
 		ActorSystem actorSystem = null;
 

flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerRegistrationTest.java

@@ -23,6 +23,7 @@ import akka.actor.ActorSystem;
 import akka.actor.InvalidActorNameException;
 import akka.actor.Terminated;
 import akka.testkit.JavaTestKit;
+import org.apache.flink.configuration.AkkaOptions;
 import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.runtime.akka.AkkaUtils;
@@ -90,10 +91,10 @@ public class TaskManagerRegistrationTest extends TestLogger {
 	@BeforeClass
 	public static void startActorSystem() {
 		config = new Configuration();
-		config.setString(ConfigConstants.AKKA_ASK_TIMEOUT, "5 s");
-		config.setString(ConfigConstants.AKKA_WATCH_HEARTBEAT_INTERVAL, "200 ms");
-		config.setString(ConfigConstants.AKKA_WATCH_HEARTBEAT_PAUSE, "2 s");
-		config.setDouble(ConfigConstants.AKKA_WATCH_THRESHOLD, 2.0);
+		config.setString(AkkaOptions.ASK_TIMEOUT, "5 s");
+		config.setString(AkkaOptions.WATCH_HEARTBEAT_INTERVAL, "200 ms");
+		config.setString(AkkaOptions.WATCH_HEARTBEAT_PAUSE, "2 s");
+		config.setInteger(AkkaOptions.WATCH_THRESHOLD, 2);
 
 		actorSystem = AkkaUtils.createLocalActorSystem(config);
 	}

flink-runtime/src/test/java/org/apache/flink/runtime/testutils/ZooKeeperTestUtils.java

@@ -18,6 +18,7 @@
 
 package org.apache.flink.runtime.testutils;
 
+import org.apache.flink.configuration.AkkaOptions;
 import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.CoreOptions;
@@ -86,10 +87,10 @@ public class ZooKeeperTestUtils {
 		config.setString(HighAvailabilityOptions.HA_STORAGE_PATH, fsStateHandlePath + "/recovery");
 
 		// Akka failure detection and execution retries
-		config.setString(ConfigConstants.AKKA_WATCH_HEARTBEAT_INTERVAL, "1000 ms");
-		config.setString(ConfigConstants.AKKA_WATCH_HEARTBEAT_PAUSE, "6 s");
-		config.setInteger(ConfigConstants.AKKA_WATCH_THRESHOLD, 9);
-		config.setString(ConfigConstants.AKKA_ASK_TIMEOUT, "100 s");
+		config.setString(AkkaOptions.WATCH_HEARTBEAT_INTERVAL, "1000 ms");
+		config.setString(AkkaOptions.WATCH_HEARTBEAT_PAUSE, "6 s");
+		config.setInteger(AkkaOptions.WATCH_THRESHOLD, 9);
+		config.setString(AkkaOptions.ASK_TIMEOUT, "100 s");
 		config.setString(HighAvailabilityOptions.HA_JOB_DELAY, "10 s");
 
 		return config;

flink-runtime/src/test/scala/org/apache/flink/runtime/akka/AkkaSslITCase.scala

@@ -20,7 +20,7 @@ package org.apache.flink.runtime.akka
 
 import akka.actor.ActorSystem
 import akka.testkit.{ImplicitSender, TestKit}
-import org.apache.flink.configuration.{ConfigConstants, Configuration}
+import org.apache.flink.configuration.{AkkaOptions, ConfigConstants, Configuration}
 import org.apache.flink.runtime.testingUtils.{TestingCluster, TestingUtils, ScalaTestingUtils}
 import org.junit.runner.RunWith
 import org.scalatest.{BeforeAndAfterAll, Matchers, WordSpecLike}
@@ -119,7 +119,7 @@ class AkkaSslITCase(_system: ActorSystem)
         val config = new Configuration()
         config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, 1)
         config.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, 1)
-        config.setString(ConfigConstants.AKKA_ASK_TIMEOUT, "2 s")
+        config.setString(AkkaOptions.ASK_TIMEOUT, "2 s")
 
         config.setBoolean(ConfigConstants.SECURITY_SSL_ENABLED, true)
         config.setString(ConfigConstants.SECURITY_SSL_KEYSTORE, "invalid.keystore")
@@ -141,7 +141,7 @@ class AkkaSslITCase(_system: ActorSystem)
         val config = new Configuration()
         config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, 1)
         config.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, 1)
-        config.setString(ConfigConstants.AKKA_ASK_TIMEOUT, "2 s")
+        config.setString(AkkaOptions.ASK_TIMEOUT, "2 s")
 
         config.setBoolean(ConfigConstants.SECURITY_SSL_ENABLED, true)
 

flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/JobManagerConnectionTest.scala

@@ -23,7 +23,7 @@ import java.net.{InetAddress, InetSocketAddress}
 import java.util.concurrent.TimeUnit
 import java.util.concurrent.atomic.{AtomicBoolean, AtomicReference}
 
-import org.apache.flink.configuration.{ConfigConstants, Configuration}
+import org.apache.flink.configuration.{AkkaOptions, ConfigConstants, Configuration}
 import org.apache.flink.runtime.akka.AkkaUtils
 import org.apache.flink.util.NetUtils
 import org.junit.Assert._
@@ -122,7 +122,7 @@ class JobManagerConnectionTest {
 
   private def createConfigWithLowTimeout() : Configuration = {
     val config = new Configuration()
-    config.setString(ConfigConstants.AKKA_LOOKUP_TIMEOUT,
+    config.setString(AkkaOptions.LOOKUP_TIMEOUT,
                      Duration(timeout, TimeUnit.MILLISECONDS).toSeconds + " s")
     config
   }

flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/RecoveryITCase.scala

@@ -21,7 +21,7 @@ package org.apache.flink.runtime.jobmanager
 import akka.actor.{ActorSystem, PoisonPill}
 import akka.testkit.{ImplicitSender, TestKit}
 import org.apache.flink.api.common.ExecutionConfig
-import org.apache.flink.configuration.{ConfigConstants, Configuration}
+import org.apache.flink.configuration.{AkkaOptions, ConfigConstants, Configuration}
 import org.apache.flink.runtime.akka.ListeningBehaviour
 import org.apache.flink.runtime.io.network.partition.ResultPartitionType
 import org.apache.flink.runtime.jobgraph.{DistributionPattern, JobGraph, JobStatus, JobVertex}
@@ -60,7 +60,7 @@ class RecoveryITCase(_system: ActorSystem)
     val config = new Configuration()
     config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, numSlots)
     config.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, numTaskManagers)
-    config.setString(ConfigConstants.AKKA_WATCH_HEARTBEAT_PAUSE, heartbeatTimeout)
+    config.setString(AkkaOptions.WATCH_HEARTBEAT_PAUSE, heartbeatTimeout)
     config.setString(ConfigConstants.RESTART_STRATEGY, "fixeddelay")
     config.setInteger(ConfigConstants.RESTART_STRATEGY_FIXED_DELAY_ATTEMPTS, 1)
     config.setString(ConfigConstants.RESTART_STRATEGY_FIXED_DELAY_DELAY, heartbeatTimeout)

flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingUtils.scala

@@ -28,7 +28,7 @@ import com.google.common.util.concurrent.MoreExecutors
 import com.typesafe.config.ConfigFactory
 import grizzled.slf4j.Logger
 import org.apache.flink.api.common.time.Time
-import org.apache.flink.configuration.{ConfigConstants, Configuration, HighAvailabilityOptions, TaskManagerOptions}
+import org.apache.flink.configuration._
 import org.apache.flink.runtime.akka.AkkaUtils
 import org.apache.flink.runtime.clusterframework.FlinkResourceManager
 import org.apache.flink.runtime.clusterframework.types.ResourceID
@@ -89,7 +89,7 @@ object TestingUtils {
     val config = new Configuration()
     config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, numSlots)
     config.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, numTMs)
-    config.setString(ConfigConstants.AKKA_ASK_TIMEOUT, timeout)
+    config.setString(AkkaOptions.ASK_TIMEOUT, timeout)
 
     val cluster = new TestingCluster(config)
 

flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/test/util/TestBaseUtils.java

@@ -27,6 +27,7 @@ import org.apache.commons.io.FileUtils;
 import org.apache.commons.io.IOUtils;
 
 import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.configuration.AkkaOptions;
 import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.HighAvailabilityOptions;
@@ -146,8 +147,8 @@ public class TestBaseUtils extends TestLogger {
 		config.setLong(TaskManagerOptions.MANAGED_MEMORY_SIZE, TASK_MANAGER_MEMORY_SIZE);
 		config.setBoolean(ConfigConstants.FILESYSTEM_DEFAULT_OVERWRITE_KEY, true);
 
-		config.setString(ConfigConstants.AKKA_ASK_TIMEOUT, DEFAULT_AKKA_ASK_TIMEOUT + "s");
-		config.setString(ConfigConstants.AKKA_STARTUP_TIMEOUT, DEFAULT_AKKA_STARTUP_TIMEOUT);
+		config.setString(AkkaOptions.ASK_TIMEOUT, DEFAULT_AKKA_ASK_TIMEOUT + "s");
+		config.setString(AkkaOptions.STARTUP_TIMEOUT, DEFAULT_AKKA_STARTUP_TIMEOUT);
 
 		config.setInteger(JobManagerOptions.WEB_PORT, 8081);
 		config.setString(JobManagerOptions.WEB_LOG_PATH, logFile.toString());
@@ -287,7 +288,7 @@ public class TestBaseUtils extends TestLogger {
 			String resultPath,
 			String[] excludePrefixes,
 			boolean inOrderOfFiles) throws IOException {
-		
+
 		checkArgument(resultPath != null, "resultPath cannot be be null");
 
 		final BufferedReader[] readers = getResultReader(resultPath, excludePrefixes, inOrderOfFiles);
@@ -328,8 +329,8 @@ public class TestBaseUtils extends TestLogger {
 			String msg = String.format(
 					"Different elements in arrays: expected %d elements and received %d\n" +
 					"files: %s\n expected: %s\n received: %s",
-					expected.length, result.length, 
-					Arrays.toString(getAllInvolvedFiles(resultPath, excludePrefixes)), 
+					expected.length, result.length,
+					Arrays.toString(getAllInvolvedFiles(resultPath, excludePrefixes)),
 					Arrays.toString(expected), Arrays.toString(result));
 			fail(msg);
 		}

flink-tests/src/test/java/org/apache/flink/test/accumulators/AccumulatorLiveITCase.java

@@ -34,6 +34,7 @@ import org.apache.flink.api.common.io.OutputFormat;
 import org.apache.flink.api.java.DataSet;
 import org.apache.flink.api.java.ExecutionEnvironment;
 import org.apache.flink.api.java.LocalEnvironment;
+import org.apache.flink.configuration.AkkaOptions;
 import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.optimizer.DataStatistics;
@@ -120,7 +121,7 @@ public class AccumulatorLiveITCase extends TestLogger {
 		Configuration config = new Configuration();
 		config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, 1);
 		config.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, 1);
-		config.setString(ConfigConstants.AKKA_ASK_TIMEOUT, TestingUtils.DEFAULT_AKKA_ASK_TIMEOUT());
+		config.setString(AkkaOptions.ASK_TIMEOUT, TestingUtils.DEFAULT_AKKA_ASK_TIMEOUT());
 		TestingCluster testingCluster = new TestingCluster(config, false, true);
 		testingCluster.start();
 

flink-tests/src/test/java/org/apache/flink/test/cancelling/CancelingTestBase.java

@@ -20,6 +20,7 @@
 package org.apache.flink.test.cancelling;
 
 import org.apache.flink.api.common.Plan;
+import org.apache.flink.configuration.AkkaOptions;
 import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.TaskManagerOptions;
@@ -88,7 +89,7 @@ public abstract class CancelingTestBase extends TestLogger {
 		config.setBoolean(ConfigConstants.FILESYSTEM_DEFAULT_OVERWRITE_KEY, true);
 		config.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, 2);
 		config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, 4);
-		config.setString(ConfigConstants.AKKA_ASK_TIMEOUT, TestingUtils.DEFAULT_AKKA_ASK_TIMEOUT());
+		config.setString(AkkaOptions.ASK_TIMEOUT, TestingUtils.DEFAULT_AKKA_ASK_TIMEOUT());
 		config.setInteger(TaskManagerOptions.MEMORY_SEGMENT_SIZE, 4096);
 		config.setInteger(TaskManagerOptions.NETWORK_NUM_BUFFERS, 2048);
 

flink-tests/src/test/java/org/apache/flink/test/checkpointing/EventTimeAllWindowCheckpointingITCase.java

@@ -23,6 +23,7 @@ import org.apache.flink.api.common.functions.ReduceFunction;
 import org.apache.flink.api.common.restartstrategy.RestartStrategies;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.api.java.tuple.Tuple4;
+import org.apache.flink.configuration.AkkaOptions;
 import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.TaskManagerOptions;
@@ -74,8 +75,8 @@ public class EventTimeAllWindowCheckpointingITCase extends TestLogger {
 		config.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, 2);
 		config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, PARALLELISM / 2);
 		config.setLong(TaskManagerOptions.MANAGED_MEMORY_SIZE, 48L);
-		config.setString(ConfigConstants.DEFAULT_AKKA_LOOKUP_TIMEOUT, "60 s");
-		config.setString(ConfigConstants.DEFAULT_AKKA_ASK_TIMEOUT, "60 s");
+		config.setString(AkkaOptions.LOOKUP_TIMEOUT, "60 s");
+		config.setString(AkkaOptions.ASK_TIMEOUT, "60 s");
 		cluster = new LocalFlinkMiniCluster(config, false);
 		cluster.start();
 

flink-tests/src/test/java/org/apache/flink/test/classloading/jar/StreamingCustomInputSplitProgram.java

@@ -25,7 +25,7 @@ import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
-import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.AkkaOptions;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.core.io.InputSplit;
 import org.apache.flink.core.io.InputSplitAssigner;
@@ -44,7 +44,7 @@ public class StreamingCustomInputSplitProgram {
 	public static void main(String[] args) throws Exception {
 				Configuration config = new Configuration();
 
-		config.setString(ConfigConstants.AKKA_ASK_TIMEOUT, "5 s");
+		config.setString(AkkaOptions.ASK_TIMEOUT, "5 s");
 
 		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 

flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/RemoteEnvironmentITCase.java

@@ -24,6 +24,7 @@ import org.apache.flink.api.java.DataSet;
 import org.apache.flink.api.java.ExecutionEnvironment;
 import org.apache.flink.api.java.io.LocalCollectionOutputFormat;
 import org.apache.flink.client.program.ProgramInvocationException;
+import org.apache.flink.configuration.AkkaOptions;
 import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.core.io.GenericInputSplit;
@@ -78,7 +79,7 @@ public class RemoteEnvironmentITCase extends TestLogger {
 	@Test(expected=FlinkException.class)
 	public void testInvalidAkkaConfiguration() throws Throwable {
 		Configuration config = new Configuration();
-		config.setString(ConfigConstants.AKKA_STARTUP_TIMEOUT, INVALID_STARTUP_TIMEOUT);
+		config.setString(AkkaOptions.STARTUP_TIMEOUT, INVALID_STARTUP_TIMEOUT);
 
 		final ExecutionEnvironment env = ExecutionEnvironment.createRemoteEnvironment(
 				cluster.getHostname(),
@@ -103,7 +104,7 @@ public class RemoteEnvironmentITCase extends TestLogger {
 	@Test
 	public void testUserSpecificParallelism() throws Exception {
 		Configuration config = new Configuration();
-		config.setString(ConfigConstants.AKKA_STARTUP_TIMEOUT, VALID_STARTUP_TIMEOUT);
+		config.setString(AkkaOptions.STARTUP_TIMEOUT, VALID_STARTUP_TIMEOUT);
 
 		final ExecutionEnvironment env = ExecutionEnvironment.createRemoteEnvironment(
 				cluster.getHostname(),

flink-tests/src/test/java/org/apache/flink/test/recovery/AbstractTaskManagerProcessFailureRecoveryTest.java

@@ -24,6 +24,7 @@ import akka.pattern.Patterns;
 import akka.util.Timeout;
 
 import org.apache.commons.io.FileUtils;
+import org.apache.flink.configuration.AkkaOptions;
 import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.JobManagerOptions;
@@ -127,11 +128,11 @@ public abstract class AbstractTaskManagerProcessFailureRecoveryTest extends Test
 			Tuple2<String, Object> localAddress = new Tuple2<String, Object>("localhost", jobManagerPort);
 
 			Configuration jmConfig = new Configuration();
-			jmConfig.setString(ConfigConstants.AKKA_WATCH_HEARTBEAT_INTERVAL, "1000 ms");
-			jmConfig.setString(ConfigConstants.AKKA_WATCH_HEARTBEAT_PAUSE, "6 s");
-			jmConfig.setInteger(ConfigConstants.AKKA_WATCH_THRESHOLD, 9);
+			jmConfig.setString(AkkaOptions.WATCH_HEARTBEAT_INTERVAL, "1000 ms");
+			jmConfig.setString(AkkaOptions.WATCH_HEARTBEAT_PAUSE, "6 s");
+			jmConfig.setInteger(AkkaOptions.WATCH_THRESHOLD, 9);
 			jmConfig.setString(ConfigConstants.RESTART_STRATEGY_FIXED_DELAY_DELAY, "10 s");
-			jmConfig.setString(ConfigConstants.AKKA_ASK_TIMEOUT, "100 s");
+			jmConfig.setString(AkkaOptions.ASK_TIMEOUT, "100 s");
 			jmConfig.setString(JobManagerOptions.ADDRESS, localAddress._1());
 			jmConfig.setInteger(JobManagerOptions.PORT, jobManagerPort);
 
@@ -409,7 +410,7 @@ public abstract class AbstractTaskManagerProcessFailureRecoveryTest extends Test
 				cfg.setLong(TaskManagerOptions.MANAGED_MEMORY_SIZE, 4L);
 				cfg.setInteger(TaskManagerOptions.NETWORK_NUM_BUFFERS, 100);
 				cfg.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, 2);
-				cfg.setString(ConfigConstants.AKKA_ASK_TIMEOUT, "100 s");
+				cfg.setString(AkkaOptions.ASK_TIMEOUT, "100 s");
 
 				TaskManager.selectNetworkInterfaceAndRunTaskManager(cfg,
 					ResourceID.generate(), TaskManager.class);

flink-tests/src/test/java/org/apache/flink/test/recovery/ChaosMonkeyITCase.java

@@ -22,6 +22,7 @@ import akka.actor.ActorRef;
 import akka.actor.ActorSystem;
 import org.apache.commons.io.FileUtils;
 import org.apache.flink.api.common.JobID;
+import org.apache.flink.configuration.AkkaOptions;
 import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.HighAvailabilityOptions;
@@ -156,9 +157,9 @@ public class ChaosMonkeyITCase extends TestLogger {
 				ZooKeeper.getConnectString(), FileStateBackendBasePath.toURI().toString());
 
 		// Akka and restart timeouts
-		config.setString(ConfigConstants.AKKA_WATCH_HEARTBEAT_INTERVAL, "1000 ms");
-		config.setString(ConfigConstants.AKKA_WATCH_HEARTBEAT_PAUSE, "6 s");
-		config.setInteger(ConfigConstants.AKKA_WATCH_THRESHOLD, 9);
+		config.setString(AkkaOptions.WATCH_HEARTBEAT_INTERVAL, "1000 ms");
+		config.setString(AkkaOptions.WATCH_HEARTBEAT_PAUSE, "6 s");
+		config.setInteger(AkkaOptions.WATCH_THRESHOLD, 9);
 
 		if (checkpointingIntervalMs >= killEvery.toMillis()) {
 			throw new IllegalArgumentException("Relax! You want to kill processes every " +

flink-tests/src/test/java/org/apache/flink/test/recovery/ProcessFailureCancelingITCase.java

@@ -29,7 +29,7 @@ import org.apache.flink.api.common.restartstrategy.RestartStrategies;
 import org.apache.flink.api.java.ExecutionEnvironment;
 import org.apache.flink.api.java.io.DiscardingOutputFormat;
 import org.apache.flink.client.program.ProgramInvocationException;
-import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.AkkaOptions;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.JobManagerOptions;
 import org.apache.flink.runtime.akka.AkkaUtils;
@@ -101,10 +101,10 @@ public class ProcessFailureCancelingITCase extends TestLogger {
 			Tuple2<String, Object> localAddress = new Tuple2<String, Object>("localhost", jobManagerPort);
 
 			Configuration jmConfig = new Configuration();
-			jmConfig.setString(ConfigConstants.AKKA_WATCH_HEARTBEAT_INTERVAL, "5 s");
-			jmConfig.setString(ConfigConstants.AKKA_WATCH_HEARTBEAT_PAUSE, "2000 s");
-			jmConfig.setInteger(ConfigConstants.AKKA_WATCH_THRESHOLD, 10);
-			jmConfig.setString(ConfigConstants.AKKA_ASK_TIMEOUT, "100 s");
+			jmConfig.setString(AkkaOptions.WATCH_HEARTBEAT_INTERVAL, "5 s");
+			jmConfig.setString(AkkaOptions.WATCH_HEARTBEAT_PAUSE, "2000 s");
+			jmConfig.setInteger(AkkaOptions.WATCH_THRESHOLD, 10);
+			jmConfig.setString(AkkaOptions.ASK_TIMEOUT, "100 s");
 			jmConfig.setString(JobManagerOptions.ADDRESS, localAddress._1());
 			jmConfig.setInteger(JobManagerOptions.PORT, jobManagerPort);
 

flink-tests/src/test/java/org/apache/flink/test/recovery/TaskManagerFailureRecoveryITCase.java

@@ -27,6 +27,7 @@ import org.apache.flink.api.common.functions.RichMapFunction;
 import org.apache.flink.api.common.restartstrategy.RestartStrategies;
 import org.apache.flink.api.java.ExecutionEnvironment;
 import org.apache.flink.api.java.io.LocalCollectionOutputFormat;
+import org.apache.flink.configuration.AkkaOptions;
 import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.TaskManagerOptions;
@@ -77,9 +78,9 @@ public class TaskManagerFailureRecoveryITCase extends TestLogger {
 			config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, PARALLELISM);
 			config.setLong(TaskManagerOptions.MANAGED_MEMORY_SIZE, 16L);
 			
-			config.setString(ConfigConstants.AKKA_WATCH_HEARTBEAT_INTERVAL, "500 ms");
-			config.setString(ConfigConstants.AKKA_WATCH_HEARTBEAT_PAUSE, "20 s");
-			config.setInteger(ConfigConstants.AKKA_WATCH_THRESHOLD, 20);
+			config.setString(AkkaOptions.WATCH_HEARTBEAT_INTERVAL, "500 ms");
+			config.setString(AkkaOptions.WATCH_HEARTBEAT_PAUSE, "20 s");
+			config.setInteger(AkkaOptions.WATCH_THRESHOLD, 20);
 
 			cluster = new LocalFlinkMiniCluster(config, false);
 

flink-tests/src/test/java/org/apache/flink/test/runtime/leaderelection/ZooKeeperLeaderElectionITCase.java

@@ -23,6 +23,7 @@ import akka.actor.Kill;
 import akka.actor.PoisonPill;
 import org.apache.curator.test.TestingServer;
 import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.configuration.AkkaOptions;
 import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.runtime.akka.AkkaUtils;
@@ -151,7 +152,7 @@ public class ZooKeeperLeaderElectionITCase extends TestLogger {
 
 		// we "effectively" disable the automatic RecoverAllJobs message and sent it manually to make
 		// sure that all TMs have registered to the JM prior to issueing the RecoverAllJobs message
-		configuration.setString(ConfigConstants.AKKA_ASK_TIMEOUT, AkkaUtils.INF_TIMEOUT().toString());
+		configuration.setString(AkkaOptions.ASK_TIMEOUT, AkkaUtils.INF_TIMEOUT().toString());
 
 		Tasks.BlockingOnceReceiver$.MODULE$.blocking_$eq(true);
 

flink-yarn/src/main/java/org/apache/flink/yarn/YarnTaskExecutorRunner.java

@@ -18,6 +18,7 @@
 
 package org.apache.flink.yarn;
 
+import org.apache.flink.configuration.AkkaOptions;
 import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.GlobalConfiguration;
@@ -127,7 +128,7 @@ public class YarnTaskExecutorRunner {
 			}
 
 			// tell akka to die in case of an error
-			configuration.setBoolean(ConfigConstants.AKKA_JVM_EXIT_ON_FATAL_ERROR, true);
+			configuration.setBoolean(AkkaOptions.JVM_EXIT_ON_FATAL_ERROR, true);
 
 			String keytabPath = null;
 			if(remoteKeytabPath != null) {

flink-yarn/src/main/java/org/apache/flink/yarn/YarnTaskManagerRunner.java

@@ -23,6 +23,7 @@ import java.io.IOException;
 import java.util.Map;
 import java.util.concurrent.Callable;
 
+import org.apache.flink.configuration.AkkaOptions;
 import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.SecurityOptions;
@@ -91,7 +92,7 @@ public class YarnTaskManagerRunner {
 		}
 
 		// tell akka to die in case of an error
-		configuration.setBoolean(ConfigConstants.AKKA_JVM_EXIT_ON_FATAL_ERROR, true);
+		configuration.setBoolean(AkkaOptions.JVM_EXIT_ON_FATAL_ERROR, true);
 
 		String localKeytabPath = null;
 		if(remoteKeytabPath != null) {