[SPARK-33195][UI] Fix stages/stage UI page fails because of uri parameters encoded twice

core/src/main/scala/org/apache/spark/status/api/v1/StagesResource.scala

@@ -16,6 +16,7 @@
  */
 package org.apache.spark.status.api.v1
 
+import java.net.{URLDecoder, URLEncoder}
 import java.util.{HashMap, List => JList, Locale}
 import javax.ws.rs.{NotFoundException => _, _}
 import javax.ws.rs.core.{Context, MediaType, MultivaluedMap, UriInfo}
@@ -152,10 +153,11 @@ private[v1] class StagesResource extends BaseAppResource {
       // information like the columns to be sorted, search value typed by the user in the search
       // box, pagination index etc. For more information on these query parameters,
       // refer https://datatables.net/manual/server-side.
-      if (uriQueryParameters.getFirst("search[value]") != null &&
-        uriQueryParameters.getFirst("search[value]").length > 0) {
+      searchValue = encodeKeyAndGetValue(uriQueryParameters, "search[value]", null)
+      if (searchValue != null && searchValue.length > 0) {
         isSearch = true
-        searchValue = uriQueryParameters.getFirst("search[value]")
+      } else {
+        searchValue = null
       }
       val _tasksToShow: Seq[TaskData] = doPagination(uriQueryParameters, stageId, stageAttemptId,
         isSearch, totalRecords.toInt)
@@ -185,15 +187,37 @@ private[v1] class StagesResource extends BaseAppResource {
     }
   }
 
+  // The request URL can be raw or encoded. To avoid the parameter key being
+  // encoded twice in queryParameters, try to encode it at most twice and lookup
+  // it in the queryParameters.
+  def encodeKeyAndGetValue(queryParameters: MultivaluedMap[String, String],
+    key: String, defaultValue: String): String = {
+    var value = queryParameters.getFirst(key)
+    if (value == null) {
+      var encodedKey = URLEncoder.encode(key, "UTF-8")
+      value = queryParameters.getFirst(encodedKey)
+      if (value == null) {
+        encodedKey = URLEncoder.encode(encodedKey, "UTF-8")
+        value = queryParameters.getFirst(encodedKey)
+        if (value == null) {
+          value = defaultValue
+        }
+      }
+    }
+    value
+  }
+
   // Performs pagination on the server side
   def doPagination(queryParameters: MultivaluedMap[String, String], stageId: Int,
     stageAttemptId: Int, isSearch: Boolean, totalRecords: Int): Seq[TaskData] = {
     var columnNameToSort = queryParameters.getFirst("columnNameToSort")
+    columnNameToSort = URLDecoder.decode(columnNameToSort, "UTF-8")
+    columnNameToSort = URLDecoder.decode(columnNameToSort, "UTF-8")
     // Sorting on Logs column will default to Index column sort
     if (columnNameToSort.equalsIgnoreCase("Logs")) {
       columnNameToSort = "Index"
     }
-    val isAscendingStr = queryParameters.getFirst("order[0][dir]")
+    val isAscendingStr = encodeKeyAndGetValue(queryParameters, "order[0][dir]", "asc")
     var pageStartIndex = 0
     var pageLength = totalRecords
     // We fetch only the desired rows upto the specified page length for all cases except when a

core/src/test/scala/org/apache/spark/status/api/v1/StagesResourceSuite.scala

@@ -0,0 +1,67 @@
+/*
+ * 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.status.api.v1
+
+import java.util.Arrays
+import javax.ws.rs.core.MultivaluedHashMap
+
+import org.scalatest.matchers.must.Matchers
+
+import org.apache.spark.SparkFunSuite
+
+class StagesResourceSuite extends SparkFunSuite with Matchers {
+
+  test("SPARK-33195: Avoid stages/stage encoding URL twice") {
+    val stageResource = new StagesResource()
+
+    // parameters not encoded
+    val queryParameters1 = new MultivaluedHashMap[String, String]()
+    queryParameters1.put("search[value]", Arrays.asList("boo"))
+    queryParameters1.put("order[0][column]", Arrays.asList("1"))
+    queryParameters1.put("order[0][dir]", Arrays.asList("desc"))
+    assert(stageResource.encodeKeyAndGetValue(
+      queryParameters1, "search[value]", "foo") == "boo")
+    assert(stageResource.encodeKeyAndGetValue(
+      queryParameters1, "order[0][column]", "0") == "1")
+    assert(stageResource.encodeKeyAndGetValue(
+      queryParameters1, "order[0][dir]", "asc") == "desc")
+
+    // parameters encoded once
+    val queryParameters2 = new MultivaluedHashMap[String, String]()
+    queryParameters2.put("search%5Bvalue%5D", Arrays.asList("boo"))
+    queryParameters2.put("order%5B0%5D%5Bcolumn%5D", Arrays.asList("1"))
+    queryParameters2.put("order%5B0%5D%5Bdir%5D", Arrays.asList("desc"))
+    assert(stageResource.encodeKeyAndGetValue(
+      queryParameters2, "search[value]", "foo") == "boo")
+    assert(stageResource.encodeKeyAndGetValue(
+      queryParameters2, "order[0][column]", "0") == "1")
+    assert(stageResource.encodeKeyAndGetValue(
+      queryParameters2, "order[0][dir]", "asc") == "desc")
+
+    // parameters encoded twice
+    val queryParameters3 = new MultivaluedHashMap[String, String]()
+    queryParameters3.put("search%255Bvalue%255D", Arrays.asList("boo"))
+    queryParameters3.put("order%255B0%255D%255Bcolumn%255D", Arrays.asList("1"))
+    queryParameters3.put("order%255B0%255D%255Bdir%255D", Arrays.asList("desc"))
+    assert(stageResource.encodeKeyAndGetValue(
+      queryParameters3, "search[value]", "foo") == "boo")
+    assert(stageResource.encodeKeyAndGetValue(
+      queryParameters3, "order[0][column]", "0") == "1")
+    assert(stageResource.encodeKeyAndGetValue(
+      queryParameters3, "order[0][dir]", "asc") == "desc")
+  }
+}