You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2015/03/02 19:43:57 UTC
[4/4] flink git commit: [FLINK-1608] [taskmanager] Hostname/address
for TaskManager can be specified in the configuration
[FLINK-1608] [taskmanager] Hostname/address for TaskManager can be specified in the configuration
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/933609f0
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/933609f0
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/933609f0
Branch: refs/heads/master
Commit: 933609f0f6a183874feeccc35b5eae47cbf57f5d
Parents: 086209c
Author: Stephan Ewen <se...@apache.org>
Authored: Tue Feb 24 22:04:09 2015 +0100
Committer: Stephan Ewen <se...@apache.org>
Committed: Mon Mar 2 18:36:47 2015 +0100
----------------------------------------------------------------------
.../flink/configuration/ConfigConstants.java | 5 ++++
.../flink/runtime/taskmanager/TaskManager.scala | 30 +++++++++++++-------
2 files changed, 24 insertions(+), 11 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/933609f0/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java b/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
index 767648a..245bebe 100644
--- a/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
+++ b/flink-core/src/main/java/org/apache/flink/configuration/ConfigConstants.java
@@ -81,6 +81,11 @@ public final class ConfigConstants {
public static final String LIBRARY_CACHE_MANAGER_CLEANUP_INTERVAL = "library-cache-manager.cleanup.interval";
/**
+ * The config parameter defining the task manager's hostname.
+ */
+ public static final String TASK_MANAGER_HOSTNAME_KEY = "taskmanager.hostname";
+
+ /**
* The config parameter defining the task manager's IPC port from the configuration.
*/
public static final String TASK_MANAGER_IPC_PORT_KEY = "taskmanager.rpc.port";
http://git-wip-us.apache.org/repos/asf/flink/blob/933609f0/flink-runtime/src/main/scala/org/apache/flink/runtime/taskmanager/TaskManager.scala
----------------------------------------------------------------------
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 7bfa370..7011eb4 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
@@ -834,19 +834,27 @@ object TaskManager {
val (jobManagerHostname, jobManagerPort) = getAndCheckJobManagerAddress(configuration)
- // try to find out the hostname of the interface from which the TaskManager
- // can connect to the JobManager. This involves a reverse name lookup
- LOG.info("Trying to determine network interface and address/hostname to use")
- val jobManagerAddress = new InetSocketAddress(jobManagerHostname, jobManagerPort)
- val taskManagerHostname = try {
- NetUtils.resolveAddress(jobManagerAddress).getHostName()
- }
- catch {
- case t: Throwable => throw new Exception("TaskManager cannot find a network interface " +
- "that can communicate with the JobManager (" + jobManagerAddress + ")", t)
+ var taskManagerHostname = configuration.getString(
+ ConfigConstants.TASK_MANAGER_HOSTNAME_KEY, null)
+
+ if (taskManagerHostname != null) {
+ LOG.info("Using configured hostname/address for TaskManager: " + taskManagerHostname)
}
+ else {
+ // try to find out the hostname of the interface from which the TaskManager
+ // can connect to the JobManager. This involves a reverse name lookup
+ LOG.info("Trying to select the network interface and address/hostname to use")
+ val jobManagerAddress = new InetSocketAddress(jobManagerHostname, jobManagerPort)
+ taskManagerHostname = try {
+ NetUtils.resolveAddress(jobManagerAddress).getHostName()
+ }
+ catch {
+ case t: Throwable => throw new Exception("TaskManager cannot find a network interface " +
+ "that can communicate with the JobManager (" + jobManagerAddress + ")", t)
+ }
- LOG.info("TaskManager will use hostname/address '{}' for communication.", taskManagerHostname)
+ LOG.info("TaskManager will use hostname/address '{}' for communication.", taskManagerHostname)
+ }
// if no task manager port has been configured, use 0 (system will pick any free port)
val actorSystemPort = configuration.getInteger(ConfigConstants.TASK_MANAGER_IPC_PORT_KEY, 0)