YARN-3547. FairScheduler: Apps that have no resource demand should not participate scheduling. (Xianyin Xin via kasha)

(cherry picked from commit 3ae2a625018bc8cf431aa19da5bf8fe4ef8c1ad4)

hadoop-yarn-project/CHANGES.txt

@@ -243,6 +243,9 @@ Release 2.8.0 - UNRELEASED
     YARN-3006. Improve the error message when attempting manual failover with 
     auto-failover enabled. (Akira AJISAKA via wangda)
 
+    YARN-3547. FairScheduler: Apps that have no resource demand should not participate 
+    scheduling. (Xianyin Xin via kasha)
+
   BUG FIXES
 
     YARN-3197. Confusing log generated by CapacityScheduler. (Varun Saxena 

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSLeafQueue.java

@@ -26,6 +26,7 @@ import java.util.List;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReadWriteLock;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.TreeSet;
 
 import com.google.common.annotations.VisibleForTesting;
 import org.apache.commons.logging.Log;
@@ -314,35 +315,33 @@ public class FSLeafQueue extends FSQueue {
       return assigned;
     }
 
-    Comparator<Schedulable> comparator = policy.getComparator();
-    writeLock.lock();
-    try {
-      Collections.sort(runnableApps, comparator);
-    } finally {
-      writeLock.unlock();
-    }
-    // Release write lock here for better performance and avoiding deadlocks.
-    // runnableApps can be in unsorted state because of this section,
-    // but we can accept it in practice since the probability is low.
+    // Apps that have resource demands.
+    TreeSet<FSAppAttempt> pendingForResourceApps =
+        new TreeSet<FSAppAttempt>(policy.getComparator());
     readLock.lock();
     try {
-      for (FSAppAttempt sched : runnableApps) {
-        if (SchedulerAppUtils.isBlacklisted(sched, node, LOG)) {
-          continue;
-        }
-
-        assigned = sched.assignContainer(node);
-        if (!assigned.equals(Resources.none())) {
-          if (LOG.isDebugEnabled()) {
-            LOG.debug("Assigned container in queue:" + getName() + " " +
-                "container:" + assigned);
-          }
-          break;
+      for (FSAppAttempt app : runnableApps) {
+        Resource pending = app.getAppAttemptResourceUsage().getPending();
+        if (!pending.equals(Resources.none())) {
+          pendingForResourceApps.add(app);
         }
       }
     } finally {
       readLock.unlock();
     }
+    for (FSAppAttempt sched : pendingForResourceApps) {
+      if (SchedulerAppUtils.isBlacklisted(sched, node, LOG)) {
+        continue;
+      }
+      assigned = sched.assignContainer(node);
+      if (!assigned.equals(Resources.none())) {
+        if (LOG.isDebugEnabled()) {
+          LOG.debug("Assigned container in queue:" + getName() + " " +
+              "container:" + assigned);
+        }
+        break;
+      }
+    }
     return assigned;
   }