You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2018/10/24 10:26:15 UTC
[GitHub] zentol closed pull request #6913: [FLINK-10425] taskmanager.host is
not respected
zentol closed pull request #6913: [FLINK-10425] taskmanager.host is not respected
URL: https://github.com/apache/flink/pull/6913
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/LaunchableMesosWorker.java b/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/LaunchableMesosWorker.java
index bb15aee0a7f..84ec2229a2a 100644
--- a/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/LaunchableMesosWorker.java
+++ b/flink-mesos/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 String toString() {
.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/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniClusterConfiguration.java b/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniClusterConfiguration.java
index f728f24111d..ca5d01d9734 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniClusterConfiguration.java
+++ b/flink-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
index 1f3818a9eeb..e06be5329cf 100644
--- 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
index 2008ad87d56..0d2ff1b915a 100644
--- 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/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerConfigurationTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerConfigurationTest.java
index 5f2de3390d4..c4726a9f54f 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskManagerConfigurationTest.java
+++ b/flink-runtime/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
index 158d3811fc6..986b76852a2 100644
--- 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
index ab62d3eeef7..a6b602892b3 100644
--- 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
index 003ea571287..c1775573765 100644
--- 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
index a419bb89869..0034b635628 100644
--- 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);
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services