Rudimentary paging support for the history UI.

The provider's list api was tweaked a little bit so that the caller
can get an atomic view of the data currently held in the provider.

Author: Marcelo Vanzin

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

@@ -83,17 +83,21 @@ class FsHistoryProvider(conf: SparkConf) extends ApplicationHistoryProvider
     }
 
     checkForLogs()
+    logCheckingThread.setDaemon(true)
     logCheckingThread.start()
   }
 
   override def stop() = {
     stopped = true
     logCheckingThread.interrupt()
     logCheckingThread.join()
+    fs.close()
   }
 
-  override def getListing(offset: Int, limit: Int): Seq[ApplicationHistoryInfo] = {
-    appList.get()
+  override def getListing(offset: Int, count: Int) = {
+    val list = appList.get()
+    val theOffset = if (offset < list.size) offset else 0
+    (list.slice(theOffset, Math.min(theOffset + count, list.size)), theOffset, list.size)
   }
 
   override def getAppInfo(appId: String): ApplicationHistoryInfo = {

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

@@ -25,17 +25,29 @@ import org.apache.spark.ui.{WebUIPage, UIUtils}
 
 private[spark] class HistoryPage(parent: HistoryServer) extends WebUIPage("") {
 
+  val pageSize = 20
+
   def render(request: HttpServletRequest): Seq[Node] = {
-    val apps = parent.getApplicationList(0, -1)
+    val requestedPage = Option(request.getParameter("page")).getOrElse("1").toInt
+    val requestedFirst = (requestedPage - 1) * pageSize
+    val (apps, actualFirst, count) = parent.getApplicationList(requestedFirst, pageSize)
+    val actualPage = (actualFirst / pageSize) + 1
+    val last = Math.min(actualFirst + pageSize, count) - 1
+    val pageCount = count / pageSize + (if (count % pageSize > 0) 1 else 0)
+
     val appTable = UIUtils.listingTable(appHeader, appRow, apps)
     val content =
       <div class="row-fluid">
         <div class="span12">
           {
-            if (apps.size > 0) {
+            if (count > 0) {
               <h4>
-                Showing {apps.size}/{apps.size}
-                Completed Application{if (apps.size > 1) "s" else ""}
+                Showing {actualFirst + 1}-{last + 1} of {count}
+                Application{if (last - actualFirst > 1) "s" else ""}
+                <span style="float: right">
+                  {if (actualPage > 1) <a href={"/?page=" + (actualPage - 1)}>&lt;</a>}
+                  {if (actualPage < pageCount) <a href={"/?page=" + (actualPage + 1)}>&gt;</a>}
+                </span>
               </h4> ++
               appTable
             } else {

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

@@ -155,11 +155,12 @@ class HistoryServer(
    * Returns a list of available applications, in descending order according to their last
    * updated time.
    *
-   * @param offset Offset of the first entry to return.
-   * @param limit Maximum number of entries to return (-1 = no limit).
+   * @param offset Starting offset for returned objects.
+   * @param count Max number of objects to return.
+   * @return 3-tuple (requested app list, adjusted offset, count of all available apps)
    */
-  def getApplicationList(offset: Int, limit: Int) = {
-    provider.getListing(offset, limit)
+  def getApplicationList(offset: Int, count: Int) = {
+    provider.getListing(offset, count)
   }
 
 }
@@ -269,13 +270,16 @@ private[spark] abstract class ApplicationHistoryProvider {
 
   /**
    * This method should return a list of applications available for the history server to
-   * show. The listing is assumed to be in descending time order (so that the parameters
-   * make sense).
+   * show. The listing is assumed to be in descending time order.
    *
-   * @param offset Offset of the first entry to return.
-   * @param limit Maximum number of entries to return (-1 = no limit).
+   * An adjusted offset should be returned if the app list has changed and the request
+   * references an invalid start offset. Otherwise, the provided offset should be returned.
+   *
+   * @param offset Starting offset for returned objects.
+   * @param count Max number of objects to return.
+   * @return 3-tuple (requested app list, adjusted offset, count of all available apps)
    */
-  def getListing(offset: Int, limit: Int): Seq[ApplicationHistoryInfo]
+  def getListing(offset: Int, count: Int): (Seq[ApplicationHistoryInfo], Int, Int)
 
   /**
    * This method should return the application information, including a rendered SparkUI.