[FLINK-12763][runtime] Yarn/MesosResourceManager do not start new wor…

…ker when requested resource profile cannot be satisfied.
mbalassi · Jul 10, 2019 · 887b8d9 · 887b8d9
1 parent 973afee
commit 887b8d9
Show file tree

Hide file tree

Showing 2 changed files with 6 additions and 4 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
@@ -437,6 +437,9 @@ protected void internalDeregisterApplication(
 
  @Override
  public Collection<ResourceProfile> startNewWorker(ResourceProfile resourceProfile) {
+ if (!slotsPerWorker.iterator().next().isMatching(resourceProfile)) {
+ return Collections.emptyList();
+ }
  LOG.info("Starting a new worker.");
  try {
  // generate new workers into persistent state and launch associated actors

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
@@ -310,11 +310,10 @@ protected void internalDeregisterApplication(
 
  @Override
  public Collection<ResourceProfile> startNewWorker(ResourceProfile resourceProfile) {
- Preconditions.checkArgument(
- ResourceProfile.UNKNOWN.equals(resourceProfile),
-  "The YarnResourceManager does not support custom ResourceProfiles yet. It assumes that all containers have the same resources.");
+ if (!slotsPerWorker.iterator().next().isMatching(resourceProfile)) {
+ return Collections.emptyList();
+ }
  requestYarnContainer();
-
  return slotsPerWorker;
  }