gaborgsomogyi commented on a change in pull request #24305: [SPARK-27294][SS]
Add multi-cluster Kafka delegation token
URL: https://github.com/apache/spark/pull/24305#discussion_r280528975
##########
File path:
external/kafka-0-10-token-provider/src/main/scala/org/apache/spark/kafka010/KafkaDelegationTokenProvider.scala
##########
@@ -38,24 +37,49 @@ private[spark] class KafkaDelegationTokenProvider
sparkConf: SparkConf,
creds: Credentials): Option[Long] = {
try {
- logDebug("Attempting to fetch Kafka security token.")
- val (token, nextRenewalDate) = KafkaTokenUtil.obtainToken(sparkConf)
- creds.addToken(token.getService, token)
- return Some(nextRenewalDate)
+ var lowestNextRenewalDate: Option[Long] = None
+ KafkaTokenSparkConf.getAllClusterConfigs(sparkConf).foreach {
clusterConf =>
+ if (delegationTokensRequired(clusterConf)) {
+ try {
+ logDebug(
+ s"Attempting to fetch Kafka security token for cluster
${clusterConf.identifier}.")
+ val (token, nextRenewalDate) =
KafkaTokenUtil.obtainToken(sparkConf, clusterConf)
+ creds.addToken(token.getService, token)
+ if (lowestNextRenewalDate.isEmpty || nextRenewalDate <
lowestNextRenewalDate.get) {
+ lowestNextRenewalDate = Some(nextRenewalDate)
+ }
+ } catch {
+ case NonFatal(e) =>
+ logWarning(s"Failed to get token from service: $serviceName " +
+ s"cluster: ${clusterConf.identifier}", e)
+ }
+ } else {
+ logDebug(
+ s"Cluster ${clusterConf.identifier} does not require delegation
token, skipping.")
+ }
+ }
+ lowestNextRenewalDate
} catch {
case NonFatal(e) =>
- logWarning(s"Failed to get token from service $serviceName", e)
+ logWarning(s"Failed to get token cluster configuration", e)
+ None
Review comment:
Having 2 level of error catching is added for the following reasons:
* Outer catch protects delegation token manager from exception coming from
for example `getAllClusterConfigs`
* Inner catch protects the inner loop and just skips a particular cluster
What I think is nice catch that `delegationTokensRequired` is not covered by
the inner block. As a result in the first token obtained properly but the
second throws exception in `delegationTokensRequired` then the whole function
will return `None`.
> That would also avoid the two levels of error catching you currently have.
Don't know how this could work with one level of catch block. Such case if
first fails the rest won't be processed. I've modified the code and share your
thoughts.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to 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]
