Skip to content

Commit 4e3f3ce

Browse files
jerryshaoMarcelo Vanzin
authored andcommitted
[SPARK-23679][YARN] Setting RM_HA_URLS for AmIpFilter to avoid redirect failure in YARN mode
## What changes were proposed in this pull request? YARN `AmIpFilter` adds a new parameter "RM_HA_URLS" to support RM HA, but Spark on YARN doesn't provide a such parameter, so it will be failed to redirect when running on RM HA. The detailed exception can be checked from JIRA. So here fixing this issue by adding "RM_HA_URLS" parameter. ## How was this patch tested? Local verification. Closes apache#22164 from jerryshao/SPARK-23679. Authored-by: jerryshao <[email protected]> Signed-off-by: Marcelo Vanzin <[email protected]>
1 parent de46df5 commit 4e3f3ce

File tree

1 file changed

+28
-1
lines changed

1 file changed

+28
-1
lines changed

resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/YarnRMClient.scala

Lines changed: 28 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ package org.apache.spark.deploy.yarn
1919

2020
import scala.collection.JavaConverters._
2121

22+
import org.apache.hadoop.conf.Configuration
2223
import org.apache.hadoop.yarn.api.records._
2324
import org.apache.hadoop.yarn.client.api.AMRMClient
2425
import org.apache.hadoop.yarn.client.api.AMRMClient.ContainerRequest
@@ -112,7 +113,16 @@ private[spark] class YarnRMClient extends Logging {
112113
val proxies = WebAppUtils.getProxyHostsAndPortsForAmFilter(conf)
113114
val hosts = proxies.asScala.map(_.split(":").head)
114115
val uriBases = proxies.asScala.map { proxy => prefix + proxy + proxyBase }
115-
Map("PROXY_HOSTS" -> hosts.mkString(","), "PROXY_URI_BASES" -> uriBases.mkString(","))
116+
val params =
117+
Map("PROXY_HOSTS" -> hosts.mkString(","), "PROXY_URI_BASES" -> uriBases.mkString(","))
118+
119+
// Handles RM HA urls
120+
val rmIds = conf.getStringCollection(YarnConfiguration.RM_HA_IDS).asScala
121+
if (rmIds != null && rmIds.nonEmpty) {
122+
params + ("RM_HA_URLS" -> rmIds.map(getUrlByRmId(conf, _)).mkString(","))
123+
} else {
124+
params
125+
}
116126
}
117127

118128
/** Returns the maximum number of attempts to register the AM. */
@@ -126,4 +136,21 @@ private[spark] class YarnRMClient extends Logging {
126136
}
127137
}
128138

139+
private def getUrlByRmId(conf: Configuration, rmId: String): String = {
140+
val addressPropertyPrefix = if (YarnConfiguration.useHttps(conf)) {
141+
YarnConfiguration.RM_WEBAPP_HTTPS_ADDRESS
142+
} else {
143+
YarnConfiguration.RM_WEBAPP_ADDRESS
144+
}
145+
146+
val addressWithRmId = if (rmId == null || rmId.isEmpty) {
147+
addressPropertyPrefix
148+
} else if (rmId.startsWith(".")) {
149+
throw new IllegalStateException(s"rmId $rmId should not already have '.' prepended.")
150+
} else {
151+
s"$addressPropertyPrefix.$rmId"
152+
}
153+
154+
conf.get(addressWithRmId)
155+
}
129156
}

0 commit comments

Comments
 (0)