You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tr...@apache.org on 2018/02/15 10:42:45 UTC
[02/22] flink git commit: [FLINK-8529] [flip6] Let Yarn entry points
use APPLICATION_MASTER_PORT
[FLINK-8529] [flip6] Let Yarn entry points use APPLICATION_MASTER_PORT
Let all Yarn entry points use the YarnConfigOptions.APPLICATION_MASTER_PORT option
to specify the valid port range for the common RpcService.
This closes #5388.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/42f49ac9
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/42f49ac9
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/42f49ac9
Branch: refs/heads/master
Commit: 42f49ac9172fe5c3ff4cc45ccf5d82142351e4e2
Parents: c7d2a0d
Author: Till Rohrmann <tr...@apache.org>
Authored: Tue Jan 30 15:15:49 2018 +0100
Committer: Till Rohrmann <tr...@apache.org>
Committed: Thu Feb 15 08:44:17 2018 +0100
----------------------------------------------------------------------
.../runtime/entrypoint/ClusterEntrypoint.java | 18 ++++++++++++++++--
.../yarn/entrypoint/YarnJobClusterEntrypoint.java | 6 ++++++
.../entrypoint/YarnSessionClusterEntrypoint.java | 6 ++++++
3 files changed, 28 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/42f49ac9/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java
index 156efdc..dfb1b4c 100755
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java
@@ -172,10 +172,14 @@ public abstract class ClusterEntrypoint implements FatalErrorHandler {
LOG.info("Initializing cluster services.");
final String bindAddress = configuration.getString(JobManagerOptions.ADDRESS);
- // TODO: Add support for port ranges
- final String portRange = String.valueOf(configuration.getInteger(JobManagerOptions.PORT));
+ final String portRange = getRPCPortRange(configuration);
commonRpcService = createRpcService(configuration, bindAddress, portRange);
+
+ // update the configuration used to create the high availability services
+ configuration.setString(JobManagerOptions.ADDRESS, commonRpcService.getAddress());
+ configuration.setInteger(JobManagerOptions.PORT, commonRpcService.getPort());
+
haServices = createHaServices(configuration, commonRpcService.getExecutor());
blobServer = new BlobServer(configuration, haServices.createBlobStore());
blobServer.start();
@@ -188,6 +192,16 @@ public abstract class ClusterEntrypoint implements FatalErrorHandler {
metricRegistry.startQueryService(actorSystem, null);
}
+ /**
+ * Returns the port range for the common {@link RpcService}.
+ *
+ * @param configuration to extract the port range from
+ * @return Port range for the common {@link RpcService}
+ */
+ protected String getRPCPortRange(Configuration configuration) {
+ return String.valueOf(configuration.getInteger(JobManagerOptions.PORT));
+ }
+
protected RpcService createRpcService(
Configuration configuration,
String bindAddress,
http://git-wip-us.apache.org/repos/asf/flink/blob/42f49ac9/flink-yarn/src/main/java/org/apache/flink/yarn/entrypoint/YarnJobClusterEntrypoint.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/entrypoint/YarnJobClusterEntrypoint.java b/flink-yarn/src/main/java/org/apache/flink/yarn/entrypoint/YarnJobClusterEntrypoint.java
index 4a9442a..ca4cb9c 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/entrypoint/YarnJobClusterEntrypoint.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/entrypoint/YarnJobClusterEntrypoint.java
@@ -39,6 +39,7 @@ import org.apache.flink.runtime.util.SignalHandler;
import org.apache.flink.util.FlinkException;
import org.apache.flink.util.Preconditions;
import org.apache.flink.yarn.YarnResourceManager;
+import org.apache.flink.yarn.configuration.YarnConfigOptions;
import org.apache.hadoop.yarn.api.ApplicationConstants;
@@ -75,6 +76,11 @@ public class YarnJobClusterEntrypoint extends JobClusterEntrypoint {
}
@Override
+ protected String getRPCPortRange(Configuration configuration) {
+ return configuration.getString(YarnConfigOptions.APPLICATION_MASTER_PORT);
+ }
+
+ @Override
protected ResourceManager<?> createResourceManager(
Configuration configuration,
ResourceID resourceId,
http://git-wip-us.apache.org/repos/asf/flink/blob/42f49ac9/flink-yarn/src/main/java/org/apache/flink/yarn/entrypoint/YarnSessionClusterEntrypoint.java
----------------------------------------------------------------------
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/entrypoint/YarnSessionClusterEntrypoint.java b/flink-yarn/src/main/java/org/apache/flink/yarn/entrypoint/YarnSessionClusterEntrypoint.java
index 64b4aca..232467a 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/entrypoint/YarnSessionClusterEntrypoint.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/entrypoint/YarnSessionClusterEntrypoint.java
@@ -37,6 +37,7 @@ import org.apache.flink.runtime.util.JvmShutdownSafeguard;
import org.apache.flink.runtime.util.SignalHandler;
import org.apache.flink.util.Preconditions;
import org.apache.flink.yarn.YarnResourceManager;
+import org.apache.flink.yarn.configuration.YarnConfigOptions;
import org.apache.hadoop.yarn.api.ApplicationConstants;
@@ -65,6 +66,11 @@ public class YarnSessionClusterEntrypoint extends SessionClusterEntrypoint {
}
@Override
+ protected String getRPCPortRange(Configuration configuration) {
+ return configuration.getString(YarnConfigOptions.APPLICATION_MASTER_PORT);
+ }
+
+ @Override
protected ResourceManager<?> createResourceManager(
Configuration configuration,
ResourceID resourceId,