Review feedback.

Author: Marcelo Vanzin

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

@@ -117,7 +117,6 @@ 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.getOrElse(NOT_STARTED)} ($appId)")
 
         val uiAclsEnabled = conf.getBoolean("spark.history.ui.acls.enable", false)
@@ -183,29 +182,28 @@ private[history] class FsHistoryProvider(conf: SparkConf) extends ApplicationHis
 
       lastModifiedTime = newLastModifiedTime
 
+      // When there are new logs, merge the new list with the existing one, maintaining
+      // the expected ordering (descending end time). Maintaining the order is important
+      // to avoid having to sort the list every time there is a request for the log list.
       if (!logInfos.isEmpty) {
-        var newApps = new mutable.LinkedHashMap[String, FsApplicationHistoryInfo]()
-
-        def addToList(info: FsApplicationHistoryInfo) = {
+        val newApps = new mutable.LinkedHashMap[String, FsApplicationHistoryInfo]()
+        def addIfAbsent(info: FsApplicationHistoryInfo) = {
           if (!newApps.contains(info.id)) {
             newApps += (info.id -> info)
           }
         }
 
-        // Merge the new apps with the existing ones, discarding any duplicates.
         val newIterator = logInfos.iterator.buffered
         val oldIterator = applications.values.iterator.buffered
-
         while (newIterator.hasNext && oldIterator.hasNext) {
           if (newIterator.head.endTime > oldIterator.head.endTime) {
-            addToList(newIterator.next)
+            addIfAbsent(newIterator.next)
           } else {
-            addToList(oldIterator.next)
+            addIfAbsent(oldIterator.next)
           }
         }
-
-        newIterator.foreach(addToList)
-        oldIterator.foreach(addToList)
+        newIterator.foreach(addIfAbsent)
+        oldIterator.foreach(addIfAbsent)
 
         applications = newApps
       }

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

@@ -42,7 +42,7 @@ import org.apache.spark.ui.JettyUtils
  * (spark.deploy.*).
  */
 private[spark]
-abstract class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, actorSystem: ActorSystem)
+class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, actorSystem: ActorSystem)
   extends SchedulerBackend with Logging
 {
   // Use an atomic variable to track total number of cores in the cluster for simplicity and speed
@@ -51,12 +51,12 @@ abstract class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, actor
   val conf = scheduler.sc.conf
   private val timeout = AkkaUtils.askTimeout(conf)
   private val akkaFrameSize = AkkaUtils.maxFrameSizeBytes(conf)
-  // Submit tasks only after (registered resources / total expected resources) 
+  // Submit tasks only after (registered resources / total expected resources)
   // is equal to at least this value, that is double between 0 and 1.
   var minRegisteredRatio =
     math.min(1, conf.getDouble("spark.scheduler.minRegisteredResourcesRatio", 0))
   // Submit tasks after maxRegisteredWaitingTime milliseconds
-  // if minRegisteredRatio has not yet been reached  
+  // if minRegisteredRatio has not yet been reached
   val maxRegisteredWaitingTime =
     conf.getInt("spark.scheduler.maxRegisteredResourcesWaitingTime", 30000)
   val createTime = System.currentTimeMillis()

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

@@ -88,18 +88,18 @@ private[spark] class SparkDeploySchedulerBackend(
   override def connected(appId: String) {
     logInfo("Connected to Spark cluster with app ID " + appId)
     this.appId = appId
-    wakeUpContext()
+    notifyContext()
   }
 
   override def disconnected() {
-    wakeUpContext()
+    notifyContext()
     if (!stopping) {
       logWarning("Disconnected from Spark cluster! Waiting for reconnection...")
     }
   }
 
   override def dead(reason: String) {
-    wakeUpContext()
+    notifyContext()
     if (!stopping) {
       logError("Application has been killed. Reason: " + reason)
       scheduler.error(reason)
@@ -127,7 +127,7 @@ private[spark] class SparkDeploySchedulerBackend(
     totalCoreCount.get() >= totalExpectedCores * minRegisteredRatio
   }
 
-  override def applicationId(): Option[String] = Some(appId)
+  override def applicationId(): Option[String] = Option(appId)
 
   private def waitForRegistration() = {
     registrationLock.synchronized {
@@ -137,7 +137,7 @@ private[spark] class SparkDeploySchedulerBackend(
     }
   }
 
-  private def wakeUpContext() = {
+  private def notifyContext() = {
     registrationLock.synchronized {
       registrationDone = true
       registrationLock.notifyAll()

project/MimaExcludes.scala

@@ -108,11 +108,6 @@ object MimaExcludes {
           MimaBuild.excludeSparkClass("storage.MemoryStore$Entry") ++
           // Class was missing "@DeveloperApi" annotation in 1.0.
           MimaBuild.excludeSparkClass("scheduler.SparkListenerApplicationStart") ++
-          // Class is "private[spark]" but for some reason not being ignored?
-          Seq(
-            ProblemFilters.exclude[AbstractClassProblem](
-              "org.apache.spark.scheduler.cluster.CoarseGrainedSchedulerBackend")
-          ) ++
           Seq(
             ProblemFilters.exclude[IncompatibleMethTypeProblem](
               "org.apache.spark.mllib.tree.impurity.Gini.calculate"),