Review feedback.

Author: Marcelo Vanzin

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

@@ -55,7 +55,7 @@ private[yarn] class YarnAllocationHandler(
     resourceManager: AMRMProtocol,
     appAttemptId: ApplicationAttemptId,
     args: ApplicationMasterArguments,
-    map: collection.Map[String, collection.Set[SplitInfo]])
+    preferredNodes: collection.Map[String, collection.Set[SplitInfo]])
   extends YarnAllocator with Logging {
 
   // These three are locked on allocatedHostToContainersMap. Complementary data structures
@@ -90,7 +90,8 @@ private[yarn] class YarnAllocationHandler(
   private val maxExecutors = args.numExecutors
   private val executorMemory = args.executorMemory
   private val executorCores = args.executorCores
-  private val (preferredHostToCount, preferredRackToCount) = generateNodeToWeight(conf, map)
+  private val (preferredHostToCount, preferredRackToCount) =
+    generateNodeToWeight(conf, preferredNodes)
 
   def getNumExecutorsRunning: Int = numExecutorsRunning.intValue
 

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

@@ -56,8 +56,7 @@ private[spark] class ApplicationMaster(args: ApplicationMasterArguments,
   private val maxNumExecutorFailures = sparkConf.getInt("spark.yarn.max.executor.failures",
     sparkConf.getInt("spark.yarn.max.worker.failures", math.max(args.numExecutors * 2, 3)))
 
-  private var finished = false
-  private var registered = false
+  @volatile private var finished = false
   private var reporterThread: Thread = _
   private var allocator: YarnAllocator = _
 
@@ -123,7 +122,8 @@ private[spark] class ApplicationMaster(args: ApplicationMasterArguments,
 
   final def finish(status: FinalApplicationStatus, diagnostics: String = "") = synchronized {
     if (!finished) {
-      logInfo(s"Finishing ApplicationMaster with $status")
+      logInfo(s"Finishing ApplicationMaster with $status"  +
+        Option(diagnostics).map(msg => s" (diag message: $msg)").getOrElse(""))
       finished = true
       reporterThread.interrupt()
       reporterThread.join()
@@ -163,6 +163,7 @@ private[spark] class ApplicationMaster(args: ApplicationMasterArguments,
       uiAddress,
       uiHistoryAddress)
 
+    allocator.allocateResources()
     reporterThread = launchReporterThread()
   }
 
@@ -329,18 +330,6 @@ private[spark] class ApplicationMaster(args: ApplicationMasterArguments,
     actorSystem.actorOf(Props(new MonitorActor(driverUrl)), name = "YarnAM")
   }
 
-  private def allocateExecutors() = {
-    logInfo("Requesting" + args.numExecutors + " executors.")
-    try {
-      while (allocator.getNumExecutorsRunning < args.numExecutors && !finished) {
-        checkNumExecutorsFailed()
-        allocator.allocateResources()
-        Thread.sleep(ALLOCATE_HEARTBEAT_INTERVAL)
-      }
-    }
-    logInfo("All executors have launched.")
-  }
-
   private def checkNumExecutorsFailed() = {
     if (allocator.getNumExecutorsFailed >= maxNumExecutorFailures) {
       finish(FinalApplicationStatus.FAILED, "Max number of executor failures reached.")

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

@@ -403,7 +403,7 @@ trait ClientBase extends Logging {
         classOf[ApplicationMaster].getName().replace("ApplicationMaster", "ExecutorLauncher")
       }
     val amArgs =
-      Seq(classOf[ApplicationMaster].getName()) ++ userClass ++
+      Seq(amClass) ++ userClass ++
       (if (args.userJar != null) Seq("--jar", args.userJar) else Nil) ++
       Seq("--executor-memory", args.executorMemory.toString,
         "--executor-cores", args.executorCores.toString,

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

@@ -26,7 +26,8 @@ import org.apache.spark.{SparkConf, SparkContext}
 import org.apache.spark.scheduler.SplitInfo
 
 /**
- * Interface that defines a Yarn RM client.
+ * Interface that defines a Yarn RM client. Abstracts away Yarn version-specific functionality that
+ * is used by Spark's AM.
  */
 trait YarnRMClient {
 

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

@@ -94,7 +94,7 @@ object YarnSparkHadoopUtil {
   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
+  // Note that it is possible for this to change : and ResourceManager will indicate that to us via
   // update response to allocate. But we are punting on handling that for now.
   private val hostToRack = new ConcurrentHashMap[String, String]()
   private val rackToHostSet = new ConcurrentHashMap[String, JSet[String]]()

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

@@ -56,7 +56,7 @@ private[yarn] class YarnAllocationHandler(
     amClient: AMRMClient[ContainerRequest],
     appAttemptId: ApplicationAttemptId,
     args: ApplicationMasterArguments,
-    map: collection.Map[String, collection.Set[SplitInfo]])
+    preferredNodes: collection.Map[String, collection.Set[SplitInfo]])
   extends YarnAllocator with Logging {
 
   // These three are locked on allocatedHostToContainersMap. Complementary data structures
@@ -94,7 +94,8 @@ private[yarn] class YarnAllocationHandler(
   private val maxExecutors = args.numExecutors
   private val executorMemory = args.executorMemory
   private val executorCores = args.executorCores
-  private val (preferredHostToCount, preferredRackToCount) = generateNodeToWeight(conf, map)
+  private val (preferredHostToCount, preferredRackToCount) =
+    generateNodeToWeight(conf, preferredNodes)
 
   override def getNumExecutorsRunning: Int = numExecutorsRunning.intValue