Repository: flink Updated Branches: refs/heads/master 9b7e42924 -> 3a4e89643
[hotfix] [yarn] Write number of slots to configuration Project: http://git-wip-us.apache.org/repos/asf/flink/repo Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/4b89b5d0 Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/4b89b5d0 Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/4b89b5d0 Branch: refs/heads/master Commit: 4b89b5d0abe4684dbeacd9a01fe3297621050eb7 Parents: 075f5b6 Author: Till Rohrmann <[email protected]> Authored: Wed Feb 7 18:21:06 2018 +0100 Committer: Till Rohrmann <[email protected]> Committed: Sun Feb 18 10:12:52 2018 +0100 ---------------------------------------------------------------------- .../org/apache/flink/yarn/AbstractYarnClusterDescriptor.java | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/flink/blob/4b89b5d0/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java ---------------------------------------------------------------------- diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java b/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java index 583ee54..d3fce4c 100644 --- a/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java +++ b/flink-yarn/src/main/java/org/apache/flink/yarn/AbstractYarnClusterDescriptor.java @@ -482,7 +482,7 @@ public abstract class AbstractYarnClusterDescriptor implements ClusterDescriptor flinkConfiguration.setString(ClusterEntrypoint.EXECUTION_MODE, executionMode.toString()); ApplicationReport report = startAppMaster( - flinkConfiguration, + new Configuration(flinkConfiguration), yarnClusterEntrypoint, jobGraph, yarnClient, @@ -794,6 +794,10 @@ public abstract class AbstractYarnClusterDescriptor implements ClusterDescriptor homeDir, ""); + configuration.setInteger( + ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, + clusterSpecification.getSlotsPerTaskManager()); + // Upload the flink configuration // write out configuration file File tmpConfigurationFile = File.createTempFile(appId + "-flink-conf.yaml", null);
