From e8e7894ab6e7fa9e1ca4ac8b4d692d966cbef361 Mon Sep 17 00:00:00 2001
From: Tathagata Das <tathagata.das1565@gmail.com>
Date: Wed, 16 Sep 2015 17:15:03 -0700
Subject: [PATCH 01/13] Set job descriptions for all streaming jobs

---
 .../spark/streaming/StreamingContext.scala    |  4 +--
 .../streaming/scheduler/JobScheduler.scala    |  5 +++
 .../streaming/scheduler/ReceiverTracker.scala |  3 ++
 .../apache/spark/streaming/ui/BatchPage.scala | 31 ++++++++++++-------
 4 files changed, 29 insertions(+), 14 deletions(-)

diff --git a/streaming/src/main/scala/org/apache/spark/streaming/StreamingContext.scala b/streaming/src/main/scala/org/apache/spark/streaming/StreamingContext.scala
index b496d1f341a0b..ec0b0720ebf64 100644
--- a/streaming/src/main/scala/org/apache/spark/streaming/StreamingContext.scala
+++ b/streaming/src/main/scala/org/apache/spark/streaming/StreamingContext.scala
@@ -44,7 +44,7 @@ import org.apache.spark.streaming.dstream._
 import org.apache.spark.streaming.receiver.{ActorReceiver, ActorSupervisorStrategy, Receiver}
 import org.apache.spark.streaming.scheduler.{JobScheduler, StreamingListener}
 import org.apache.spark.streaming.ui.{StreamingJobProgressListener, StreamingTab}
