Cleanup a couple more constants.

Author: Marcelo Vanzin

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

@@ -58,11 +58,6 @@ private[yarn] class YarnAllocationHandler(
     map: collection.Map[String, collection.Set[SplitInfo]])
   extends YarnAllocator with Logging {
 
-  private val ANY_HOST = "*"
-  // All requests are issued with same priority : we do not (yet) have any distinction between
-  // request types (like map/reduce in hadoop for example)
-  private val PRIORITY = 1
-
   // These three are locked on allocatedHostToContainersMap. Complementary data structures
   // allocatedHostToContainersMap : containers which are running : host, Set<containerid>
   // allocatedContainerToHostMap: container to host mapping.
@@ -362,7 +357,7 @@ private[yarn] class YarnAllocationHandler(
     for (container <- hostContainers) {
       val candidateHost = container.getHostName
       val candidateNumContainers = container.getNumContainers
-      assert(ANY_HOST != candidateHost)
+      assert(YarnSparkHadoopUtil.ANY_HOST != candidateHost)
 
       val rack = YarnSparkHadoopUtil.lookupRack(conf, candidateHost)
       if (rack != null) {
@@ -376,7 +371,8 @@ private[yarn] class YarnAllocationHandler(
       new ArrayBuffer[ResourceRequest](rackToCounts.size)
     for ((rack, count) <- rackToCounts){
       requestedContainers +=
-        createResourceRequest(AllocationType.RACK, rack, count, PRIORITY)
+        createResourceRequest(AllocationType.RACK, rack, count,
+          YarnSparkHadoopUtil.RM_REQUEST_PRIORITY)
     }
 
     requestedContainers.toList
@@ -407,7 +403,7 @@ private[yarn] class YarnAllocationHandler(
       logDebug("numExecutors: " + numExecutors + ", host preferences: " +
         preferredHostToCount.isEmpty)
       resourceRequests = List(createResourceRequest(
-        AllocationType.ANY, null, numExecutors, PRIORITY))
+        AllocationType.ANY, null, numExecutors, YarnSparkHadoopUtil.RM_REQUEST_PRIORITY))
     } else {
       // request for all hosts in preferred nodes and for numExecutors -
       // candidates.size, request by default allocation policy.
@@ -421,7 +417,7 @@ private[yarn] class YarnAllocationHandler(
             AllocationType.HOST,
             candidateHost,
             requiredCount,
-            PRIORITY)
+            YarnSparkHadoopUtil.RM_REQUEST_PRIORITY)
         }
       }
       val rackContainerRequests: List[ResourceRequest] = createRackResourceRequests(
@@ -431,7 +427,7 @@ private[yarn] class YarnAllocationHandler(
         AllocationType.ANY,
         resource = null,
         numExecutors,
-        PRIORITY)
+        YarnSparkHadoopUtil.RM_REQUEST_PRIORITY)
 
       val containerRequests: ArrayBuffer[ResourceRequest] = new ArrayBuffer[ResourceRequest](
         hostContainerRequests.size + rackContainerRequests.size + 1)
@@ -481,7 +477,7 @@ private[yarn] class YarnAllocationHandler(
     // There must be a third request - which is ANY : that will be specially handled.
     requestType match {
       case AllocationType.HOST => {
-        assert(ANY_HOST != resource)
+        assert(YarnSparkHadoopUtil.ANY_HOST != resource)
         val hostname = resource
         val nodeLocal = createResourceRequestImpl(hostname, numExecutors, priority)
 
@@ -495,7 +491,7 @@ private[yarn] class YarnAllocationHandler(
         createResourceRequestImpl(rack, numExecutors, priority)
       }
       case AllocationType.ANY => createResourceRequestImpl(
-        ANY_HOST, numExecutors, priority)
+        YarnSparkHadoopUtil.ANY_HOST, numExecutors, priority)
       case _ => throw new IllegalArgumentException(
         "Unexpected/unsupported request type: " + requestType)
     }

yarn/common/src/main/scala/org/apache/spark/deploy/yarn/YarnSparkHadoopUtil.scala

@@ -87,6 +87,12 @@ object YarnSparkHadoopUtil {
   // Additional memory overhead - in mb.
   val DEFAULT_MEMORY_OVERHEAD = 384
 
+  val ANY_HOST = "*"
+
+  // All RM requests are issued with same priority : we do not (yet) have any distinction between
+  // request types (like map/reduce in hadoop for example)
+  val RM_REQUEST_PRIORITY = 1
+
   // Host to rack map - saved from allocation requests. We are expecting this not to change.
   // Note that it is possible for this to change : and ResurceManager will indicate that to us via
   // update response to allocate. But we are punting on handling that for now.

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

@@ -59,11 +59,6 @@ private[yarn] class YarnAllocationHandler(
     map: collection.Map[String, collection.Set[SplitInfo]])
   extends YarnAllocator with Logging {
 
-  private val ANY_HOST = "*"
-  // All requests are issued with same priority : we do not (yet) have any distinction between
-  // request types (like map/reduce in hadoop for example)
-  private val PRIORITY = 1
-
   // These three are locked on allocatedHostToContainersMap. Complementary data structures
   // allocatedHostToContainersMap : containers which are running : host, Set<containerid>
   // allocatedContainerToHostMap: container to host mapping.
@@ -390,7 +385,7 @@ private[yarn] class YarnAllocationHandler(
 
     for (container <- hostContainers) {
       val candidateHost = container.getNodes.last
-      assert(ANY_HOST != candidateHost)
+      assert(YarnSparkHadoopUtil.ANY_HOST != candidateHost)
 
       val rack = YarnSparkHadoopUtil.lookupRack(conf, candidateHost)
       if (rack != null) {
@@ -406,7 +401,7 @@ private[yarn] class YarnAllocationHandler(
         AllocationType.RACK,
         rack,
         count,
-        PRIORITY)
+        YarnSparkHadoopUtil.RM_REQUEST_PRIORITY)
     }
 
     requestedContainers
@@ -437,7 +432,7 @@ private[yarn] class YarnAllocationHandler(
           AllocationType.ANY,
           resource = null,
           numExecutors,
-          PRIORITY).toList
+          YarnSparkHadoopUtil.RM_REQUEST_PRIORITY).toList
       } else {
         // Request for all hosts in preferred nodes and for numExecutors -
         // candidates.size, request by default allocation policy.
@@ -450,7 +445,7 @@ private[yarn] class YarnAllocationHandler(
               AllocationType.HOST,
               candidateHost,
               requiredCount,
-              PRIORITY)
+              YarnSparkHadoopUtil.RM_REQUEST_PRIORITY)
           }
         }
         val rackContainerRequests: List[ContainerRequest] = createRackResourceRequests(
@@ -460,7 +455,7 @@ private[yarn] class YarnAllocationHandler(
           AllocationType.ANY,
           resource = null,
           numExecutors,
-          PRIORITY)
+          YarnSparkHadoopUtil.RM_REQUEST_PRIORITY)
 
         val containerRequestBuffer = new ArrayBuffer[ContainerRequest](
           hostContainerRequests.size + rackContainerRequests.size() + anyContainerRequests.size)
@@ -509,7 +504,7 @@ private[yarn] class YarnAllocationHandler(
     // There must be a third request, which is ANY. That will be specially handled.
     requestType match {
       case AllocationType.HOST => {
-        assert(ANY_HOST != resource)
+        assert(YarnSparkHadoopUtil.ANY_HOST != resource)
         val hostname = resource
         val nodeLocal = constructContainerRequests(
           Array(hostname),