[SPARK-7504] [YARN] NullPointerException when initializing SparkContext in YARN-cluster mode

Removed checking for null in AM.
Refactored configuration-check to SparkConf.

Author: zzvara

core/src/main/scala/org/apache/spark/SparkConf.scala

@@ -448,6 +448,15 @@ class SparkConf(loadDefaults: Boolean) extends Cloneable with Logging {
         }
       }
     }
+
+    // System property spark.yarn.app.id must be set if user code ran by AM on a YARN cluster
+    // yarn-standalone is deprecated, but still supported
+    if ((get("spark.master") == "yarn-cluster" || get("spark.master") == "yarn-standalone") &&
+      get("spark.yarn.app.id", null) == null) {
+      throw new SparkException("Detected yarn-cluster mode, but isn't running on a cluster. " +
+        "Deployment to YARN is not supported directly by SparkContext. Please use spark-submit.")
+    }
+
   }
 
   /**

core/src/main/scala/org/apache/spark/SparkContext.scala

@@ -371,13 +371,6 @@ class SparkContext(config: SparkConf) extends Logging with ExecutorAllocationCli
       throw new SparkException("An application name must be set in your configuration")
     }
 
-    // Thread name has been set to "Driver" if user code ran by AM on a YARN cluster
-    if (master == "yarn-cluster" &&
-        Thread.currentThread().getName != "Driver") {
-      throw new SparkException("Detected yarn-cluster mode, but isn't running on a cluster. " +
-        "Deployment to YARN is not supported directly by SparkContext. Please use spark-submit.")
-    }
-
     if (_conf.getBoolean("spark.logConf", false)) {
       logInfo("Spark configuration:\n" + _conf.toDebugString)
     }
@@ -1477,6 +1470,7 @@ class SparkContext(config: SparkConf) extends Logging with ExecutorAllocationCli
   def addJar(path: String) {
     if (path == null) {
       logWarning("null specified as parameter to addJar")
+      // yarn-standalone is deprecated, but still supported
     } else {
       var key = ""
       if (path.contains("\\")) {

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

@@ -573,16 +573,10 @@ object ApplicationMaster extends Logging {
   }
 
   private[spark] def sparkContextInitialized(sc: SparkContext): Unit = {
-    if (master == null){
-      throw new SparkException("ApplicationMaster is not initialized!")
-    }
     master.sparkContextInitialized(sc)
   }
 
   private[spark] def sparkContextStopped(sc: SparkContext): Boolean = {
-    if (master == null){
-      throw new SparkException("ApplicationMaster is not initialized!")
-    }
     master.sparkContextStopped(sc)
   }