Improve sorting of app attempts.

Make sure running attempts show up first. Also, when an app has running attempts, don't show it in the complete app list.
nemccarthy · Apr 23, 2015 · 1aa309d · 1aa309d
1 parent 2ad77e7
commit 1aa309d
Show file tree

Hide file tree

Showing 3 changed files with 25 additions and 7 deletions.
diff --git a/core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala b/core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala
@@ -270,7 +270,9 @@ private[history] class FsHistoryProvider(conf: SparkConf) extends ApplicationHis
     val newIterator = newApps.iterator.buffered
     val oldIterator = applications.values.iterator.buffered
     while (newIterator.hasNext && oldIterator.hasNext) {
-      if (compareAppInfo(newIterator.head, oldIterator.head)) {
+      if (newAppMap.contains(oldIterator.head.id)) {
+        oldIterator.next()
+      } else if (compareAppInfo(newIterator.head, oldIterator.head)) {
         addIfAbsent(newIterator.next())
       } else {
         addIfAbsent(oldIterator.next())
@@ -328,13 +330,29 @@ private[history] class FsHistoryProvider(conf: SparkConf) extends ApplicationHis
   private def compareAppInfo(
       i1: FsApplicationHistoryInfo,
       i2: FsApplicationHistoryInfo): Boolean = {
-    compareAttemptInfo(i1.attempts.head, i2.attempts.head)
+    val a1 = i1.attempts.head
+    val a2 = i2.attempts.head
+    if (a1.endTime != a2.endTime) a1.endTime >= a2.endTime else a1.startTime >= a2.startTime
   }
 
+  /**
+   * Comparison function that defines the sort order for application attempts within the same
+   * application. Order is: running attempts before complete attempts, running attempts sorted
+   * by start time, completed attempts sorted by end time.
+   *
+   * Normally applications should have a single running attempt; but failure to call sc.stop()
+   * may cause multiple running attempts to show up.
+   *
+   * @return Whether `a1` should precede `a2`.
+   */
   private def compareAttemptInfo(
-      i1: FsApplicationAttemptInfo,
-      i2: FsApplicationAttemptInfo): Boolean = {
-    if (i1.endTime != i2.endTime) i1.endTime >= i2.endTime else i1.startTime >= i2.startTime
+      a1: FsApplicationAttemptInfo,
+      a2: FsApplicationAttemptInfo): Boolean = {
+    if (a1.completed == a2.completed) {
+      if (a1.completed) a1.endTime >= a2.endTime else a1.startTime >= a2.startTime
+    } else {
+      !a1.completed
+    }
   }
 
   /**

diff --git a/core/src/main/scala/org/apache/spark/deploy/history/HistoryPage.scala b/core/src/main/scala/org/apache/spark/deploy/history/HistoryPage.scala
@@ -38,7 +38,7 @@ private[history] class HistoryPage(parent: HistoryServer) extends WebUIPage("")
       Option(request.getParameter("showIncomplete")).getOrElse("false").toBoolean
 
     val allApps = parent.getApplicationList()
-      .filter(_.attempts.exists(_.completed != requestedIncomplete))
+      .filter(_.attempts.head.completed != requestedIncomplete)
     val allAppsSize = allApps.size
 
     val actualFirst = if (requestedFirst < allAppsSize) requestedFirst else 0

diff --git a/core/src/test/scala/org/apache/spark/deploy/history/FsHistoryProviderSuite.scala b/core/src/test/scala/org/apache/spark/deploy/history/FsHistoryProviderSuite.scala
@@ -244,7 +244,7 @@ class FsHistoryProviderSuite extends FunSuite with BeforeAndAfter with Matchers
     updateAndCheck(provider) { list =>
       list.size should be (1)
       list.head.attempts.size should be (2)
-      list.head.attempts.head.attemptId should be ("attempt1")
+      list.head.attempts.head.attemptId should be ("attempt2")
     }
 
     val completedAttempt2 = newLogFile("app1", Some("attempt2"), inProgress = false)