[FLINK-10425][runtime] Use TaskManagerOptions#HOST

tongcheng-elong · Oct 24, 2018 · 5fa36c9 · 5fa36c9
1 parent 5953bb6
commit 5fa36c9
Show file tree

Hide file tree

Showing 9 changed files with 13 additions and 19 deletions.
diff --git a/.../src/main/java/org/apache/flink/mesos/runtime/clusterframework/LaunchableMesosWorker.java b/.../src/main/java/org/apache/flink/mesos/runtime/clusterframework/LaunchableMesosWorker.java
@@ -18,8 +18,8 @@
 
 package org.apache.flink.mesos.runtime.clusterframework;
 
-import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.TaskManagerOptions;
 import org.apache.flink.mesos.Utils;
 import org.apache.flink.mesos.scheduler.LaunchableTask;
 import org.apache.flink.mesos.util.MesosArtifactResolver;
@@ -232,7 +232,7 @@ public Protos.TaskInfo launch(Protos.SlaveID slaveId, MesosResourceAllocation al
  .matcher(taskManagerHostnameOption.get())
  .replaceAll(Matcher.quoteReplacement(taskID.getValue()));
 
- dynamicProperties.setString(ConfigConstants.TASK_MANAGER_HOSTNAME_KEY, taskManagerHostname);
+ dynamicProperties.setString(TaskManagerOptions.HOST, taskManagerHostname);
  }
 
  // take needed ports for the TM

diff --git a/...-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniClusterConfiguration.java b/...-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniClusterConfiguration.java
@@ -19,7 +19,6 @@
 package org.apache.flink.runtime.minicluster;
 
 import org.apache.flink.api.common.time.Time;
-import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.JobManagerOptions;
 import org.apache.flink.configuration.RestOptions;
