[FLINK-12763][runtime] Yarn/MesosResourceManager set SlotManager to f…

…ail unfulfillable requests on started.
vijikarthi · Jul 10, 2019 · d99877b · d99877b
1 parent 887b8d9
commit d99877b
Show file tree

Hide file tree

Showing 3 changed files with 10 additions and 0 deletions.
diff --git a/...s/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosResourceManager.java b/...s/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosResourceManager.java
@@ -201,6 +201,7 @@ public MesosResourceManager(
 		final ContaineredTaskManagerParameters containeredTaskManagerParameters = taskManagerParameters.containeredParameters();
 		this.slotsPerWorker = updateTaskManagerConfigAndCreateWorkerSlotProfiles(
 			flinkConfig, containeredTaskManagerParameters.taskManagerTotalMemoryMB(), containeredTaskManagerParameters.numSlots());
+		setFailUnfulfillableRequest(true);
 	}
 
 	protected ActorRef createSelfActor() {

diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java
@@ -1062,6 +1062,14 @@ protected abstract void internalDeregisterApplication(
 	 */
 	public abstract boolean stopWorker(WorkerType worker);
 
+	/**
+	 * Set {@link SlotManager} whether to fail unfulfillable slot requests.
+	 * @param failUnfulfillableRequest whether to fail unfulfillable requests
+	 */
+	protected void setFailUnfulfillableRequest(boolean failUnfulfillableRequest) {
+		slotManager.setFailUnfulfillableRequest(failUnfulfillableRequest);
+	}
+
 	// ------------------------------------------------------------------------
 	//  Static utility classes
 	// ------------------------------------------------------------------------

diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnResourceManager.java
@@ -185,6 +185,7 @@ public YarnResourceManager(
 		this.resource = Resource.newInstance(defaultTaskManagerMemoryMB, defaultCpus);
 
 		this.slotsPerWorker = updateTaskManagerConfigAndCreateWorkerSlotProfiles(flinkConfig, defaultTaskManagerMemoryMB, numberOfTaskSlots);
+		setFailUnfulfillableRequest(true);
 	}
 
 	protected AMRMClientAsync<AMRMClient.ContainerRequest> createAndStartResourceManagerClient(