fix

Author: iRakson

core/src/main/resources/org/apache/spark/ui/static/webui.js

@@ -82,7 +82,8 @@ $(function() {
   collapseTablePageLoad('collapse-aggregated-poolActiveStages','aggregated-poolActiveStages');
   collapseTablePageLoad('collapse-aggregated-tasks','aggregated-tasks');
   collapseTablePageLoad('collapse-aggregated-rdds','aggregated-rdds');
-  collapseTablePageLoad('collapse-aggregated-activeBatches','aggregated-activeBatches');
+  collapseTablePageLoad('collapse-aggregated-waitingBatches','aggregated-waitingBatches');
+  collapseTablePageLoad('collapse-aggregated-runningBatches','aggregated-runningBatches');
   collapseTablePageLoad('collapse-aggregated-completedBatches','aggregated-completedBatches');
   collapseTablePageLoad('collapse-aggregated-runningExecutions','aggregated-runningExecutions');
   collapseTablePageLoad('collapse-aggregated-completedExecutions','aggregated-completedExecutions');

streaming/src/main/scala/org/apache/spark/streaming/ui/AllBatchesTable.scala

@@ -31,17 +31,14 @@ private[ui] class StreamingPagedTable(
     batches: Seq[BatchUIData],
     basePath: String,
     subPath: String,
-    isRunningTable: Boolean,
-    isWaitingTable: Boolean,
-    isCompletedTable: Boolean,
     batchInterval: Long) extends PagedTable[BatchUIData] {
 
   private val(sortColumn, desc, pageSize) = getTableParameters(request, tableTag, "Batch Time")
   private val parameterPath = s"$basePath/$subPath/?${getParameterOtherTable(request, tableTag)}"
   private val encodedSortColumn = URLEncoder.encode(sortColumn, UTF_8.name())
 
   private val firstFailureReason: Option[String] =
-    if (!isWaitingTable) {
+    if (!tableTag.equals("waitingBatches")) {
       getFirstFailureReason(batches)
     } else {
       None
@@ -64,13 +61,17 @@ private[ui] class StreamingPagedTable(
   }
 
   private def createOutputOperationProgressBar(batch: BatchUIData): Seq[Node] = {
-    SparkUIUtils.makeProgressBar(
-      started = batch.numActiveOutputOp,
-      completed = batch.numCompletedOutputOp,
-      failed = batch.numFailedOutputOp,
-      skipped = 0,
-      reasonToNumKilled = Map.empty,
-      total = batch.outputOperations.size)
+    <td class="progress-cell">
+      {
+        SparkUIUtils.makeProgressBar(
+          started = batch.numActiveOutputOp,
+          completed = batch.numCompletedOutputOp,
+          failed = batch.numFailedOutputOp,
+          skipped = 0,
+          reasonToNumKilled = Map.empty,
+          total = batch.outputOperations.size)
+      }
+    </td>
   }
 
   override def tableId: String = s"$tableTag-table"
@@ -98,19 +99,17 @@ private[ui] class StreamingPagedTable(
     new StreamingDataSource(batches, pageSize, sortColumn, desc)
 
   override def headers: Seq[Node] = {
-    // tuple containing tooltips for header fields
-    val tooltips = ("Time taken by Streaming scheduler to submit jobs of a batch",
-      "Time taken to process all jobs of a batch", "Total time taken to handle a batch")
     // headers, sortable and tooltips
     val headersAndCssClasses: Seq[(String, Boolean, Option[String])] = {
       Seq(
         ("Batch Time", true, None),
         ("Records", true, None),
-        ("Scheduling Delay", true, Some(tooltips._1)),
-        ("Processing Time", true, Some(tooltips._2))) ++ {
-        if (isCompletedTable) {
+        ("Scheduling Delay", true, Some("Time taken by Streaming scheduler to submit jobs " +
+          "of a batch")),
+        ("Processing Time", true, Some("Time taken to process all jobs of a batch"))) ++ {
+        if (tableTag.equals("completedBatches")) {
           Seq(
-            ("Total Delay", true, Some(tooltips._3)),
+            ("Total Delay", true, Some("Total time taken to handle a batch")),
             ("Output Ops: Succeeded/Total", false, None))
         } else {
           Seq(
@@ -144,26 +143,26 @@ private[ui] class StreamingPagedTable(
     val formattedTotalDelay = totalDelay.map(SparkUIUtils.formatDuration).getOrElse("-")
 
     <tr>
-      <td id = {batchTimeId} isFailed = {batch.isFailed.toString}>
+      <td id={batchTimeId} isFailed={batch.isFailed.toString}>
         <a href={s"batch?id=$batchTime"}>
           {formattedBatchTime}
         </a>
       </td>
-      <td> {numRecords.toString} Records </td>
+      <td> {numRecords.toString} records </td>
       <td> {formattedSchedulingDelay} </td>
       <td> {formattedProcessingTime} </td>
       {
-        if (isCompletedTable) {
-          <td> {formattedTotalDelay} </td>
-          <td class="progress-cell"> {createOutputOperationProgressBar(batch)} </td> ++ {
+        if (tableTag.equals("completedBatches")) {
+          <td> {formattedTotalDelay} </td> ++
+          createOutputOperationProgressBar(batch) ++ {
             if (firstFailureReason.nonEmpty) {
               getFirstFailureTableCell(batch)
             } else {
               Nil
             }
           }
-        } else if (isRunningTable) {
-          <td class="progress-cell"> {createOutputOperationProgressBar(batch)} </td>
+        } else if (tableTag.equals("runningBatches")) {
+          createOutputOperationProgressBar(batch) ++
           <td> processing </td>  ++ {
             if (firstFailureReason.nonEmpty) {
               getFirstFailureTableCell(batch)
@@ -172,7 +171,7 @@ private[ui] class StreamingPagedTable(
             }
           }
         } else {
-          <td class="progress-cell"> {createOutputOperationProgressBar(batch)} </td>
+          createOutputOperationProgressBar(batch) ++
           <td> queued </td> ++ {
             if (firstFailureReason.nonEmpty) {
               // Waiting batches have not run yet, so must have no failure reasons.

streaming/src/main/scala/org/apache/spark/streaming/ui/StreamingPage.scala

@@ -439,22 +439,13 @@ private[ui] class StreamingPage(parent: StreamingTab)
     val interval: Long = listener.batchDuration
     val streamingPage = Option(request.getParameter(s"$tableTag.page")).map(_.toInt).getOrElse(1)
 
-    val tableType = tableTag match {
-      case "runningBatches" => (true, false, false)
-      case "waitingBatches" => (false, true, false)
-      case "completedBatches" => (false, false, true)
-    }
-
     try {
       new StreamingPagedTable(
         request,
         tableTag,
         batches,
         SparkUIUtils.prependBaseUri(request, parent.basePath),
         "streaming",
-        tableType._1,
-        tableType._2,
-        tableType._3,
         interval
       ).table(streamingPage)
     } catch {

streaming/src/test/scala/org/apache/spark/streaming/UISeleniumSuite.scala

@@ -63,7 +63,7 @@ class UISeleniumSuite
       .setMaster("local")
       .setAppName("test")
       .set(UI_ENABLED, true)
-    val ssc = new StreamingContext(conf, Seconds(1))
+    val ssc = new StreamingContext(conf, Milliseconds(100))
     assert(ssc.sc.ui.isDefined, "Spark UI is not started!")
     ssc
   }
@@ -104,7 +104,7 @@ class UISeleniumSuite
         find(cssSelector( """ul li a[href*="streaming"]""")) should not be (None)
       }
 
-      eventually(timeout(10.seconds), interval(50.milliseconds)) {
+      eventually(timeout(10.seconds), interval(500.milliseconds)) {
         // check whether streaming page exists
         go to (sparkUI.webUrl.stripSuffix("/") + "/streaming")
         val h3Text = findAll(cssSelector("h3")).map(_.text).toSeq
@@ -125,20 +125,47 @@ class UISeleniumSuite
 
         // Check batch tables
         val h4Text = findAll(cssSelector("h4")).map(_.text).toSeq
+        h4Text.exists(_.matches("Running Batches \\(\\d+\\)")) should be (true)
+        h4Text.exists(_.matches("Waiting Batches \\(\\d+\\)")) should be (true)
         h4Text.exists(_.matches("Completed Batches \\(last \\d+ out of \\d+\\)")) should be (true)
 
         val arrow = 0x25BE.toChar
+        findAll(cssSelector("""#runningBatches-table th""")).map(_.text).toList should be {
+          List(s"Batch Time $arrow", "Records", "Scheduling Delay", "Processing Time",
+            "Output Ops: Succeeded/Total", "Status")
+        }
+        findAll(cssSelector("""#waitingBatches-table th""")).map(_.text).toList should be {
+          List(s"Batch Time $arrow", "Records", "Scheduling Delay", "Processing Time",
+            "Output Ops: Succeeded/Total", "Status")
+        }
         findAll(cssSelector("""#completedBatches-table th""")).map(_.text).toList should be {
           List(s"Batch Time $arrow", "Records", "Scheduling Delay", "Processing Time",
             "Total Delay", "Output Ops: Succeeded/Total")
         }
 
-        val batchLinks =
-          findAll(cssSelector("""#completedBatches-table a""")).flatMap(_.attribute("href")).toSeq
+        val pageSize = 3
+        val pagedTablePath = "/streaming/?completedBatches.sort=Batch+Time" +
+          "&completedBatches.desc=true&completedBatches.page=1" +
+          s"&completedBatches.pageSize=$pageSize#completedBatches"
+
+        go to (sparkUI.webUrl.stripSuffix("/") + pagedTablePath)
+        val completedTableRows = findAll(cssSelector("""#completedBatches-table tr"""))
+          .map(_.text).toList
+        // header row + pagesize
+        completedTableRows.length should be (1 + pageSize)
+
+        val sortedBatchTimePath = "/streaming/?&completedBatches.sort=Batch+Time" +
+          "&completedBatches.desc=false&completedBatches.pageSize=3#completedBatches"
+
+        // sort batches in ascending order of batch time
+        go to (sparkUI.webUrl.stripSuffix("/") + sortedBatchTimePath)
+
+        val batchLinks = findAll(cssSelector("""#completedBatches-table td a"""))
+          .flatMap(_.attribute("href")).toSeq
         batchLinks.size should be >= 1
 
         // Check a normal batch page
-        go to (batchLinks.last) // Last should be the first batch, so it will have some jobs
+        go to (batchLinks.head) // Head is the first batch, so it will have some jobs
         val summaryText = findAll(cssSelector("li strong")).map(_.text).toSeq
         summaryText should contain ("Batch Duration:")
         summaryText should contain ("Input data size:")