[SPARK-12708][UI] Sorting task error in Stages Page when yarn mode.

If sort column contains slash(e.g. "Executor ID / Host") when yarn mode,sort fail with following message.

![spark-12708](https://cloud.githubusercontent.com/assets/6679275/12193320/80814f8c-b62a-11e5-9914-7bf3907029df.png)

Ｉt's similar to SPARK-4313 .

Author: root <root@R520T1.(none)>
Author: Koyo Yoshida <koyo0615@gmail.com>

Closes #10663 from yoshidakuy/SPARK-12708.

core/src/main/scala/org/apache/spark/ui/UIUtils.scala

@@ -17,6 +17,7 @@
 
 package org.apache.spark.ui
 
+import java.net.URLDecoder
 import java.text.SimpleDateFormat
 import java.util.{Date, Locale}
 
@@ -451,4 +452,19 @@ private[spark] object UIUtils extends Logging {
         <span class="description-input">{desc}</span>
     }
   }
+
+  /**
+   * Decode URLParameter if URL is encoded by YARN-WebAppProxyServlet.
+   * Due to YARN-2844: WebAppProxyServlet cannot handle urls which contain encoded characters
+   * Therefore we need to decode it until we get the real URLParameter.
+   */
+  def decodeURLParameter(urlParam: String): String = {
+    var param = urlParam
+    var decodedParam = URLDecoder.decode(param, "UTF-8")
+    while (param != decodedParam) {
+      param = decodedParam
+      decodedParam = URLDecoder.decode(param, "UTF-8")
+    }
+    param
+  }
 }

core/src/main/scala/org/apache/spark/ui/exec/ExecutorThreadDumpPage.scala

@@ -17,7 +17,6 @@
 
 package org.apache.spark.ui.exec
 
-import java.net.URLDecoder
 import javax.servlet.http.HttpServletRequest
 
 import scala.util.Try
@@ -30,18 +29,8 @@ private[ui] class ExecutorThreadDumpPage(parent: ExecutorsTab) extends WebUIPage
   private val sc = parent.sc
 
   def render(request: HttpServletRequest): Seq[Node] = {
-    val executorId = Option(request.getParameter("executorId")).map {
-      executorId =>
-        // Due to YARN-2844, "<driver>" in the url will be encoded to "%25253Cdriver%25253E" when
-        // running in yarn-cluster mode. `request.getParameter("executorId")` will return
-        // "%253Cdriver%253E". Therefore we need to decode it until we get the real id.
-        var id = executorId
-        var decodedId = URLDecoder.decode(id, "UTF-8")
-        while (id != decodedId) {
-          id = decodedId
-          decodedId = URLDecoder.decode(id, "UTF-8")
-        }
-        id
+    val executorId = Option(request.getParameter("executorId")).map { executorId =>
+      UIUtils.decodeURLParameter(executorId)
     }.getOrElse {
       throw new IllegalArgumentException(s"Missing executorId parameter")
     }

core/src/main/scala/org/apache/spark/ui/jobs/PoolPage.scala

@@ -31,8 +31,11 @@ private[ui] class PoolPage(parent: StagesTab) extends WebUIPage("pool") {
 
   def render(request: HttpServletRequest): Seq[Node] = {
     listener.synchronized {
-      val poolName = request.getParameter("poolname")
-      require(poolName != null && poolName.nonEmpty, "Missing poolname parameter")
+      val poolName = Option(request.getParameter("poolname")).map { poolname =>
+        UIUtils.decodeURLParameter(poolname)
+      }.getOrElse {
+        throw new IllegalArgumentException(s"Missing poolname parameter")
+      }
 
       val poolToActiveStages = listener.poolToActiveStages
       val activeStages = poolToActiveStages.get(poolName) match {
@@ -44,7 +47,9 @@ private[ui] class PoolPage(parent: StagesTab) extends WebUIPage("pool") {
         killEnabled = parent.killEnabled)
 
       // For now, pool information is only accessible in live UIs
-      val pools = sc.map(_.getPoolForName(poolName).get).toSeq
+      val pools = sc.map(_.getPoolForName(poolName).getOrElse {
+        throw new IllegalArgumentException(s"Unknown poolname: $poolName")
+      }).toSeq
       val poolTable = new PoolTable(pools, parent)
 
       val content =

core/src/main/scala/org/apache/spark/ui/jobs/PoolTable.scala

@@ -17,6 +17,8 @@
 
 package org.apache.spark.ui.jobs
 
+import java.net.URLEncoder
+
 import scala.collection.mutable.HashMap
 import scala.xml.Node
 
@@ -59,7 +61,7 @@ private[ui] class PoolTable(pools: Seq[Schedulable], parent: StagesTab) {
       case None => 0
     }
     val href = "%s/stages/pool?poolname=%s"
-      .format(UIUtils.prependBaseUri(parent.basePath), p.name)
+      .format(UIUtils.prependBaseUri(parent.basePath), URLEncoder.encode(p.name, "UTF-8"))
     <tr>
       <td>
         <a href={href}>{p.name}</a>

core/src/main/scala/org/apache/spark/ui/jobs/StagePage.scala

@@ -100,7 +100,9 @@ private[ui] class StagePage(parent: StagesTab) extends WebUIPage("stage") {
       val parameterTaskPrevPageSize = request.getParameter("task.prevPageSize")
 
       val taskPage = Option(parameterTaskPage).map(_.toInt).getOrElse(1)
-      val taskSortColumn = Option(parameterTaskSortColumn).getOrElse("Index")
+      val taskSortColumn = Option(parameterTaskSortColumn).map { sortColumn =>
+        UIUtils.decodeURLParameter(sortColumn)
+      }.getOrElse("Index")
       val taskSortDesc = Option(parameterTaskSortDesc).map(_.toBoolean).getOrElse(false)
       val taskPageSize = Option(parameterTaskPageSize).map(_.toInt).getOrElse(100)
       val taskPrevPageSize = Option(parameterTaskPrevPageSize).map(_.toInt).getOrElse(taskPageSize)

core/src/test/scala/org/apache/spark/ui/UIUtilsSuite.scala

@@ -67,6 +67,20 @@ class UIUtilsSuite extends SparkFunSuite {
       s"\nRunning progress bar should round down\n\nExpected:\n$expected\nGenerated:\n$generated")
   }
 
+  test("decodeURLParameter (SPARK-12708: Sorting task error in Stages Page when yarn mode.)") {
+    val encoded1 = "%252F"
+    val decoded1 = "/"
+    val encoded2 = "%253Cdriver%253E"
+    val decoded2 = "<driver>"
+
+    assert(decoded1 === decodeURLParameter(encoded1))
+    assert(decoded2 === decodeURLParameter(encoded2))
+
+    // verify that no affect to decoded URL.
+    assert(decoded1 === decodeURLParameter(decoded1))
+    assert(decoded2 === decodeURLParameter(decoded2))
+  }
+
   private def verify(
       desc: String, expected: Elem, errorMsg: String = "", baseUrl: String = ""): Unit = {
     val generated = makeDescription(desc, baseUrl)