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 2019/07/10 09:36:10 UTC
[flink] 09/16: [FLINK-12763][runtime] Yarn/MesosResourceManager set
SlotManager to fail unfulfillable requests on started.
This is an automated email from the ASF dual-hosted git repository.
sewen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit d99877baf701cb5cdf2493465eb7174d8f1c6d4a
Author: Xintong Song <to...@gmail.com>
AuthorDate: Sat Jul 6 18:06:17 2019 +0800
[FLINK-12763][runtime] Yarn/MesosResourceManager set SlotManager to fail unfulfillable requests on started.
---
.../mesos/runtime/clusterframework/MesosResourceManager.java | 1 +
.../org/apache/flink/runtime/resourcemanager/ResourceManager.java | 8 ++++++++
.../src/main/java/org/apache/flink/yarn/YarnResourceManager.java | 1 +
3 files changed, 10 insertions(+)
diff --git a/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosResourceManager.java b/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosResourceManager.java
index 6a53935..ad0ed58 100755
--- a/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosResourceManager.java
+++ b/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosResourceManager.java
@@ -201,6 +201,7 @@ public class MesosResourceManager extends ResourceManager<RegisteredMesosWorkerN
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
index b845d1d..3ee321f 100755
--- 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 @@ public abstract class ResourceManager<WorkerType extends ResourceIDRetrievable>
*/
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
index 2e980d9..1bb29cb 100755
--- 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 class YarnResourceManager extends ResourceManager<YarnWorkerNode> impleme
this.resource = Resource.newInstance(defaultTaskManagerMemoryMB, defaultCpus);
this.slotsPerWorker = updateTaskManagerConfigAndCreateWorkerSlotProfiles(flinkConfig, defaultTaskManagerMemoryMB, numberOfTaskSlots);
+ setFailUnfulfillableRequest(true);
}
protected AMRMClientAsync<AMRMClient.ContainerRequest> createAndStartResourceManagerClient(