Heavily add documentation to Client* classes + various clean-ups

Author: andrewor14

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

@@ -48,9 +48,14 @@ private[spark] class Client(
 
   val yarnConf: YarnConfiguration = new YarnConfiguration(hadoopConf)
 
+  /* ------------------------------------------------------------------------------------- *
+   | The following methods have much in common in the stable and alpha versions of Client, |
+   | but cannot be implemented in the parent trait due to subtle API differences across    |
+   | hadoop versions.                                                                      |
+   * ------------------------------------------------------------------------------------- */
+
   /** Submit an application running our ApplicationMaster to the ResourceManager. */
   override def submitApplication(): ApplicationId = {
-    // Initialize and start the client service.
     init(yarnConf)
     start()
 
@@ -64,10 +69,8 @@ private[spark] class Client(
     // Verify whether the cluster has enough resources for our AM.
     verifyClusterResources(newAppResponse)
 
-    // Set up ContainerLaunchContext to launch our AM container.
+    // Set up the appropriate contexts to launch our AM.
     val containerContext = createContainerLaunchContext(newAppResponse)
-
-    // Set up ApplicationSubmissionContext to submit our AM.
     val appContext = createApplicationSubmissionContext(appId, containerContext)
 
     // Finally, submit and monitor the application.
@@ -77,7 +80,9 @@ private[spark] class Client(
   }
 
   /**
-   *
+   * Set up a context for launching our ApplicationMaster container.
+   * In the Yarn alpha API, the memory requirements of this container must be set in
+   * the ContainerLaunchContext instead of the ApplicationSubmissionContext.
    */
   override def createContainerLaunchContext(newAppResponse: GetNewApplicationResponse)
       : ContainerLaunchContext = {
@@ -88,9 +93,7 @@ private[spark] class Client(
     containerContext
   }
 
-  /**
-   *
-   */
+  /** Set up the context for submitting our ApplicationMaster. */
   def createApplicationSubmissionContext(
       appId: ApplicationId,
       containerContext: ContainerLaunchContext): ApplicationSubmissionContext = {
@@ -104,25 +107,30 @@ private[spark] class Client(
   }
 
   /**
-   *
+   * Set up security tokens for launching our ApplicationMaster container.
+   * ContainerLaunchContext#setContainerTokens is renamed `setTokens` in the stable API.
    */
-  override def getAMMemory(newApp: GetNewApplicationResponse): Int = {
-    val minResMemory = newApp.getMinimumResourceCapability().getMemory()
-    val amMemory = ((args.amMemory / minResMemory) * minResMemory) +
-      ((if ((args.amMemory % minResMemory) == 0) 0 else minResMemory) - memoryOverhead)
-    amMemory
-  }
-
-  /** */
   override def setupSecurityToken(amContainer: ContainerLaunchContext): Unit = {
     val dob = new DataOutputBuffer()
     credentials.writeTokenStorageToStream(dob)
     amContainer.setContainerTokens(ByteBuffer.wrap(dob.getData()))
   }
 
+  /**
+   * Return the amount of memory for launching the ApplicationMaster container (MB).
+   * GetNewApplicationResponse#getMinimumResourceCapability does not exist in the stable API.
+   */
+  override def getAMMemory(newAppResponse: GetNewApplicationResponse): Int = {
+    val minResMemory = newAppResponse.getMinimumResourceCapability().getMemory()
+    val amMemory = ((args.amMemory / minResMemory) * minResMemory) +
+      ((if ((args.amMemory % minResMemory) == 0) 0 else minResMemory) - memoryOverhead)
+    amMemory
+  }
+
   /**
    * Return the security token used by this client to communicate with the ApplicationMaster.
    * If no security is enabled, the token returned by the report is null.
+   * ApplicationReport#getClientToken is renamed `getClientToAMToken` in the stable API.
    */
   override def getClientToken(report: ApplicationReport): String =
     Option(report.getClientToken).getOrElse("")

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

@@ -24,7 +24,7 @@ import org.apache.spark.util.{Utils, IntParam, MemoryParam}
 
 
 // TODO: Add code and support for ensuring that yarn resource 'tasks' are location aware !
-private[spark] class ClientArguments(val args: Array[String], val sparkConf: SparkConf) {
+private[spark] class ClientArguments(args: Array[String], sparkConf: SparkConf) {
   var addJars: String = null
   var files: String = null
   var archives: String = null
@@ -47,6 +47,7 @@ private[spark] class ClientArguments(val args: Array[String], val sparkConf: Spa
   loadDefaultArgs()
   validateArgs()
 
+  /** Load any default arguments provided through environment variables and Spark properties. */
   private def loadDefaultArgs(): Unit = {
     // For backward compatibility, SPARK_YARN_DIST_{ARCHIVES/FILES} should be resolved to hdfs://,
     // while spark.yarn.dist.{archives/files} should be resolved to file:// (SPARK-2051).
@@ -62,6 +63,10 @@ private[spark] class ClientArguments(val args: Array[String], val sparkConf: Spa
       .orNull
   }
 
+  /**
+   * Fail fast if any arguments provided are invalid.
+   * This is intended to be called only after the provided arguments have been parsed.
+   */
   private def validateArgs(): Unit = {
     Map[Boolean, String](
       (numExecutors <= 0) -> "You must specify at least 1 executor!",