[SPARK-3287] [YARN]When ResourceManager High Availability is enabled, Applicat...

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

@@ -17,7 +17,7 @@
 
 package org.apache.spark.ui
 
-import java.net.{InetSocketAddress, URL}
+import java.net.{InetSocketAddress, URL, URLDecoder}
 import javax.servlet.DispatcherType
 import javax.servlet.http.{HttpServlet, HttpServletRequest, HttpServletResponse}
 
@@ -147,15 +147,25 @@ private[spark] object JettyUtils extends Logging {
           val holder : FilterHolder = new FilterHolder()
           holder.setClassName(filter)
           // Get any parameters for each filter
-          val paramName = "spark." + filter + ".params"
-          val params = conf.get(paramName, "").split(',').map(_.trim()).toSet
+          var paramName = "spark." + filter + ".params"
+          var params = conf.get(paramName, "").split(',').map(_.trim()).toSet
           params.foreach {
             case param : String =>
               if (!param.isEmpty) {
                 val parts = param.split("=")
                 if (parts.length == 2) holder.setInitParameter(parts(0), parts(1))
              }
           }
+          paramName = "spark." + filter + ".encodedparams"
+          params = conf.get(paramName, "").split(',').map(_.trim()).toSet
+          params.foreach {
+            case param : String =>
+              if (!param.isEmpty) {
+                val parts = param.split("=")
+                if (parts.length == 2) holder.setInitParameter(parts(0),
+                    URLDecoder.decode(parts(1), "UTF-8"))
+             }
+          }
           val enumDispatcher = java.util.EnumSet.of(DispatcherType.ASYNC, DispatcherType.ERROR,
             DispatcherType.FORWARD, DispatcherType.INCLUDE, DispatcherType.REQUEST)
           handlers.foreach { case(handler) => handler.addFilter(holder, "/*", enumDispatcher) }

yarn/alpha/src/main/scala/org/apache/spark/deploy/yarn/YarnRMClientImpl.scala

@@ -75,7 +75,7 @@ private class YarnRMClientImpl(args: ApplicationMasterArguments) extends YarnRMC
   }
 
   override def getProxyHostAndPort(conf: YarnConfiguration) =
-    YarnConfiguration.getProxyHostAndPort(conf)
+    List(YarnConfiguration.getProxyHostAndPort(conf))
 
   override def getMaxRegAttempts(conf: YarnConfiguration) =
     conf.getInt(YarnConfiguration.RM_AM_MAX_RETRIES, YarnConfiguration.DEFAULT_RM_AM_MAX_RETRIES)

yarn/common/src/main/scala/org/apache/spark/deploy/yarn/ApplicationMaster.scala

@@ -19,6 +19,7 @@ package org.apache.spark.deploy.yarn
 
 import java.io.IOException
 import java.net.Socket
+import java.net.URLEncoder
 import java.util.concurrent.atomic.AtomicReference
 
 import scala.collection.JavaConversions._
@@ -32,6 +33,7 @@ import org.apache.hadoop.util.ShutdownHookManager
 import org.apache.hadoop.yarn.api._
 import org.apache.hadoop.yarn.api.records._
 import org.apache.hadoop.yarn.conf.YarnConfiguration
+import org.apache.hadoop.yarn.webapp.util.WebAppUtils
 
 import org.apache.spark.{Logging, SecurityManager, SparkConf, SparkContext, SparkEnv}
 import org.apache.spark.deploy.SparkHadoopUtil
@@ -324,17 +326,24 @@ private[spark] class ApplicationMaster(args: ApplicationMasterArguments,
   /** Add the Yarn IP filter that is required for properly securing the UI. */
   private def addAmIpFilter() = {
     val amFilter = "org.apache.hadoop.yarn.server.webproxy.amfilter.AmIpFilter"
-    val proxy = client.getProxyHostAndPort(yarnConf)
-    val parts = proxy.split(":")
-    val proxyBase = System.getenv(ApplicationConstants.APPLICATION_WEB_PROXY_BASE_ENV)
-    val uriBase = "http://" + proxy + proxyBase
-    val params = "PROXY_HOST=" + parts(0) + "," + "PROXY_URI_BASE=" + uriBase
-
+    val proxies = client.getProxyHostsAndPorts(yarnConf)
+    var sbProxies = new StringBuilder
+    var sbUrlBases = new StringBuilder
+    for (proxy <- proxies) {
+        sbProxies ++= proxy.split(":")(0)
+        sbProxies +=','
+        sbUrlBases ++= WebAppUtils.getHttpSchemePrefix(yarnConf)
+        sbUrlBases ++= proxy
+        sbUrlBases ++= System.getenv(ApplicationConstants.APPLICATION_WEB_PROXY_BASE_ENV)
+        sbUrlBases +=','
+    }
+    var params = "PROXY_HOSTS=" + URLEncoder.encode(sbProxies.toString(), "UTF-8") + ","
+    params ++= "PROXY_URI_BASES=" + URLEncoder.encode(sbUrlBases.toString(), "UTF-8")
     if (isDriver) {
       System.setProperty("spark.ui.filters", amFilter)
-      System.setProperty(s"spark.$amFilter.params", params)
+      System.setProperty(s"spark.$amFilter.encodedparams", params)
     } else {
-      actor ! AddWebUIFilter(amFilter, params, proxyBase)
+      actor ! AddWebUIFilter(amFilter, params, sbUrlBases.toString())
     }
   }
 

yarn/common/src/main/scala/org/apache/spark/deploy/yarn/YarnRMClient.scala

@@ -59,7 +59,7 @@ trait YarnRMClient {
   def getAttemptId(): ApplicationAttemptId
 
   /** Returns the RM's proxy host and port. */
-  def getProxyHostAndPort(conf: YarnConfiguration): String
+  def getProxyHostsAndPorts(conf: YarnConfiguration): scala.collection.mutable.Buffer[String]
 
   /** Returns the maximum number of attempts to register the AM. */
   def getMaxRegAttempts(conf: YarnConfiguration): Int

yarn/stable/src/main/scala/org/apache/spark/deploy/yarn/YarnRMClientImpl.scala

@@ -18,6 +18,7 @@
 package org.apache.spark.deploy.yarn
 
 import scala.collection.{Map, Set}
+import scala.collection.JavaConversions._
 
 import org.apache.hadoop.yarn.api._
 import org.apache.hadoop.yarn.api.protocolrecords._
@@ -68,7 +69,8 @@ private class YarnRMClientImpl(args: ApplicationMasterArguments) extends YarnRMC
     appAttemptId
   }
 
-  override def getProxyHostAndPort(conf: YarnConfiguration) = WebAppUtils.getProxyHostAndPort(conf)
+  override def getProxyHostsAndPorts(conf: YarnConfiguration) = 
+      WebAppUtils.getProxyHostsAndPortsForAmFilter(conf)
 
   override def getMaxRegAttempts(conf: YarnConfiguration) =
     conf.getInt(YarnConfiguration.RM_AM_MAX_ATTEMPTS, YarnConfiguration.DEFAULT_RM_AM_MAX_ATTEMPTS)