Fix SparkConf usage.

Author: harishreedharan

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

@@ -101,9 +101,8 @@ class YarnSparkHadoopUtil extends SparkHadoopUtil {
   }
 
   private[spark] override def scheduleLoginFromKeytab(): Unit = {
-    val principal = sparkConf.get("spark.yarn.principal")
-    val keytab = sparkConf.get("spark.yarn.keytab")
-    if (principal != null) {
+    sparkConf.getOption("spark.yarn.principal").foreach { principal =>
+      val keytab = sparkConf.get("spark.yarn.keytab")
       val delegationTokenRenewerRunnable =
         new Runnable {
           override def run(): Unit = {
@@ -155,8 +154,7 @@ class YarnSparkHadoopUtil extends SparkHadoopUtil {
 
   override def updateCredentialsIfRequired(): Unit = {
     try {
-      val credentialsFile = sparkConf.get("spark.yarn.credentials.file")
-      if (credentialsFile != null && !credentialsFile.isEmpty) {
+      sparkConf.getOption("spark.yarn.credentials.file").foreach { credentialsFile =>
         val credentialsFilePath = new Path(credentialsFile)
         val remoteFs = FileSystem.get(conf)
         val stagingDirPath = new Path(remoteFs.getHomeDirectory, credentialsFilePath.getParent)

yarn/src/test/scala/org/apache/spark/deploy/yarn/YarnClusterSuite.scala

@@ -85,6 +85,7 @@ class YarnClusterSuite extends FunSuite with BeforeAndAfterAll with Matchers wit
     tempDir = Utils.createTempDir()
     logConfDir = new File(tempDir, "log4j")
     logConfDir.mkdir()
+    System.setProperty("SPARK_YARN_MODE", "true")
 
     val logConfFile = new File(logConfDir, "log4j.properties")
     Files.write(LOG4J_CONF, logConfFile, UTF_8)
@@ -124,6 +125,7 @@ class YarnClusterSuite extends FunSuite with BeforeAndAfterAll with Matchers wit
 
   override def afterAll() {
     yarnCluster.stop()
+    System.clearProperty("SPARK_YARN_MODE")
     super.afterAll()
   }