-import org.apache.spark.util.{CallSite, ShutdownHookManager, Utils}
+import org.apache.spark.util.{CallSite, ShutdownHookManager}
 
 /**
  * Main entry point for Spark Streaming functionality. It provides methods used to create
@@ -198,7 +198,7 @@ class StreamingContext private[streaming] (
 
   private var state: StreamingContextState = INITIALIZED
 
-  private val startSite = new AtomicReference[CallSite](null)
+  private[streaming] val startSite = new AtomicReference[CallSite](null)
 
   private var shutdownHookRef: AnyRef = _
 
diff --git a/streaming/src/main/scala/org/apache/spark/streaming/scheduler/JobScheduler.scala b/streaming/src/main/scala/org/apache/spark/streaming/scheduler/JobScheduler.scala
index 0cd39594ee923..3619461622a3d 100644
--- a/streaming/src/main/scala/org/apache/spark/streaming/scheduler/JobScheduler.scala
+++ b/streaming/src/main/scala/org/apache/spark/streaming/scheduler/JobScheduler.scala
@@ -25,6 +25,7 @@ import scala.util.{Failure, Success}
 import org.apache.spark.Logging
 import org.apache.spark.rdd.PairRDDFunctions
 import org.apache.spark.streaming._
+import org.apache.spark.streaming.ui.UIUtils
 import org.apache.spark.util.{EventLoop, ThreadUtils}
 
 
@@ -193,6 +194,10 @@ class JobScheduler(val ssc: StreamingContext) extends Logging {
     def run() {
       ssc.sc.setLocalProperty(JobScheduler.BATCH_TIME_PROPERTY_KEY, job.time.milliseconds.toString)
       ssc.sc.setLocalProperty(JobScheduler.OUTPUT_OP_ID_PROPERTY_KEY, job.outputOpId.toString)
+      val formattedTime = UIUtils.formatBatchTime(
+        job.time.milliseconds, ssc.graph.batchDuration.milliseconds, showYYYYMMSS = false)
+      ssc.sc.setJobDescription(
+        s"Streaming job from [output operation ${job.outputOpId}, batch time ${formattedTime}]")
       try {
         // We need to assign `eventLoop` to a temp variable. Otherwise, because
         // `JobScheduler.stop(false)` may set `eventLoop` to null when this method is running, then
diff --git a/streaming/src/main/scala/org/apache/spark/streaming/scheduler/ReceiverTracker.scala b/streaming/src/main/scala/org/apache/spark/streaming/scheduler/ReceiverTracker.scala
index f86fd44b48719..3941e42e0cfee 100644
--- a/streaming/src/main/scala/org/apache/spark/streaming/scheduler/ReceiverTracker.scala
+++ b/streaming/src/main/scala/org/apache/spark/streaming/scheduler/ReceiverTracker.scala
@@ -554,6 +554,9 @@ class ReceiverTracker(ssc: StreamingContext, skipReceiverLaunch: Boolean = false
           ssc.sc.makeRDD(Seq(receiver -> scheduledExecutors))
         }
       receiverRDD.setName(s"Receiver $receiverId")
+      ssc.sparkContext.setJobDescription(s"Streaming job running receiver $receiverId")
+      ssc.sparkContext.setCallSite(ssc.startSite.get)
+
       val future = ssc.sparkContext.submitJob[Receiver[_], Unit, Unit](
         receiverRDD, startReceiverFunc, Seq(0), (_, _) => Unit, ())
       // We will keep restarting the receiver job until ReceiverTracker is stopped
diff --git a/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchPage.scala b/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchPage.scala
index 90d1b0fadecfc..92cac93d36122 100644
--- a/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchPage.scala
+++ b/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchPage.scala
@@ -19,14 +19,14 @@ package org.apache.spark.streaming.ui
 
 import javax.servlet.http.HttpServletRequest
 
-import scala.xml.{NodeSeq, Node, Text, Unparsed}
+import scala.xml.{Node, NodeSeq, Text, Unparsed}
 
 import org.apache.commons.lang3.StringEscapeUtils
 
 import org.apache.spark.streaming.Time
-import org.apache.spark.ui.{UIUtils => SparkUIUtils, WebUIPage}
-import org.apache.spark.streaming.ui.StreamingJobProgressListener.{SparkJobId, OutputOpId}
+import org.apache.spark.streaming.ui.StreamingJobProgressListener.{OutputOpId, SparkJobId}
 import org.apache.spark.ui.jobs.UIData.JobUIData
+import org.apache.spark.ui.{UIUtils => SparkUIUtils, WebUIPage}
 
 private[ui] case class SparkJobIdWithUIData(sparkJobId: SparkJobId, jobUIData: Option[JobUIData])
 
@@ -207,16 +207,23 @@ private[ui] class BatchPage(parent: StreamingTab) extends WebUIPage("batch") {
             sparkListener.stageIdToInfo.get(sparkJob.stageIds.max)
           }
         }
-    val lastStageData = lastStageInfo.flatMap { s =>
-      sparkListener.stageIdToData.get((s.stageId, s.attemptId))
-    }
-
-    val lastStageName = lastStageInfo.map(_.name).getOrElse("(Unknown Stage Name)")
-    val lastStageDescription = lastStageData.flatMap(_.description).getOrElse("")
+    Text(lastStageInfo.map { _.name }.getOrElse("(Unknown Stage)"))
+    lastStageInfo match {
+      case Some(stageInfo) =>
+        val details = if (stageInfo.details.nonEmpty) {
+          <span onclick="this.parentNode.querySelector('.stage-details').classList.toggle('collapsed')"
+                class="expand-details">
+            +details
+          </span> ++
+          <div class="stage-details collapsed">
+            <pre>{stageInfo.details}</pre>
+          </div>
+        }
 
-    <span class="description-input" title={lastStageDescription}>
-      {lastStageDescription}
-    </span> ++ Text(lastStageName)
+        <div> {stageInfo.name} {details} </div>
+      case None =>
+        Text("(Unknown)")
+    }
   }
 
   private def failureReasonCell(failureReason: String): Seq[Node] = {

From 130e82e6461dbd5b88e454af6df6a7df84c14cc4 Mon Sep 17 00:00:00 2001
From: Tathagata Das <tathagata.das1565@gmail.com>
Date: Wed, 16 Sep 2015 19:58:02 -0700
Subject: [PATCH 02/13] Added link to streaming UI in desc of streaming jobs

---
 .../org/apache/spark/ui/jobs/StageTable.scala | 24 +++++++++++++++----
 .../streaming/scheduler/JobScheduler.scala    |  3 ++-
 2 files changed, 22 insertions(+), 5 deletions(-)

diff --git a/core/src/main/scala/org/apache/spark/ui/jobs/StageTable.scala b/core/src/main/scala/org/apache/spark/ui/jobs/StageTable.scala
index 99812db4912a3..5d93ff289c100 100644
--- a/core/src/main/scala/org/apache/spark/ui/jobs/StageTable.scala
+++ b/core/src/main/scala/org/apache/spark/ui/jobs/StageTable.scala
@@ -17,11 +17,11 @@
 
 package org.apache.spark.ui.jobs
 
-import scala.xml.Node
-import scala.xml.Text
-
 import java.util.Date
 
+import scala.util.control.NonFatal
+import scala.xml.{Node, Text}
+
 import org.apache.commons.lang3.StringEscapeUtils
 
 import org.apache.spark.scheduler.StageInfo
@@ -116,7 +116,23 @@ private[ui] class StageTableBase(
       stageData <- listener.stageIdToData.get((s.stageId, s.attemptId))
       desc <- stageData.description
     } yield {
-      <span class="description-input" title={desc}>{desc}</span>
+      // If the description can be parsed as HTML and has only relative links, then render
+      // as HTML, otherwise render as escaped string
+      try {
+        val xml = scala.xml.XML.loadString(s"""<span class="description-input">$desc</span>""")
+        val allLinks = xml \\ "_" flatMap { _.attributes } filter { _.key == "href" }
+        val areAllLinksRelative = allLinks.forall { _.value.toString.startsWith ("/") }
+        if (areAllLinksRelative) {
+          xml
+        } else {
+          println("some not relative; all links: " + allLinks.mkString(";"))
+          <span class="description-input"> {desc} </span>
+        }
+      } catch {
+        case NonFatal(e) =>
+          println(e)
+          <span class="description-input"> {desc} </span>
+      }
     }
     <div>{stageDesc.getOrElse("")} {killLink} {nameLink} {details}</div>
   }
diff --git a/streaming/src/main/scala/org/apache/spark/streaming/scheduler/JobScheduler.scala b/streaming/src/main/scala/org/apache/spark/streaming/scheduler/JobScheduler.scala
index 3619461622a3d..d0512d559ef84 100644
--- a/streaming/src/main/scala/org/apache/spark/streaming/scheduler/JobScheduler.scala
+++ b/streaming/src/main/scala/org/apache/spark/streaming/scheduler/JobScheduler.scala
@@ -196,8 +196,9 @@ class JobScheduler(val ssc: StreamingContext) extends Logging {
       ssc.sc.setLocalProperty(JobScheduler.OUTPUT_OP_ID_PROPERTY_KEY, job.outputOpId.toString)
       val formattedTime = UIUtils.formatBatchTime(
         job.time.milliseconds, ssc.graph.batchDuration.milliseconds, showYYYYMMSS = false)
+      val batchLink = s"/streaming/batch/?id=${job.time.milliseconds}"
       ssc.sc.setJobDescription(
-        s"Streaming job from [output operation ${job.outputOpId}, batch time ${formattedTime}]")
+        s"""Streaming job from <a href="$batchLink">[output operation ${job.outputOpId}, batch time ${formattedTime}]</a>""")
       try {
         // We need to assign `eventLoop` to a temp variable. Otherwise, because
         // `JobScheduler.stop(false)` may set `eventLoop` to null when this method is running, then

From 7206f2e9d4c42edf5a6a231b5cbde7309a62f041 Mon Sep 17 00:00:00 2001
From: Tathagata Das <tathagata.das1565@gmail.com>
Date: Wed, 16 Sep 2015 20:04:00 -0700
Subject: [PATCH 03/13] Addressed comments

---
 .../main/scala/org/apache/spark/streaming/ui/BatchPage.scala | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchPage.scala b/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchPage.scala
index 92cac93d36122..5d8cd57053250 100644
--- a/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchPage.scala
+++ b/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchPage.scala
@@ -19,7 +19,7 @@ package org.apache.spark.streaming.ui
 
 import javax.servlet.http.HttpServletRequest
 
-import scala.xml.{Node, NodeSeq, Text, Unparsed}
+import scala.xml._
 
 import org.apache.commons.lang3.StringEscapeUtils
 
@@ -207,7 +207,6 @@ private[ui] class BatchPage(parent: StreamingTab) extends WebUIPage("batch") {
             sparkListener.stageIdToInfo.get(sparkJob.stageIds.max)
           }
         }
-    Text(lastStageInfo.map { _.name }.getOrElse("(Unknown Stage)"))
     lastStageInfo match {
       case Some(stageInfo) =>
         val details = if (stageInfo.details.nonEmpty) {
@@ -218,6 +217,8 @@ private[ui] class BatchPage(parent: StreamingTab) extends WebUIPage("batch") {
           <div class="stage-details collapsed">
             <pre>{stageInfo.details}</pre>
           </div>
+        } else {
+          NodeSeq.Empty
         }
 
         <div> {stageInfo.name} {details} </div>

From b15012f3a2b4f4205723d6ea37bf22bfe7c300f7 Mon Sep 17 00:00:00 2001
From: Tathagata Das <tathagata.das1565@gmail.com>
Date: Wed, 16 Sep 2015 20:30:05 -0700
Subject: [PATCH 04/13] More fixes

---
 .../apache/spark/ui/jobs/AllJobsPage.scala    | 29 +++++++++++++++++--
 .../org/apache/spark/ui/jobs/StageTable.scala | 12 ++++----
 .../streaming/scheduler/JobScheduler.scala    | 19 +++++++-----
 .../apache/spark/streaming/ui/BatchPage.scala |  7 +++--
 4 files changed, 50 insertions(+), 17 deletions(-)

diff --git a/core/src/main/scala/org/apache/spark/ui/jobs/AllJobsPage.scala b/core/src/main/scala/org/apache/spark/ui/jobs/AllJobsPage.scala
index e72547df7254b..dfba40a37303d 100644
--- a/core/src/main/scala/org/apache/spark/ui/jobs/AllJobsPage.scala
+++ b/core/src/main/scala/org/apache/spark/ui/jobs/AllJobsPage.scala
@@ -18,11 +18,14 @@
 package org.apache.spark.ui.jobs
 
 import scala.collection.mutable.{HashMap, ListBuffer}
-import scala.xml.{Node, NodeSeq, Unparsed, Utility}
+import scala.util.control.NonFatal
+import scala.xml._
 
 import java.util.Date
 import javax.servlet.http.HttpServletRequest
 
+import org.apache.commons.lang3.StringEscapeUtils
+
 import org.apache.spark.ui.{ToolTips, UIUtils, WebUIPage}
 import org.apache.spark.ui.jobs.UIData.{ExecutorUIData, JobUIData}
 import org.apache.spark.JobExecutionStatus
@@ -224,6 +227,28 @@ private[ui] class AllJobsPage(parent: JobsTab) extends WebUIPage("") {
       }
       val formattedDuration = duration.map(d => UIUtils.formatDuration(d)).getOrElse("Unknown")
       val formattedSubmissionTime = job.submissionTime.map(UIUtils.formatDate).getOrElse("Unknown")
+      val jobDescription = {
+        val d = lastStageDescription
+        // If the description can be parsed as HTML and has only relative links, then render
+        // as HTML, otherwise render as escaped string
+        try {
+          // Try to load the description as unescaped HTML
+          val xml = XML.loadString("<span class=\"description-input\" " +
+            s"title=${"\"" + StringEscapeUtils.escapeHtml4(d) + "\""}>$d</span>")
+          val allLinks = xml \\ "_" flatMap { _.attributes } filter { _.key == "href" }
+          val areAllLinksRelative = allLinks.forall { _.value.toString.startsWith ("/") }
+          if (areAllLinksRelative) {
+            xml
+          } else {
+            <span class="description-input" title={d}> {d} </span>
+          }
+        } catch {
+          case NonFatal(e) =>
+            <span class="description-input" title={d}> {d} </span>
+        }
+      }
+
+
       val detailUrl =
         "%s/jobs/job?id=%s".format(UIUtils.prependBaseUri(parent.basePath), job.jobId)
       <tr id={"job-" + job.jobId}>
@@ -231,7 +256,7 @@ private[ui] class AllJobsPage(parent: JobsTab) extends WebUIPage("") {
           {job.jobId} {job.jobGroup.map(id => s"($id)").getOrElse("")}
         </td>
         <td>
-          <span class="description-input" title={lastStageDescription}>{lastStageDescription}</span>
+          {jobDescription}
           <a href={detailUrl} class="name-link">{lastStageName}</a>
         </td>
         <td sorttable_customkey={job.submissionTime.getOrElse(-1).toString}>
diff --git a/core/src/main/scala/org/apache/spark/ui/jobs/StageTable.scala b/core/src/main/scala/org/apache/spark/ui/jobs/StageTable.scala
index 5d93ff289c100..5231b8b552797 100644
--- a/core/src/main/scala/org/apache/spark/ui/jobs/StageTable.scala
+++ b/core/src/main/scala/org/apache/spark/ui/jobs/StageTable.scala
@@ -20,7 +20,7 @@ package org.apache.spark.ui.jobs
 import java.util.Date
 
 import scala.util.control.NonFatal
-import scala.xml.{Node, Text}
+import scala.xml.{Node, Text, XML}
 
 import org.apache.commons.lang3.StringEscapeUtils
 
@@ -119,19 +119,21 @@ private[ui] class StageTableBase(
       // If the description can be parsed as HTML and has only relative links, then render
       // as HTML, otherwise render as escaped string
       try {
-        val xml = scala.xml.XML.loadString(s"""<span class="description-input">$desc</span>""")
+        // Try to load the description as unescaped HTML
+        val xml = XML.loadString("<span class=\"description-input\" " +
+          s"title=${"\"" + StringEscapeUtils.escapeHtml4(desc) + "\""}>$desc</span>")
         val allLinks = xml \\ "_" flatMap { _.attributes } filter { _.key == "href" }
         val areAllLinksRelative = allLinks.forall { _.value.toString.startsWith ("/") }
         if (areAllLinksRelative) {
           xml
         } else {
-          println("some not relative; all links: " + allLinks.mkString(";"))
-          <span class="description-input"> {desc} </span>
+          println("links not relative")
+          <span class="description-input" title={desc}> {desc} </span>
         }
       } catch {
         case NonFatal(e) =>
           println(e)
-          <span class="description-input"> {desc} </span>
+          <span class="description-input" title={desc}> {desc} </span>
       }
     }
     <div>{stageDesc.getOrElse("")} {killLink} {nameLink} {details}</div>
diff --git a/streaming/src/main/scala/org/apache/spark/streaming/scheduler/JobScheduler.scala b/streaming/src/main/scala/org/apache/spark/streaming/scheduler/JobScheduler.scala
index d0512d559ef84..32d995dc42f27 100644
--- a/streaming/src/main/scala/org/apache/spark/streaming/scheduler/JobScheduler.scala
+++ b/streaming/src/main/scala/org/apache/spark/streaming/scheduler/JobScheduler.scala
@@ -191,15 +191,20 @@ class JobScheduler(val ssc: StreamingContext) extends Logging {
   }
 
   private class JobHandler(job: Job) extends Runnable with Logging {
+    import JobScheduler._
+
     def run() {
-      ssc.sc.setLocalProperty(JobScheduler.BATCH_TIME_PROPERTY_KEY, job.time.milliseconds.toString)
-      ssc.sc.setLocalProperty(JobScheduler.OUTPUT_OP_ID_PROPERTY_KEY, job.outputOpId.toString)
-      val formattedTime = UIUtils.formatBatchTime(
-        job.time.milliseconds, ssc.graph.batchDuration.milliseconds, showYYYYMMSS = false)
-      val batchLink = s"/streaming/batch/?id=${job.time.milliseconds}"
-      ssc.sc.setJobDescription(
-        s"""Streaming job from <a href="$batchLink">[output operation ${job.outputOpId}, batch time ${formattedTime}]</a>""")
       try {
+        val formattedTime = UIUtils.formatBatchTime(
+          job.time.milliseconds, ssc.graph.batchDuration.milliseconds, showYYYYMMSS = false)
+        val batchUrl = s"/streaming/batch/?id=${job.time.milliseconds}"
+        val batchLinkText = s"[output operation ${job.outputOpId}, batch time ${formattedTime}]"
+
+        ssc.sc.setJobDescription(
+          s"""Streaming job from <a href="$batchUrl">$batchLinkText</a>""")
+        ssc.sc.setLocalProperty(BATCH_TIME_PROPERTY_KEY, job.time.milliseconds.toString)
+        ssc.sc.setLocalProperty(OUTPUT_OP_ID_PROPERTY_KEY, job.outputOpId.toString)
+
         // We need to assign `eventLoop` to a temp variable. Otherwise, because
         // `JobScheduler.stop(false)` may set `eventLoop` to null when this method is running, then
         // it's possible that when `post` is called, `eventLoop` happens to null.
diff --git a/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchPage.scala b/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchPage.scala
index 5d8cd57053250..9129c1f26abd4 100644
--- a/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchPage.scala
+++ b/streaming/src/main/scala/org/apache/spark/streaming/ui/BatchPage.scala
@@ -210,9 +210,10 @@ private[ui] class BatchPage(parent: StreamingTab) extends WebUIPage("batch") {
     lastStageInfo match {
       case Some(stageInfo) =>
         val details = if (stageInfo.details.nonEmpty) {
-          <span onclick="this.parentNode.querySelector('.stage-details').classList.toggle('collapsed')"
-                class="expand-details">
-            +details
+          <span
+            onclick="this.parentNode.querySelector('.stage-details').classList.toggle('collapsed')"
+            class="expand-details">
+              +details
           </span> ++
           <div class="stage-details collapsed">
             <pre>{stageInfo.details}</pre>

From fd0d435ecce8ca06e78615339a1e9ab713d422d3 Mon Sep 17 00:00:00 2001
From: Tathagata Das <tathagata.das1565@gmail.com>
Date: Thu, 17 Sep 2015 01:35:55 -0700
Subject: [PATCH 05/13] Remove println

---
 core/src/main/scala/org/apache/spark/ui/jobs/StageTable.scala | 2 --
 1 file changed, 2 deletions(-)

diff --git a/core/src/main/scala/org/apache/spark/ui/jobs/StageTable.scala b/core/src/main/scala/org/apache/spark/ui/jobs/StageTable.scala
index 5231b8b552797..92e6ca3967eb7 100644
--- a/core/src/main/scala/org/apache/spark/ui/jobs/StageTable.scala
+++ b/core/src/main/scala/org/apache/spark/ui/jobs/StageTable.scala
@@ -127,12 +127,10 @@ private[ui] class StageTableBase(
         if (areAllLinksRelative) {
           xml
         } else {
-          println("links not relative")
           <span class="description-input" title={desc}> {desc} </span>
         }
       } catch {
         case NonFatal(e) =>
-          println(e)
           <span class="description-input" title={desc}> {desc} </span>
       }
     }

From b5cb02e3b2193b16b4b10117bdeb92933248dfb2 Mon Sep 17 00:00:00 2001
From: Tathagata Das <tathagata.das1565@gmail.com>
Date: Thu, 17 Sep 2015 02:05:09 -0700
Subject: [PATCH 06/13] Use scala.xml's own escaping

---
 .../org/apache/spark/ui/jobs/AllJobsPage.scala     | 14 ++++++--------
 .../org/apache/spark/ui/jobs/StageTable.scala      |  4 ++--
 2 files changed, 8 insertions(+), 10 deletions(-)

diff --git a/core/src/main/scala/org/apache/spark/ui/jobs/AllJobsPage.scala b/core/src/main/scala/org/apache/spark/ui/jobs/AllJobsPage.scala
index dfba40a37303d..ebb4b6df068b8 100644
--- a/core/src/main/scala/org/apache/spark/ui/jobs/AllJobsPage.scala
+++ b/core/src/main/scala/org/apache/spark/ui/jobs/AllJobsPage.scala
@@ -17,18 +17,16 @@
 
 package org.apache.spark.ui.jobs
 
-import scala.collection.mutable.{HashMap, ListBuffer}
-import scala.util.control.NonFatal
-import scala.xml._
-
 import java.util.Date
 import javax.servlet.http.HttpServletRequest
 
-import org.apache.commons.lang3.StringEscapeUtils
+import scala.collection.mutable.{HashMap, ListBuffer}
+import scala.util.control.NonFatal
+import scala.xml._
 
-import org.apache.spark.ui.{ToolTips, UIUtils, WebUIPage}
-import org.apache.spark.ui.jobs.UIData.{ExecutorUIData, JobUIData}
 import org.apache.spark.JobExecutionStatus
+import org.apache.spark.ui.jobs.UIData.{ExecutorUIData, JobUIData}
+import org.apache.spark.ui.{ToolTips, UIUtils, WebUIPage}
 
 /** Page showing list of all ongoing and recently finished jobs */
 private[ui] class AllJobsPage(parent: JobsTab) extends WebUIPage("") {
@@ -234,7 +232,7 @@ private[ui] class AllJobsPage(parent: JobsTab) extends WebUIPage("") {
         try {
           // Try to load the description as unescaped HTML
           val xml = XML.loadString("<span class=\"description-input\" " +
-            s"title=${"\"" + StringEscapeUtils.escapeHtml4(d) + "\""}>$d</span>")
+            s"title=${"\"" + Utility.escape(d) + "\""}>$d</span>")
           val allLinks = xml \\ "_" flatMap { _.attributes } filter { _.key == "href" }
           val areAllLinksRelative = allLinks.forall { _.value.toString.startsWith ("/") }
           if (areAllLinksRelative) {
diff --git a/core/src/main/scala/org/apache/spark/ui/jobs/StageTable.scala b/core/src/main/scala/org/apache/spark/ui/jobs/StageTable.scala
index 92e6ca3967eb7..e594051d9cc9e 100644
--- a/core/src/main/scala/org/apache/spark/ui/jobs/StageTable.scala
+++ b/core/src/main/scala/org/apache/spark/ui/jobs/StageTable.scala
@@ -20,7 +20,7 @@ package org.apache.spark.ui.jobs
 import java.util.Date
 
 import scala.util.control.NonFatal
-import scala.xml.{Node, Text, XML}
+import scala.xml.{Utility, Node, Text, XML}
 
 import org.apache.commons.lang3.StringEscapeUtils
 
@@ -121,7 +121,7 @@ private[ui] class StageTableBase(
       try {
         // Try to load the description as unescaped HTML
         val xml = XML.loadString("<span class=\"description-input\" " +
-          s"title=${"\"" + StringEscapeUtils.escapeHtml4(desc) + "\""}>$desc</span>")
+          s"title=${"\"" + Utility.escape(desc) + "\""}>$desc</span>")
         val allLinks = xml \\ "_" flatMap { _.attributes } filter { _.key == "href" }
         val areAllLinksRelative = allLinks.forall { _.value.toString.startsWith ("/") }
         if (areAllLinksRelative) {

From d2ab91d2e79c61d7fe137f7cd89bc291bfb8944c Mon Sep 17 00:00:00 2001
From: Tathagata Das <tathagata.das1565@gmail.com>
Date: Fri, 18 Sep 2015 03:53:06 -0700
Subject: [PATCH 07/13] Addressed comments, and added base URL prepending to
 links in description

---
 .../scala/org/apache/spark/ui/UIUtils.scala   | 43 ++++++++++++++++++-
 .../apache/spark/ui/jobs/AllJobsPage.scala    | 23 +---------
 .../org/apache/spark/ui/jobs/StageTable.scala | 21 +--------
 .../spark/streaming/StreamingContext.scala    |  6 ++-
 .../streaming/scheduler/ReceiverTracker.scala |  2 +-
 5 files changed, 50 insertions(+), 45 deletions(-)

diff --git a/core/src/main/scala/org/apache/spark/ui/UIUtils.scala b/core/src/main/scala/org/apache/spark/ui/UIUtils.scala
index f2da417724104..b1193fe03f30b 100644
--- a/core/src/main/scala/org/apache/spark/ui/UIUtils.scala
+++ b/core/src/main/scala/org/apache/spark/ui/UIUtils.scala
@@ -20,7 +20,9 @@ package org.apache.spark.ui
 import java.text.SimpleDateFormat
 import java.util.{Locale, Date}
 
-import scala.xml.{Node, Text, Unparsed}
+import scala.util.control.NonFatal
+import scala.xml._
+import scala.xml.transform.{RewriteRule, RuleTransformer}
 
 import org.apache.spark.Logging
 import org.apache.spark.ui.scope.RDDOperationGraph
@@ -395,4 +397,43 @@ private[spark] object UIUtils extends Logging {
     </script>
   }
 
+  /**
+   * Convert a description string to HTML. It will try to parse the string as HTML and sanitize
+   * any links. If that fails, then whole string will treated as a simple text.
+   */
+  def makeDescription(desc: String, basePathUri: String): NodeSeq = {
+    import scala.language.postfixOps
+
+    // If the description can be parsed as HTML and has only relative links, then render
+    // as HTML, otherwise render as escaped string
+    try {
+      // Try to load the description as unescaped HTML
+      val xml = XML.loadString("<span class=\"description-input\" " +
+        s"title=${"\"" + Utility.escape(desc) + "\""}>$desc</span>")
+      val allLinks = xml \\ "_" flatMap { _.attributes } filter { _.key == "href" }
+      val areAllLinksRelative = allLinks.forall { _.value.toString.startsWith ("/") }
+
+      // If all the links are relative then, transform the links to absolute links
+      // with basePathUri
+      if (areAllLinksRelative) {
+        val rule = new RewriteRule() {
+          override def transform(n: Node): Seq[Node] = {
+            n match {
+              case e: Elem if e \ "@href" nonEmpty =>
+                val relativePath = e.attribute("href").get.toString
+                val fullUri = s"${basePathUri.stripSuffix("/")}/${relativePath.stripPrefix("/")}"
+                e % Attribute(null, "href", fullUri, Null)
+              case _ => n
+            }
+          }
+        }
+        new RuleTransformer(rule).transform(xml)
+      } else {
+        <span class="description-input" title={desc}> {desc} </span>
+      }
+    } catch {
+      case NonFatal(e) =>
+        <span class="description-input" title={desc}> {desc} </span>
+    }
+  }
 }
diff --git a/core/src/main/scala/org/apache/spark/ui/jobs/AllJobsPage.scala b/core/src/main/scala/org/apache/spark/ui/jobs/AllJobsPage.scala
index ebb4b6df068b8..041cd55ea483b 100644
--- a/core/src/main/scala/org/apache/spark/ui/jobs/AllJobsPage.scala
+++ b/core/src/main/scala/org/apache/spark/ui/jobs/AllJobsPage.scala
@@ -21,7 +21,6 @@ import java.util.Date
 import javax.servlet.http.HttpServletRequest
 
 import scala.collection.mutable.{HashMap, ListBuffer}
-import scala.util.control.NonFatal
 import scala.xml._
 
 import org.apache.spark.JobExecutionStatus
@@ -225,27 +224,7 @@ private[ui] class AllJobsPage(parent: JobsTab) extends WebUIPage("") {
       }
       val formattedDuration = duration.map(d => UIUtils.formatDuration(d)).getOrElse("Unknown")
       val formattedSubmissionTime = job.submissionTime.map(UIUtils.formatDate).getOrElse("Unknown")
-      val jobDescription = {
-        val d = lastStageDescription
-        // If the description can be parsed as HTML and has only relative links, then render
-        // as HTML, otherwise render as escaped string
-        try {
-          // Try to load the description as unescaped HTML
-          val xml = XML.loadString("<span class=\"description-input\" " +
-            s"title=${"\"" + Utility.escape(d) + "\""}>$d</span>")
-          val allLinks = xml \\ "_" flatMap { _.attributes } filter { _.key == "href" }
-          val areAllLinksRelative = allLinks.forall { _.value.toString.startsWith ("/") }
-          if (areAllLinksRelative) {
-            xml
-          } else {
-            <span class="description-input" title={d}> {d} </span>
-          }
-        } catch {
-          case NonFatal(e) =>
-            <span class="description-input" title={d}> {d} </span>
-        }
-      }
-
+      val jobDescription = UIUtils.makeDescription(lastStageDescription, parent.basePath)
 
       val detailUrl =
         "%s/jobs/job?id=%s".format(UIUtils.prependBaseUri(parent.basePath), job.jobId)
diff --git a/core/src/main/scala/org/apache/spark/ui/jobs/StageTable.scala b/core/src/main/scala/org/apache/spark/ui/jobs/StageTable.scala
index e594051d9cc9e..ea806d09b6009 100644
--- a/core/src/main/scala/org/apache/spark/ui/jobs/StageTable.scala
+++ b/core/src/main/scala/org/apache/spark/ui/jobs/StageTable.scala
@@ -19,8 +19,7 @@ package org.apache.spark.ui.jobs
 
 import java.util.Date
 
-import scala.util.control.NonFatal
-import scala.xml.{Utility, Node, Text, XML}
+import scala.xml.{Node, Text}
 
 import org.apache.commons.lang3.StringEscapeUtils
 
@@ -116,23 +115,7 @@ private[ui] class StageTableBase(
       stageData <- listener.stageIdToData.get((s.stageId, s.attemptId))
       desc <- stageData.description
     } yield {
-      // If the description can be parsed as HTML and has only relative links, then render
-      // as HTML, otherwise render as escaped string
-      try {
-        // Try to load the description as unescaped HTML
-        val xml = XML.loadString("<span class=\"description-input\" " +
-          s"title=${"\"" + Utility.escape(desc) + "\""}>$desc</span>")
-        val allLinks = xml \\ "_" flatMap { _.attributes } filter { _.key == "href" }
-        val areAllLinksRelative = allLinks.forall { _.value.toString.startsWith ("/") }
-        if (areAllLinksRelative) {
-          xml
-        } else {
-          <span class="description-input" title={desc}> {desc} </span>
-        }
-      } catch {
-        case NonFatal(e) =>
-          <span class="description-input" title={desc}> {desc} </span>
-      }
+      UIUtils.makeDescription(desc, basePathUri)
     }
     <div>{stageDesc.getOrElse("")} {killLink} {nameLink} {details}</div>
   }
diff --git a/streaming/src/main/scala/org/apache/spark/streaming/StreamingContext.scala b/streaming/src/main/scala/org/apache/spark/streaming/StreamingContext.scala
index ec0b0720ebf64..ed2abe7a3bf55 100644
--- a/streaming/src/main/scala/org/apache/spark/streaming/StreamingContext.scala
+++ b/streaming/src/main/scala/org/apache/spark/streaming/StreamingContext.scala
@@ -198,7 +198,9 @@ class StreamingContext private[streaming] (
 
   private var state: StreamingContextState = INITIALIZED
 
-  private[streaming] val startSite = new AtomicReference[CallSite](null)
+  private val startSite = new AtomicReference[CallSite](null)
+
+  private[streaming] def getStartSite(): CallSite = startSite.get()
 
   private var shutdownHookRef: AnyRef = _
 
@@ -735,7 +737,7 @@ object StreamingContext extends Logging {
         throw new IllegalStateException(
           "Only one StreamingContext may be started in this JVM. " +
             "Currently running StreamingContext was started at" +
-            activeContext.get.startSite.get.longForm)
+            activeContext.get.getStartSite().longForm)
       }
     }
   }
diff --git a/streaming/src/main/scala/org/apache/spark/streaming/scheduler/ReceiverTracker.scala b/streaming/src/main/scala/org/apache/spark/streaming/scheduler/ReceiverTracker.scala
index 3941e42e0cfee..a76a8a1219328 100644
--- a/streaming/src/main/scala/org/apache/spark/streaming/scheduler/ReceiverTracker.scala
+++ b/streaming/src/main/scala/org/apache/spark/streaming/scheduler/ReceiverTracker.scala
@@ -555,7 +555,7 @@ class ReceiverTracker(ssc: StreamingContext, skipReceiverLaunch: Boolean = false
         }
       receiverRDD.setName(s"Receiver $receiverId")
       ssc.sparkContext.setJobDescription(s"Streaming job running receiver $receiverId")
-      ssc.sparkContext.setCallSite(ssc.startSite.get)
+      ssc.sparkContext.setCallSite(ssc.getStartSite())
 
       val future = ssc.sparkContext.submitJob[Receiver[_], Unit, Unit](
         receiverRDD, startReceiverFunc, Seq(0), (_, _) => Unit, ())

From 21c90eafa20cb8433675401372b4c5c223cf0957 Mon Sep 17 00:00:00 2001
From: Tathagata Das <tathagata.das1565@gmail.com>
Date: Mon, 21 Sep 2015 18:08:17 -0700
Subject: [PATCH 08/13] Updated test

---
 .../org/apache/spark/streaming/StreamingContextSuite.scala      | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/streaming/src/test/scala/org/apache/spark/streaming/StreamingContextSuite.scala b/streaming/src/test/scala/org/apache/spark/streaming/StreamingContextSuite.scala
index 3b9d0d15ea04c..c7a877142b374 100644
--- a/streaming/src/test/scala/org/apache/spark/streaming/StreamingContextSuite.scala
+++ b/streaming/src/test/scala/org/apache/spark/streaming/StreamingContextSuite.scala
@@ -204,7 +204,7 @@ class StreamingContextSuite extends SparkFunSuite with BeforeAndAfter with Timeo
 
     // Verify streaming jobs have expected thread-local properties
     assert(jobGroupFound === null)
-    assert(jobDescFound === null)
+    assert(jobDescFound.contains("Streaming job from"))
     assert(jobInterruptFound === "false")
 
     // Verify current thread's thread-local properties have not changed

From 4a45c825705d6af72c0bec3673be231cab852410 Mon Sep 17 00:00:00 2001
From: Tathagata Das <tathagata.das1565@gmail.com>
Date: Mon, 21 Sep 2015 22:17:08 -0700
Subject: [PATCH 09/13] Added check for anchors, and unit tests

---
 .../scala/org/apache/spark/ui/UIUtils.scala   | 56 +++++++++++--------
 .../org/apache/spark/ui/UIUtilsSuite.scala    | 48 ++++++++++++++++
 2 files changed, 82 insertions(+), 22 deletions(-)
 create mode 100644 core/src/test/scala/org/apache/spark/ui/UIUtilsSuite.scala

diff --git a/core/src/main/scala/org/apache/spark/ui/UIUtils.scala b/core/src/main/scala/org/apache/spark/ui/UIUtils.scala
index b1193fe03f30b..562482330f724 100644
--- a/core/src/main/scala/org/apache/spark/ui/UIUtils.scala
+++ b/core/src/main/scala/org/apache/spark/ui/UIUtils.scala
@@ -18,7 +18,7 @@
 package org.apache.spark.ui
 
 import java.text.SimpleDateFormat
-import java.util.{Locale, Date}
+import java.util.{Date, Locale}
 
 import scala.util.control.NonFatal
 import scala.xml._
@@ -408,32 +408,44 @@ private[spark] object UIUtils extends Logging {
     // as HTML, otherwise render as escaped string
     try {
       // Try to load the description as unescaped HTML
-      val xml = XML.loadString("<span class=\"description-input\" " +
-        s"title=${"\"" + Utility.escape(desc) + "\""}>$desc</span>")
-      val allLinks = xml \\ "_" flatMap { _.attributes } filter { _.key == "href" }
-      val areAllLinksRelative = allLinks.forall { _.value.toString.startsWith ("/") }
-
-      // If all the links are relative then, transform the links to absolute links
-      // with basePathUri
-      if (areAllLinksRelative) {
-        val rule = new RewriteRule() {
-          override def transform(n: Node): Seq[Node] = {
-            n match {
-              case e: Elem if e \ "@href" nonEmpty =>
-                val relativePath = e.attribute("href").get.toString
-                val fullUri = s"${basePathUri.stripSuffix("/")}/${relativePath.stripPrefix("/")}"
-                e % Attribute(null, "href", fullUri, Null)
-              case _ => n
-            }
+      val xml = XML.loadString(s"""<span class="description-input">$desc</span>""")
+
+      // Verify that this has only anchors and span (we are wrapping in span)
+      val allowedNodeLabels = Set("a", "span")
+      val illegalNodes = xml \\ "_"  filterNot { case node: Node =>
+        allowedNodeLabels.contains(node.label)
+      }
+      if (illegalNodes.nonEmpty) {
+        throw new IllegalArgumentException(
+          "Only HTML anchors allowed in job descriptions\n" +
+            illegalNodes.map { n => s"${n.label} in $n"}.mkString("\n\t"))
+      }
+
+      // Verify that all links are relative links starting with "/"
+      val allLinks =
+        xml \\ "a" flatMap { _.attributes } filter { _.key == "href" } map { _.value.toString }
+      if (allLinks.exists { ! _.startsWith ("/") }) {
+        throw new IllegalArgumentException(
+          "Links in job descriptions must be relative:\n" + allLinks.mkString("\n\t"))
+      }
+
+      // Prepend the relative links with basePathUri
+      val rule = new RewriteRule() {
+        override def transform(n: Node): Seq[Node] = {
+          n match {
+            case e: Elem if e \ "@href" nonEmpty =>
+              val relativePath = e.attribute("href").get.toString
+              val fullUri = s"${basePathUri.stripSuffix("/")}/${relativePath.stripPrefix("/")}"
+              e % Attribute(null, "href", fullUri, Null)
+            case _ => n
           }
         }
-        new RuleTransformer(rule).transform(xml)
-      } else {
-        <span class="description-input" title={desc}> {desc} </span>
       }
+      new RuleTransformer(rule).transform(xml)
     } catch {
       case NonFatal(e) =>
-        <span class="description-input" title={desc}> {desc} </span>
+        logWarning(s"Invalid job description: $desc ", e)
+        <span class="description-input">{desc}</span>
     }
   }
 }
diff --git a/core/src/test/scala/org/apache/spark/ui/UIUtilsSuite.scala b/core/src/test/scala/org/apache/spark/ui/UIUtilsSuite.scala
new file mode 100644
index 0000000000000..fde64375ada7f
--- /dev/null
+++ b/core/src/test/scala/org/apache/spark/ui/UIUtilsSuite.scala
@@ -0,0 +1,48 @@
+package org.apache.spark.ui
+
+import scala.xml.Elem
+
+import org.apache.spark.SparkFunSuite
+
+class UIUtilsSuite extends SparkFunSuite {
+  import UIUtils._
+
+  test("makeDescription") {
+    verify(
+      """test <a href="/link"> text </a>""",
+      <span class="description-input">test <a href="/link"> text </a></span>,
+      "Correctly formatted text with only anchors and relative links should generate HTML"
+    )
+
+    verify(
+      """test <a href="/link" text </a>""",
+      <span class="description-input">{"""test <a href="/link" text </a>"""}</span>,
+      "Badly formatted text should make the description be treated as a streaming instead of HTML"
+    )
+
+    verify(
+      """test <a href="link"> text </a>""",
+      <span class="description-input">{"""test <a href="link"> text </a>"""}</span>,
+      "Non-relative links should make the description be treated as a string instead of HTML"
+    )
+
+    verify(
+      """test<a><img></img></a>""",
+      <span class="description-input">{"""test<a><img></img></a>"""}</span>,
+      "Non-anchor elements should make the description be treated as a string instead of HTML"
+    )
+    
+    verify(
+      """test <a href="/link"> text </a>""",
+      <span class="description-input">test <a href="base/link"> text </a></span>,
+      baseUrl = "base",
+      errorMsg = "Base URL should be prepended to html links"
+    )
+  }
+
+  def verify(desc: String, expected: Elem, errorMsg: String = "", baseUrl: String = ""): Unit = {
+    val generated = makeDescription(desc, baseUrl)
+    assert(generated.sameElements(expected), 
+      s"\n$errorMsg\n\nExpected:\n$expected\nGenerated:\n$generated")
+  }
+}

From 4a0aba8d995659ef9a0f92081e18278f29dd4259 Mon Sep 17 00:00:00 2001
From: Tathagata Das <tathagata.das1565@gmail.com>
Date: Tue, 22 Sep 2015 02:01:14 -0700
Subject: [PATCH 10/13] Add license

---
 .../org/apache/spark/ui/UIUtilsSuite.scala      | 17 +++++++++++++++++
 1 file changed, 17 insertions(+)

diff --git a/core/src/test/scala/org/apache/spark/ui/UIUtilsSuite.scala b/core/src/test/scala/org/apache/spark/ui/UIUtilsSuite.scala
index fde64375ada7f..a728f2d491a6d 100644
--- a/core/src/test/scala/org/apache/spark/ui/UIUtilsSuite.scala
+++ b/core/src/test/scala/org/apache/spark/ui/UIUtilsSuite.scala
@@ -1,3 +1,20 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
 package org.apache.spark.ui
 
 import scala.xml.Elem

From f9524c61fd56ba07099e96db86f95cff88520aa4 Mon Sep 17 00:00:00 2001
From: Tathagata Das <tathagata.das1565@gmail.com>
Date: Tue, 22 Sep 2015 02:14:54 -0700
Subject: [PATCH 11/13] style fix

---
 core/src/test/scala/org/apache/spark/ui/UIUtilsSuite.scala | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/core/src/test/scala/org/apache/spark/ui/UIUtilsSuite.scala b/core/src/test/scala/org/apache/spark/ui/UIUtilsSuite.scala
index a728f2d491a6d..39d9629b930cf 100644
--- a/core/src/test/scala/org/apache/spark/ui/UIUtilsSuite.scala
+++ b/core/src/test/scala/org/apache/spark/ui/UIUtilsSuite.scala
@@ -48,7 +48,7 @@ class UIUtilsSuite extends SparkFunSuite {
       <span class="description-input">{"""test<a><img></img></a>"""}</span>,
       "Non-anchor elements should make the description be treated as a string instead of HTML"
     )
-    
+
     verify(
       """test <a href="/link"> text </a>""",
       <span class="description-input">test <a href="base/link"> text </a></span>,
@@ -59,7 +59,7 @@ class UIUtilsSuite extends SparkFunSuite {
 
   def verify(desc: String, expected: Elem, errorMsg: String = "", baseUrl: String = ""): Unit = {
     val generated = makeDescription(desc, baseUrl)
-    assert(generated.sameElements(expected), 
+    assert(generated.sameElements(expected),
       s"\n$errorMsg\n\nExpected:\n$expected\nGenerated:\n$generated")
   }
 }

From 2fbfd1f215f80072b15b65d6e43beb82156acf5d Mon Sep 17 00:00:00 2001
From: Tathagata Das <tathagata.das1565@gmail.com>
Date: Tue, 22 Sep 2015 12:12:09 -0700
Subject: [PATCH 12/13] fixed test

---
 core/src/main/scala/org/apache/spark/ui/UIUtils.scala      | 7 ++++---
 .../apache/spark/streaming/scheduler/ReceiverTracker.scala | 4 ++--
 2 files changed, 6 insertions(+), 5 deletions(-)

diff --git a/core/src/main/scala/org/apache/spark/ui/UIUtils.scala b/core/src/main/scala/org/apache/spark/ui/UIUtils.scala
index 562482330f724..f37e262a52adf 100644
--- a/core/src/main/scala/org/apache/spark/ui/UIUtils.scala
+++ b/core/src/main/scala/org/apache/spark/ui/UIUtils.scala
@@ -398,8 +398,9 @@ private[spark] object UIUtils extends Logging {
   }
 
   /**
-   * Convert a description string to HTML. It will try to parse the string as HTML and sanitize
-   * any links. If that fails, then whole string will treated as a simple text.
+   * Returns HTML rendering of a job or stage description. It will try to parse the string as HTML
+   * and make sure that it only contains anchors with relative links. Otherwise, the whole string
+   * will rendered as a simple escaped text.
    */
   def makeDescription(desc: String, basePathUri: String): NodeSeq = {
     import scala.language.postfixOps
@@ -426,7 +427,7 @@ private[spark] object UIUtils extends Logging {
         xml \\ "a" flatMap { _.attributes } filter { _.key == "href" } map { _.value.toString }
       if (allLinks.exists { ! _.startsWith ("/") }) {
         throw new IllegalArgumentException(
-          "Links in job descriptions must be relative:\n" + allLinks.mkString("\n\t"))
+          "Links in job descriptions must be root-relative:\n" + allLinks.mkString("\n\t"))
       }
 
       // Prepend the relative links with basePathUri
diff --git a/streaming/src/main/scala/org/apache/spark/streaming/scheduler/ReceiverTracker.scala b/streaming/src/main/scala/org/apache/spark/streaming/scheduler/ReceiverTracker.scala
index a76a8a1219328..204e6142fd6cf 100644
--- a/streaming/src/main/scala/org/apache/spark/streaming/scheduler/ReceiverTracker.scala
+++ b/streaming/src/main/scala/org/apache/spark/streaming/scheduler/ReceiverTracker.scala
@@ -30,7 +30,7 @@ import org.apache.spark.rdd.RDD
 import org.apache.spark.rpc._
 import org.apache.spark.streaming.{StreamingContext, Time}
 import org.apache.spark.streaming.receiver._
-import org.apache.spark.util.{ThreadUtils, SerializableConfiguration}
+import org.apache.spark.util.{Utils, ThreadUtils, SerializableConfiguration}
 
 
 /** Enumeration to identify current state of a Receiver */
@@ -555,7 +555,7 @@ class ReceiverTracker(ssc: StreamingContext, skipReceiverLaunch: Boolean = false
         }
       receiverRDD.setName(s"Receiver $receiverId")
       ssc.sparkContext.setJobDescription(s"Streaming job running receiver $receiverId")
-      ssc.sparkContext.setCallSite(ssc.getStartSite())
+      ssc.sparkContext.setCallSite(Option(ssc.getStartSite()).getOrElse(Utils.getCallSite()))
 
       val future = ssc.sparkContext.submitJob[Receiver[_], Unit, Unit](
         receiverRDD, startReceiverFunc, Seq(0), (_, _) => Unit, ())

From 606b2847f49c5a4e769af88a0054199050d2f880 Mon Sep 17 00:00:00 2001
From: Tathagata Das <tathagata.das1565@gmail.com>
Date: Tue, 22 Sep 2015 19:20:58 -0700
Subject: [PATCH 13/13] Addressed Andrew's comments

---
 core/src/main/scala/org/apache/spark/ui/UIUtils.scala     | 8 ++++++--
 .../src/test/scala/org/apache/spark/ui/UIUtilsSuite.scala | 3 ++-
 2 files changed, 8 insertions(+), 3 deletions(-)

diff --git a/core/src/main/scala/org/apache/spark/ui/UIUtils.scala b/core/src/main/scala/org/apache/spark/ui/UIUtils.scala
index f37e262a52adf..21dc8f0b65485 100644
--- a/core/src/main/scala/org/apache/spark/ui/UIUtils.scala
+++ b/core/src/main/scala/org/apache/spark/ui/UIUtils.scala
@@ -399,8 +399,12 @@ private[spark] object UIUtils extends Logging {
 
   /**
    * Returns HTML rendering of a job or stage description. It will try to parse the string as HTML
-   * and make sure that it only contains anchors with relative links. Otherwise, the whole string
-   * will rendered as a simple escaped text.
+   * and make sure that it only contains anchors with root-relative links. Otherwise,
+   * the whole string will rendered as a simple escaped text.
+   *
+   * Note: In terms of security, only anchor tags with root relative links are supported. So any
+   * attempts to embed links outside Spark UI, or other tags like <script> will cause in the whole
+   * description to be treated as plain text.
    */
   def makeDescription(desc: String, basePathUri: String): NodeSeq = {
     import scala.language.postfixOps
diff --git a/core/src/test/scala/org/apache/spark/ui/UIUtilsSuite.scala b/core/src/test/scala/org/apache/spark/ui/UIUtilsSuite.scala
index 39d9629b930cf..2b693c165180f 100644
--- a/core/src/test/scala/org/apache/spark/ui/UIUtilsSuite.scala
+++ b/core/src/test/scala/org/apache/spark/ui/UIUtilsSuite.scala
@@ -57,7 +57,8 @@ class UIUtilsSuite extends SparkFunSuite {
     )
   }
 
-  def verify(desc: String, expected: Elem, errorMsg: String = "", baseUrl: String = ""): Unit = {
+  private def verify(
+      desc: String, expected: Elem, errorMsg: String = "", baseUrl: String = ""): Unit = {
     val generated = makeDescription(desc, baseUrl)
     assert(generated.sameElements(expected),
       s"\n$errorMsg\n\nExpected:\n$expected\nGenerated:\n$generated")