sujith71955 commented on a change in pull request #23429: [SPARK-26432][CORE]
Obtain HBase delegation token operation compatible with HBase 2.x.x version API
URL: https://github.com/apache/spark/pull/23429#discussion_r245353023
##########
File path:
core/src/main/scala/org/apache/spark/deploy/security/HBaseDelegationTokenProvider.scala
##########
@@ -50,12 +53,70 @@ private[security] class HBaseDelegationTokenProvider
creds.addToken(token.getService, token)
} catch {
case NonFatal(e) =>
- logWarning(s"Failed to get token from service $serviceName", e)
+ logWarning(s"Failed to get token from service $serviceName" + "Trying
to invoke" +
+ " service with hbase connection ")
+ // Seems to be spark is trying to get the token from HBase 2.x.x
version or above where the
+ // obtainToken(Configuration conf) API is been removed. Lets try
obtaining the token from
+ // another compatible API of HBase service.
+ obtainDelegationTokensWithHBaseConn(hadoopConf, creds)
}
-
None
}
+ /**
+ * The HBase client API used in below method is introduced from HBase
0.98.9,1.0.0 version
+ * to invokethis api first connection object has to be retrieved from
ConnectionFactory and the
+ * same connection can be passed to
+ * Token<AuthenticationTokenIdentifier> obtainToken(Connection conn) API
+ *
+ * @param hadoopConf
+ * @param creds
+ */
+ def obtainDelegationTokensWithHBaseConn(hadoopConf: Configuration, creds:
Credentials): Unit = {
+ val mirror = universe.runtimeMirror(Utils.getContextOrSparkClassLoader)
+ // Token<AuthenticationTokenIdentifier> obtainToken(Configuration conf) is
a deprecated
+ // method and in Hbase 2.0.0 the method is already removed. there is one
more public
+ // consistent API Token<AuthenticationTokenIdentifier>
obtainToken(Connection conn),
+ // in TokenUtil class which can be invoked using reflection.
+ val obtainHBaseConnection = mirror.classLoader.
+ loadClass("org.apache.hadoop.hbase.client.ConnectionFactory").
+ getMethod("createConnection", classOf[Configuration])
+ val conn = obtainHBaseConnection.invoke(null, hbaseConf(hadoopConf))
+ try {
+ val arrayOfMethod: Array[Method] = ClassLoader.getSystemClassLoader
+
.loadClass("org.apache.hadoop.hbase.security.token.TokenUtil").getMethods
+ val obtainParamTypeClass = getClassParamType(arrayOfMethod)
+ if (null == obtainParamTypeClass) {
+ logDebug(s"Failed to invoke service to get token from service
$serviceName")
+ }
+ else {
+ val obtainTokenMethod = mirror.classLoader.
+ loadClass("org.apache.hadoop.hbase.security.token.TokenUtil").
+ getMethod("obtainToken", obtainParamTypeClass)
+ logDebug("Attempting to fetch HBase security token.")
+ val token = obtainTokenMethod.invoke(null, conn)
+ .asInstanceOf[Token[_ <: TokenIdentifier]]
+ logInfo(s"Get token from HBase: ${token.toString}")
+ creds.addToken(token.getService, token)
+ }
+ }
+ catch {
+ case NonFatal(e) =>
+ logError(s"Failed to get token from service $serviceName", e)
+ }
+ finally {
+ conn.asInstanceOf[Closeable].close()
+ }
+ }
+
+ def getClassParamType(arrayOfMethod: Array[Method]): Class[_] = {
Review comment:
yes, should be
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
