send empty resource requests when we aren't asking for containers

Author: tgravescs

yarn/alpha/src/main/scala/org/apache/spark/deploy/yarn/YarnAllocationHandler.scala

@@ -50,12 +50,13 @@ private[yarn] class YarnAllocationHandler(
   override protected def allocateContainers(count: Int): YarnAllocateResponse = {
     var resourceRequests: List[ResourceRequest] = null
 
-    // default.
-    if (count <= 0 || preferredHostToCount.isEmpty) {
-      logDebug("numExecutors: " + count + ", host preferences: " +
-        preferredHostToCount.isEmpty)
-      resourceRequests = List(createResourceRequest(
-        AllocationType.ANY, null, count, YarnSparkHadoopUtil.RM_REQUEST_PRIORITY))
+    logDebug("numExecutors: " + count)
+    if (count <= 0) {
+      resourceRequests = List()
+    } else if (preferredHostToCount.isEmpty) {
+        logDebug("host preferences is empty")
+        resourceRequests = List(createResourceRequest(
+          AllocationType.ANY, null, count, YarnSparkHadoopUtil.RM_REQUEST_PRIORITY))
     } else {
       // request for all hosts in preferred nodes and for numExecutors -
       // candidates.size, request by default allocation policy.

yarn/stable/src/main/scala/org/apache/spark/deploy/yarn/YarnAllocationHandler.scala

@@ -87,9 +87,11 @@ private[yarn] class YarnAllocationHandler(
 
   private def addResourceRequests(numExecutors: Int) {
     val containerRequests: List[ContainerRequest] =
-      if (numExecutors <= 0 || preferredHostToCount.isEmpty) {
-        logDebug("numExecutors: " + numExecutors + ", host preferences: " +
-          preferredHostToCount.isEmpty)
+      if (numExecutors <= 0) {
+        logDebug("numExecutors: " + numExecutors)
+        List()
+      } else if (preferredHostToCount.isEmpty) {
+        logDebug("host preferences is empty")
         createResourceRequests(
           AllocationType.ANY,
           resource = null,