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_r247312012
##########
File path:
core/src/main/scala/org/apache/spark/deploy/security/HBaseDelegationTokenProvider.scala
##########
@@ -50,12 +52,59 @@ 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. " +
+ s"Retrying 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 invoke this 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
+ */
+ private def obtainDelegationTokensWithHBaseConn(
+ hadoopConf: Configuration, creds: Credentials): Unit = {
+ val mirror = universe.runtimeMirror(Utils.getContextOrSparkClassLoader)
+ var connectionHbaseClient = null.asInstanceOf[Closeable]
+ try {
+ // 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 connectionFactoryClass = mirror.classLoader.
+ loadClass("org.apache.hadoop.hbase.client.ConnectionFactory").
+ getMethod("createConnection", classOf[Configuration])
+ connectionHbaseClient = connectionFactoryClass.invoke(null,
hbaseConf(hadoopConf))
+ .asInstanceOf[Closeable]
+ val connectionParamTypeClassRef = mirror.classLoader
+ .loadClass("org.apache.hadoop.hbase.client.Connection")
+ val obtainTokenMethod = mirror.classLoader.
+ loadClass("org.apache.hadoop.hbase.security.token.TokenUtil").
+ getMethod("obtainToken", connectionParamTypeClassRef)
+ logDebug("Attempting to fetch HBase security token.")
+ val token = obtainTokenMethod.invoke(null, connectionHbaseClient)
+ .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)
Review comment:
I thought both the attempts of retrieving delegation token got failed, so
basically spark will not able interact with hbase service in this scenario
which will in-turn fail the user attempt to do some operations based on hbase.
so i thought better to log this as error instead of just warning. in the
scenario warning user can still successfully do his operations with some
limitations at max. anyways i updated the level to error as the old code also
was logging this failure as warning. Thanks.
----------------------------------------------------------------
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]
