dongjoon-hyun commented on code in PR #57285:
URL: https://github.com/apache/spark/pull/57285#discussion_r3641946836
##########
core/src/main/scala/org/apache/spark/deploy/security/HadoopDelegationTokenManager.scala:
##########
@@ -226,24 +266,40 @@ private[spark] class HadoopDelegationTokenManager(
*
* @return Credentials containing the new tokens.
*/
- private def obtainTokensAndScheduleRenewal(ugi: UserGroupInformation):
Credentials = {
- ugi.doAs(new PrivilegedExceptionAction[Credentials]() {
- override def run(): Credentials = {
- val (creds, nextRenewal) = obtainDelegationTokens()
-
- // Calculate the time when new credentials should be created, based on
the configured
- // ratio.
- val now = System.currentTimeMillis
- val ratio = sparkConf.get(CREDENTIALS_RENEWAL_INTERVAL_RATIO)
- val delay = (ratio * (nextRenewal - now)).toLong
- logInfo(log"Calculated delay on renewal is ${MDC(LogKeys.DELAY,
delay)}," +
- log" based on next renewal ${MDC(LogKeys.NEXT_RENEWAL_TIME,
nextRenewal)}" +
- log" and the ratio ${MDC(LogKeys.CREDENTIALS_RENEWAL_INTERVAL_RATIO,
ratio)}," +
- log" and current time ${MDC(LogKeys.CURRENT_TIME, now)}")
- scheduleRenewal(delay)
- creds
+ private def obtainTokensAndScheduleRenewal(): Credentials = {
+ val (creds, nextRenewal, failureCount) = if (hasKerberosCredentials) {
+ val freshUGI = doLogin()
+ val currentUser = UserGroupInformation.getCurrentUser()
+ val result = freshUGI.doAs(
+ new PrivilegedExceptionAction[(Credentials, Long, Int)]() {
+ override def run(): (Credentials, Long, Int) = {
+ obtainDelegationTokens()
+ }
+ })
+ if (!currentUser.equals(freshUGI)) {
+ FileSystem.closeAllForUGI(freshUGI)
}
- })
+ result
+ } else if (sparkConf.get(CREDENTIALS_DIRECT_PROVIDERS_ENABLED)) {
+ obtainDelegationTokens(isolateFailures = true)
+ } else {
+ (new Credentials(), Long.MaxValue, 0)
Review Comment:
Is this valid?
This is adjacent to but distinct from @peter-toth 's finding 7: that fix
keys off failureCount, which is 0 here because no provider is ever invoked.
This fallback branch silently stops renewal. It returns `(new Credentials(),
Long.MaxValue, 0)`, so the code below sends an *empty* `UpdateDelegationTokens`
to the executors and — since `failureCount == 0`, the retry-wait path from
finding 7 doesn't trigger — schedules the next renewal at `0.75 *
(Long.MaxValue - now)`, i.e. effectively never. There is no recovery short of a
driver restart.
The branch is reachable: with `spark.kerberos.renewal.credentials=ccache`
and `directProviders.enabled=false`, `renewalEnabled` is true at `start()`, but
if the ticket cache expires before a later renewal fires,
`hasKerberosCredentials` becomes false and this branch is taken. Previously
this scenario threw from `doLogin()`/the providers, and `updateTokensTask`'s
catch rescheduled after `CREDENTIALS_RENEWAL_RETRY_WAIT` — so a user who
refreshed their ticket cache would recover on the next retry. That retry loop
is lost here, which is a regression for existing ccache deployments.
Suggest throwing instead, so the existing catch in `updateTokensTask`
handles it and keeps retrying:
```suggestion
// Reachable e.g. when ccache-based Kerberos credentials expired after
start().
// Throw so that updateTokensTask() reschedules a retry instead of
silently
// distributing empty credentials and never renewing again.
throw new IllegalStateException("Cannot obtain delegation tokens: no
Kerberos " +
"credentials are available and direct credential providers are not
enabled.")
```
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
