[FLINK-12152] Make the vcore that Application Master used configurabl…

…e for Flink on YARN This closes apache#8438.
Jin-H · May 26, 2019 · 4f558e4 · 4f558e4
1 parent 6af4737
commit 4f558e4
Show file tree

Hide file tree

Showing 3 changed files with 22 additions and 1 deletion.
diff --git a/docs/_includes/generated/yarn_config_configuration.html b/docs/_includes/generated/yarn_config_configuration.html
@@ -32,6 +32,11 @@
             <td style="word-wrap: break-word;">-1</td>
             <td>The port where the application master RPC system is listening.</td>
         </tr>
+        <tr>
+            <td><h5>yarn.appmaster.vcores</h5></td>
+            <td style="word-wrap: break-word;">1</td>
+            <td>The number of virtual cores (vcores) used by YARN application master.</td>
+        </tr>
         <tr>
             <td><h5>yarn.containers.vcores</h5></td>
             <td style="word-wrap: break-word;">-1</td>

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
@@ -242,6 +242,14 @@ private void isReadyForDeployment(ClusterSpecification clusterSpecification) thr
 			throw new YarnDeploymentException("Couldn't get cluster description, please check on the YarnConfiguration", e);
 		}
 
+		int configuredAmVcores = flinkConfiguration.getInteger(YarnConfigOptions.APP_MASTER_VCORES);
+		if (configuredAmVcores > numYarnMaxVcores) {
+			throw new IllegalConfigurationException(
+				String.format("The number of requested virtual cores for application master %d" +
+						" exceeds the maximum number of virtual cores %d available in the Yarn Cluster.",
+					configuredAmVcores, numYarnMaxVcores));
+		}
+
 		int configuredVcores = flinkConfiguration.getInteger(YarnConfigOptions.VCORES, clusterSpecification.getSlotsPerTaskManager());
 		// don't configure more than the maximum configured number of vcores
 		if (configuredVcores > numYarnMaxVcores) {
@@ -971,7 +979,7 @@ public ApplicationReport startAppMaster(
 		// Set up resource type requirements for ApplicationMaster
 		Resource capability = Records.newRecord(Resource.class);
 		capability.setMemory(clusterSpecification.getMasterMemoryMB());
-		capability.setVirtualCores(1);
+		capability.setVirtualCores(flinkConfiguration.getInteger(YarnConfigOptions.APP_MASTER_VCORES));
 
 		final String customApplicationName = customName != null ? customName : applicationName;
 

diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/configuration/YarnConfigOptions.java b/flink-yarn/src/main/java/org/apache/flink/yarn/configuration/YarnConfigOptions.java
@@ -49,6 +49,14 @@ public class YarnConfigOptions {
 			.defaultValue(-1)
 			.withDescription("The port where the application master RPC system is listening.");
 
+	/**
+	 * The vcores used by YARN application master.
+	 */
+	public static final ConfigOption<Integer> APP_MASTER_VCORES =
+		key("yarn.appmaster.vcores")
+		.defaultValue(1)
+		.withDescription("The number of virtual cores (vcores) used by YARN application master.");
+
 	/**
 	 * Defines whether user-jars are included in the system class path for per-job-clusters as well as their positioning
 	 * in the path. They can be positioned at the beginning ("FIRST"), at the end ("LAST"), or be positioned based on