@@ -85,7 +84,7 @@ public String getJobManagerBindAddress() {
  public String getTaskManagerBindAddress() {
  return commonBindAddress != null ?
  commonBindAddress :
- configuration.getString(ConfigConstants.TASK_MANAGER_HOSTNAME_KEY, "localhost");
+ configuration.getString(TaskManagerOptions.HOST, "localhost");
  }
 
  public String getResourceManagerBindAddress() {

diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
@@ -19,7 +19,6 @@
 package org.apache.flink.runtime.taskexecutor;
 
 import org.apache.flink.api.common.time.Time;
-import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.ConfigurationUtils;
 import org.apache.flink.configuration.GlobalConfiguration;
@@ -405,7 +404,7 @@ public static RpcService createRpcService(
  checkNotNull(configuration);
  checkNotNull(haServices);
 
- String taskManagerHostname = configuration.getString(ConfigConstants.TASK_MANAGER_HOSTNAME_KEY, null);
+ String taskManagerHostname = configuration.getString(TaskManagerOptions.HOST);
 
  if (taskManagerHostname != null) {
  LOG.info("Using configured hostname/address for TaskManager: {}.", taskManagerHostname);

diff --git a/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala b/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
@@ -1735,8 +1735,7 @@ object TaskManager {
  highAvailabilityServices: HighAvailabilityServices)
  : (String, java.util.Iterator[Integer]) = {
 
- var taskManagerHostname = configuration.getString(
- ConfigConstants.TASK_MANAGER_HOSTNAME_KEY, null)
+ var taskManagerHostname = configuration.getString(TaskManagerOptions.HOST)
 
  if (taskManagerHostname != null) {
  LOG.info("Using configured hostname/address for TaskManager: " + taskManagerHostname)

diff --git a/...time/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerConfigurationTest.java b/...time/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerConfigurationTest.java
@@ -19,7 +19,6 @@
 package org.apache.flink.runtime.taskmanager;
 
 import net.jcip.annotations.NotThreadSafe;
-import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.GlobalConfiguration;
 import org.apache.flink.configuration.IllegalConfigurationException;
@@ -65,7 +64,7 @@ public void testUsePreconfiguredNetworkInterface() throws Exception {
  final String TEST_HOST_NAME = "testhostname";
 
  Configuration config = new Configuration();
- config.setString(ConfigConstants.TASK_MANAGER_HOSTNAME_KEY, TEST_HOST_NAME);
+ config.setString(TaskManagerOptions.HOST, TEST_HOST_NAME);
  config.setString(JobManagerOptions.ADDRESS, "localhost");
  config.setInteger(JobManagerOptions.PORT, 7891);
 
@@ -89,7 +88,7 @@ public void testUsePreconfiguredNetworkInterface() throws Exception {
  public void testActorSystemPortConfig() throws Exception {
  // config with pre-configured hostname to speed up tests (no interface selection)
  Configuration config = new Configuration();
- config.setString(ConfigConstants.TASK_MANAGER_HOSTNAME_KEY, "localhost");
+ config.setString(TaskManagerOptions.HOST, "localhost");
  config.setString(JobManagerOptions.ADDRESS, "localhost");
  config.setInteger(JobManagerOptions.PORT, 7891);
 

diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerStartupTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerStartupTest.java
@@ -18,7 +18,6 @@
 
 package org.apache.flink.runtime.taskmanager;
 
-import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.CoreOptions;
 import org.apache.flink.configuration.IllegalConfigurationException;
@@ -247,7 +246,7 @@ public void testStartupWhenNetworkStackFailsToInitialize() throws Exception {
  blocker = new ServerSocket(0, 50, InetAddress.getByName("localhost"));
 
  final Configuration cfg = new Configuration();
- cfg.setString(ConfigConstants.TASK_MANAGER_HOSTNAME_KEY, "localhost");
+ cfg.setString(TaskManagerOptions.HOST, "localhost");
  cfg.setInteger(TaskManagerOptions.DATA_PORT, blocker.getLocalPort());
  cfg.setString(TaskManagerOptions.MANAGED_MEMORY_SIZE, "1m");
  ActorSystem actorSystem = AkkaUtils.createLocalActorSystem(cfg);

diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/akka/AkkaSslITCase.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/akka/AkkaSslITCase.scala
@@ -50,7 +50,7 @@ class AkkaSslITCase(_system: ActorSystem)
 
  val config = new Configuration()
  config.setString(JobManagerOptions.ADDRESS, "127.0.0.1")
- config.setString(ConfigConstants.TASK_MANAGER_HOSTNAME_KEY, "127.0.0.1")
+ config.setString(TaskManagerOptions.HOST, "127.0.0.1")
  config.setInteger(TaskManagerOptions.NUM_TASK_SLOTS, 1)
  config.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, 1)
 
@@ -77,7 +77,7 @@ class AkkaSslITCase(_system: ActorSystem)
 
  val config = new Configuration()
  config.setString(JobManagerOptions.ADDRESS, "127.0.0.1")
- config.setString(ConfigConstants.TASK_MANAGER_HOSTNAME_KEY, "127.0.0.1")
+ config.setString(TaskManagerOptions.HOST, "127.0.0.1")
  config.setInteger(TaskManagerOptions.NUM_TASK_SLOTS, 1)
  config.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, 1)
 

diff --git a/flink-tests/src/test/java/org/apache/flink/test/runtime/IPv6HostnamesITCase.java b/flink-tests/src/test/java/org/apache/flink/test/runtime/IPv6HostnamesITCase.java
@@ -22,7 +22,6 @@
 import org.apache.flink.api.java.DataSet;
 import org.apache.flink.api.java.ExecutionEnvironment;
 import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.JobManagerOptions;
 import org.apache.flink.configuration.TaskManagerOptions;
@@ -77,7 +76,7 @@ private Configuration getConfiguration() {
 
  Configuration config = new Configuration();
  config.setString(JobManagerOptions.ADDRESS, addressString);
- config.setString(ConfigConstants.TASK_MANAGER_HOSTNAME_KEY, addressString);
+ config.setString(TaskManagerOptions.HOST, addressString);
  config.setString(TaskManagerOptions.MANAGED_MEMORY_SIZE, "16m");
  return config;
  }

diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnTaskExecutorRunner.java b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnTaskExecutorRunner.java
@@ -19,10 +19,10 @@
 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;
 import org.apache.flink.configuration.SecurityOptions;
+import org.apache.flink.configuration.TaskManagerOptions;
 import org.apache.flink.core.fs.FileSystem;
 import org.apache.flink.runtime.clusterframework.BootstrapTools;
 import org.apache.flink.runtime.clusterframework.types.ResourceID;
@@ -133,7 +133,7 @@ private static void run(String[] args) {
  // use the hostname passed by job manager
  final String taskExecutorHostname = ENV.get(YarnResourceManager.ENV_FLINK_NODE_ID);
  if (taskExecutorHostname != null) {
- configuration.setString(ConfigConstants.TASK_MANAGER_HOSTNAME_KEY, taskExecutorHostname);
+ configuration.setString(TaskManagerOptions.HOST, taskExecutorHostname);
  }
 
  SecurityUtils.install(sc);