[SPARK-10541][WEB UI] Allow ApplicationHistoryProviders to provide their own...

[SPARK-10541][WEB UI] Allow ApplicationHistoryProviders to provide their own text when there aren't any complete apps

## What changes were proposed in this pull request?

I've added a method to `ApplicationHistoryProvider` that returns the html paragraph to display when there are no applications. This allows providers other than `FsHistoryProvider` to determine what is printed. The current hard coded text is now moved into `FsHistoryProvider` since it assumed that's what was being used before.

I chose to make the function return html rather than text because the current text block had inline html in it and it allows a new implementation of `ApplicationHistoryProvider` more versatility. I did not see any security issues with this since injecting html here requires implementing `ApplicationHistoryProvider` and can't be done outside of code.

## How was this patch tested?

Manual testing and dev/run-tests

No visible changes to the UI

Author: Alex Bozarth <ajbozart@us.ibm.com>

Closes #15490 from ajbozarth/spark10541.

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

@@ -19,6 +19,8 @@ package org.apache.spark.deploy.history
 
 import java.util.zip.ZipOutputStream
 
+import scala.xml.Node
+
 import org.apache.spark.SparkException
 import org.apache.spark.ui.SparkUI
 
@@ -114,4 +116,8 @@ private[history] abstract class ApplicationHistoryProvider {
    */
   def getApplicationInfo(appId: String): Option[ApplicationHistoryInfo]
 
+  /**
+   * @return html text to display when the application list is empty
+   */
+  def getEmptyListingHtml(): Seq[Node] = Seq.empty
 }

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

@@ -23,6 +23,7 @@ import java.util.concurrent.{Executors, ExecutorService, TimeUnit}
 import java.util.zip.{ZipEntry, ZipOutputStream}
 
 import scala.collection.mutable
+import scala.xml.Node
 
 import com.google.common.io.ByteStreams
 import com.google.common.util.concurrent.{MoreExecutors, ThreadFactoryBuilder}
@@ -262,6 +263,17 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
     }
   }
 
+  override def getEmptyListingHtml(): Seq[Node] = {
+    <p>
+      Did you specify the correct logging directory? Please verify your setting of
+      <span style="font-style:italic">spark.history.fs.logDirectory</span>
+      listed above and whether you have the permissions to access it.
+      <br/>
+      It is also possible that your application did not run to
+      completion or did not stop the SparkContext.
+    </p>
+  }
+
   override def getConfig(): Map[String, String] = {
     val safeMode = if (isFsInSafeMode()) {
       Map("HDFS State" -> "In safe mode, application logs not available.")

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

@@ -47,13 +47,7 @@ private[history] class HistoryPage(parent: HistoryServer) extends WebUIPage("")
             } else if (requestedIncomplete) {
               <h4>No incomplete applications found!</h4>
             } else {
-              <h4>No completed applications found!</h4> ++
-                <p>Did you specify the correct logging directory?
-                  Please verify your setting of <span style="font-style:italic">
-                  spark.history.fs.logDirectory</span> and whether you have the permissions to
-                  access it.<br /> It is also possible that your application did not run to
-                  completion or did not stop the SparkContext.
-                </p>
+              <h4>No completed applications found!</h4> ++ parent.emptyListingHtml
             }
             }
 

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

@@ -22,6 +22,7 @@ import java.util.zip.ZipOutputStream
 import javax.servlet.http.{HttpServlet, HttpServletRequest, HttpServletResponse}
 
 import scala.util.control.NonFatal
+import scala.xml.Node
 
 import org.eclipse.jetty.servlet.{ServletContextHandler, ServletHolder}
 
@@ -193,6 +194,13 @@ class HistoryServer(
     provider.writeEventLogs(appId, attemptId, zipStream)
   }
 
+  /**
+   * @return html text to display when the application list is empty
+   */
+  def emptyListingHtml(): Seq[Node] = {
+    provider.getEmptyListingHtml()
+  }
+
   /**
    * Returns the provider configuration to show in the listing page.
    *