From c622a87c44e0621e1b3024fdca9b2aa3c508615b Mon Sep 17 00:00:00 2001
From: jerryshao <sshao@hortonworks.com>
Date: Wed, 29 Mar 2017 10:09:58 -0700
Subject: [PATCH] [SPARK-20059][YARN] Use the correct classloader for
HBaseCredentialProvider
## What changes were proposed in this pull request?
Currently we use system classloader to find HBase jars, if it is specified by `--jars`, then it will be failed with ClassNotFound issue. So here changing to use child classloader.
Also putting added jars and main jar into classpath of submitted application in yarn cluster mode, otherwise HBase jars specified with `--jars` will never be honored in cluster mode, and fetching tokens in client side will always be failed.
## How was this patch tested?
Unit test and local verification.
Author: jerryshao <sshao@hortonworks.com>
Closes #17388 from jerryshao/SPARK-20059.
---
.../main/scala/org/apache/spark/deploy/SparkSubmit.scala | 7 ++++++-
.../scala/org/apache/spark/deploy/SparkSubmitSuite.scala | 7 ++++++-
.../deploy/yarn/security/HBaseCredentialProvider.scala | 5 +++--
3 files changed, 15 insertions(+), 4 deletions(-)
diff --git a/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala b/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala
index 1e50eb6635..77005aa904 100644
--- a/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala
@@ -485,12 +485,17 @@ object SparkSubmit extends CommandLineUtils {
// In client mode, launch the application main class directly
// In addition, add the main application jar and any added jars (if any) to the classpath
- if (deployMode == CLIENT) {
+ // Also add the main application jar and any added jars to classpath in case YARN client
+ // requires these jars.
+ if (deployMode == CLIENT || isYarnCluster) {
childMainClass = args.mainClass
if (isUserJar(args.primaryResource)) {
childClasspath += args.primaryResource
}
if (args.jars != null) { childClasspath ++= args.jars.split(",") }
+ }
+
+ if (deployMode == CLIENT) {
if (args.childArgs != null) { childArgs ++= args.childArgs }
}
diff --git a/core/src/test/scala/org/apache/spark/deploy/SparkSubmitSuite.scala b/core/src/test/scala/org/apache/spark/deploy/SparkSubmitSuite.scala
index 9417930d02..a591b98bca 100644
--- a/core/src/test/scala/org/apache/spark/deploy/SparkSubmitSuite.scala
+++ b/core/src/test/scala/org/apache/spark/deploy/SparkSubmitSuite.scala
@@ -213,7 +213,12 @@ class SparkSubmitSuite
childArgsStr should include ("--arg arg1 --arg arg2")
childArgsStr should include regex ("--jar .*thejar.jar")
mainClass should be ("org.apache.spark.deploy.yarn.Client")
- classpath should have length (0)
+
+ // In yarn cluster mode, also adding jars to classpath
+ classpath(0) should endWith ("thejar.jar")
+ classpath(1) should endWith ("one.jar")
+ classpath(2) should endWith ("two.jar")
+ classpath(3) should endWith ("three.jar")
sysProps("spark.executor.memory") should be ("5g")
sysProps("spark.driver.memory") should be ("4g")
diff --git a/resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/security/HBaseCredentialProvider.scala b/resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/security/HBaseCredentialProvider.scala
index 5571df09a2..5adeb8e605 100644
--- a/resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/security/HBaseCredentialProvider.scala
+++ b/resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/security/HBaseCredentialProvider.scala
@@ -26,6 +26,7 @@ import org.apache.hadoop.security.token.{Token, TokenIdentifier}
import org.apache.spark.SparkConf
import org.apache.spark.internal.Logging
+import org.apache.spark.util.Utils
private[security] class HBaseCredentialProvider extends ServiceCredentialProvider with Logging {
@@ -36,7 +37,7 @@ private[security] class HBaseCredentialProvider extends ServiceCredentialProvide
sparkConf: SparkConf,
creds: Credentials): Option[Long] = {
try {
- val mirror = universe.runtimeMirror(getClass.getClassLoader)
+ val mirror = universe.runtimeMirror(Utils.getContextOrSparkClassLoader)
val obtainToken = mirror.classLoader.
loadClass("org.apache.hadoop.hbase.security.token.TokenUtil").
getMethod("obtainToken", classOf[Configuration])
@@ -60,7 +61,7 @@ private[security] class HBaseCredentialProvider extends ServiceCredentialProvide
private def hbaseConf(conf: Configuration): Configuration = {
try {
- val mirror = universe.runtimeMirror(getClass.getClassLoader)
+ val mirror = universe.runtimeMirror(Utils.getContextOrSparkClassLoader)
val confCreate = mirror.classLoader.
loadClass("org.apache.hadoop.hbase.HBaseConfiguration").
getMethod("create", classOf[Configuration])
--
GitLab