Review feedback. Mostly, more consistent use of Scala's Option.

Author: Marcelo Vanzin

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

@@ -1255,6 +1255,8 @@ class SparkContext(config: SparkConf) extends Logging {
 
   /** Post the application start event */
   private def postApplicationStart() {
+    // Note: this code assumes that the task scheduler has been initialized and has contacted
+    // the cluster manager to get an application ID (in case the cluster manager provides one).
     listenerBus.post(SparkListenerApplicationStart(appName, taskScheduler.applicationId(),
       startTime, sparkUser))
   }

core/src/main/scala/org/apache/spark/deploy/history/ApplicationHistoryProvider.scala

@@ -42,7 +42,7 @@ private[spark] abstract class ApplicationHistoryProvider {
    * @param appId The application ID.
    * @return The application's UI, or null if application is not found.
    */
-  def getAppUI(appId: String): SparkUI
+  def getAppUI(appId: String): Option[SparkUI]
 
   /**
    * Called when the server is shutting down.

core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala

@@ -31,6 +31,8 @@ import org.apache.spark.util.Utils
 private[history] class FsHistoryProvider(conf: SparkConf) extends ApplicationHistoryProvider
   with Logging {
 
+  private val NOT_STARTED = "<Not Started>"
+
   // Interval between each check for event log updates
   private val UPDATE_INTERVAL_MS = conf.getInt("spark.history.fs.updateInterval",
     conf.getInt("spark.history.updateInterval", 10)) * 1000
@@ -98,7 +100,7 @@ private[history] class FsHistoryProvider(conf: SparkConf) extends ApplicationHis
 
   override def getListing() = appList.values
 
-  override def getAppUI(appId: String): SparkUI = {
+  override def getAppUI(appId: String): Option[SparkUI] = {
     try {
       appList.get(appId).map(info => {
         val (replayBus, appListener) = createReplayBus(fs.getFileStatus(
@@ -114,15 +116,16 @@ private[history] class FsHistoryProvider(conf: SparkConf) extends ApplicationHis
         replayBus.replay()
 
         // Note that this does not have any effect due to SPARK-2169.
-        ui.setAppName(s"${appListener.appName} ($appId)")
+        ui.setAppName(s"${appListener.appName.getOrElse(NOT_STARTED)} ($appId)")
 
         val uiAclsEnabled = conf.getBoolean("spark.history.ui.acls.enable", false)
         ui.getSecurityManager.setUIAcls(uiAclsEnabled)
-        ui.getSecurityManager.setViewAcls(appListener.sparkUser, appListener.viewAcls)
+        ui.getSecurityManager.setViewAcls(appListener.sparkUser.getOrElse(NOT_STARTED),
+          appListener.viewAcls.getOrElse(""))
         ui
-      }).getOrElse(null)
+      })
     } catch {
-      case e: FileNotFoundException => null
+      case e: FileNotFoundException => None
     }
   }
 
@@ -161,18 +164,18 @@ private[history] class FsHistoryProvider(conf: SparkConf) extends ApplicationHis
             new FsApplicationHistoryInfo(
               dir.getPath().getName(),
               appListener.appId.getOrElse(dir.getPath().getName()),
-              appListener.appName,
-              appListener.startTime,
-              appListener.endTime,
+              appListener.appName.getOrElse(NOT_STARTED),
+              appListener.startTime.getOrElse(-1L),
+              appListener.endTime.getOrElse(-1L),
               getModificationTime(dir),
-              appListener.sparkUser)
+              appListener.sparkUser.getOrElse(NOT_STARTED))
           } catch {
             case e: Exception =>
               logInfo(s"Failed to load application log data from $dir.", e)
               null
           }
         }
-        .sortBy { info => -info.endTime }
+        .sortBy { info => if (info != null) -info.endTime else -1 }
 
       mostRecentLogModTime = newMostRecentModTime
 
@@ -228,7 +231,8 @@ private[history] class FsHistoryProvider(conf: SparkConf) extends ApplicationHis
 
     val uiAclsEnabled = conf.getBoolean("spark.history.ui.acls.enable", false)
     ui.getSecurityManager.setUIAcls(uiAclsEnabled)
-    ui.getSecurityManager.setViewAcls(appListener.sparkUser, appListener.viewAcls)
+    ui.getSecurityManager.setViewAcls(appListener.sparkUser.getOrElse(NOT_STARTED),
+      appListener.viewAcls.getOrElse(""))
     ui
   }
 

core/src/main/scala/org/apache/spark/deploy/history/HistoryServer.scala

@@ -52,10 +52,7 @@ class HistoryServer(
 
   private val appLoader = new CacheLoader[String, SparkUI] {
     override def load(key: String): SparkUI = {
-      val ui = provider.getAppUI(key)
-      if (ui == null) {
-        throw new NoSuchElementException()
-      }
+      val ui = provider.getAppUI(key).getOrElse(throw new NoSuchElementException())
       attachSparkUI(ui)
       ui
     }

core/src/main/scala/org/apache/spark/scheduler/ApplicationEventListener.scala

@@ -24,39 +24,38 @@ package org.apache.spark.scheduler
  * from multiple applications are seen, the behavior is unspecified.
  */
 private[spark] class ApplicationEventListener extends SparkListener {
-  var appName = "<Not Started>"
+  var appName: Option[String] = None
   var appId: Option[String] = None
-  var sparkUser = "<Not Started>"
-  var startTime = -1L
-  var endTime = -1L
-  var viewAcls = ""
+  var sparkUser: Option[String] = None
+  var startTime: Option[Long] = None
+  var endTime: Option[Long] = None
+  var viewAcls: Option[String] = None
   var enableViewAcls = false
 
-  def applicationStarted = startTime != -1
+  def applicationStarted = startTime.isDefined
 
-  def applicationCompleted = endTime != -1
+  def applicationCompleted = endTime.isDefined
 
   def applicationDuration: Long = {
-    val difference = endTime - startTime
-    if (applicationStarted && applicationCompleted && difference > 0) difference else -1L
+    if (applicationStarted && applicationCompleted) endTime.get - startTime.get else -1
   }
 
   override def onApplicationStart(applicationStart: SparkListenerApplicationStart) {
-    appName = applicationStart.appName
+    appName = Some(applicationStart.appName)
     appId = applicationStart.appId
-    startTime = applicationStart.time
-    sparkUser = applicationStart.sparkUser
+    startTime = Some(applicationStart.time)
+    sparkUser = Some(applicationStart.sparkUser)
   }
 
   override def onApplicationEnd(applicationEnd: SparkListenerApplicationEnd) {
-    endTime = applicationEnd.time
+    endTime = Some(applicationEnd.time)
   }
 
   override def onEnvironmentUpdate(environmentUpdate: SparkListenerEnvironmentUpdate) {
     synchronized {
       val environmentDetails = environmentUpdate.environmentDetails
       val allProperties = environmentDetails("Spark Properties").toMap
-      viewAcls = allProperties.getOrElse("spark.ui.view.acls", "")
+      viewAcls = allProperties.get("spark.ui.view.acls")
       enableViewAcls = allProperties.getOrElse("spark.ui.acls.enable", "false").toBoolean
     }
   }

core/src/main/scala/org/apache/spark/scheduler/cluster/SparkDeploySchedulerBackend.scala

@@ -83,18 +83,18 @@ private[spark] class SparkDeploySchedulerBackend(
   override def connected(appId: String) {
     logInfo("Connected to Spark cluster with app ID " + appId)
     this.appId = appId
-    notifyRegistered()
+    wakeUpContext()
   }
 
   override def disconnected() {
-    notifyRegistered()
+    wakeUpContext()
     if (!stopping) {
       logWarning("Disconnected from Spark cluster! Waiting for reconnection...")
     }
   }
 
   override def dead(reason: String) {
-    notifyRegistered()
+    wakeUpContext()
     if (!stopping) {
       logError("Application has been killed. Reason: " + reason)
       scheduler.error(reason)
@@ -129,7 +129,7 @@ private[spark] class SparkDeploySchedulerBackend(
     }
   }
 
-  private def notifyRegistered() = {
+  private def wakeUpContext() = {
     registrationLock.synchronized {
       registrationDone = true
       registrationLock.notifyAll()

core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/CoarseMesosSchedulerBackend.scala

@@ -304,6 +304,6 @@ private[spark] class CoarseMesosSchedulerBackend(
     slaveLost(d, s)
   }
 
-  override def applicationId(): Option[String] =
-    Some(frameworkId).map(id => Some(id.getValue())).getOrElse(null)
+  override def applicationId(): Option[String] = None
+
 }

core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosSchedulerBackend.scala

@@ -337,6 +337,6 @@ private[spark] class MesosSchedulerBackend(
   // TODO: query Mesos for number of cores
   override def defaultParallelism() = sc.conf.getInt("spark.default.parallelism", 8)
 
-  override def applicationId(): Option[String] =
-    Some(frameworkId).map(id => Some(id.getValue())).getOrElse(null)
+  override def applicationId(): Option[String] = None
+
 }