Skip to content
Closed
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,8 @@ import scala.util.Try

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.hadoop.ipc.RemoteException
import org.apache.hadoop.ipc.StandbyException
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This two line of imports can be merged into one line.

import org.apache.hadoop.mapred.Master
import org.apache.hadoop.security.Credentials
import org.apache.hadoop.security.token.delegation.AbstractDelegationTokenIdentifier
Expand All @@ -48,9 +50,16 @@ private[security] class HadoopFSCredentialProvider
val tmpCreds = new Credentials()
val tokenRenewer = getTokenRenewer(hadoopConf)
hadoopFSsToAccess(hadoopConf, sparkConf).foreach { dst =>
val dstFs = dst.getFileSystem(hadoopConf)
logInfo("getting token for: " + dst)
dstFs.addDelegationTokens(tokenRenewer, tmpCreds)
try {
val dstFs = dst.getFileSystem(hadoopConf)
logInfo("getting token for: " + dst)
dstFs.addDelegationTokens(tokenRenewer, tmpCreds)
} catch {
case e: StandbyException =>
logWarning(s"Can't get token from ${dst} for it is in state standby", e)
case e: RemoteException =>
logWarning(s"Can't get token from ${dst}", e)
}
}

// Get the token renewal interval if it is not set. It will only be called once.
Expand Down Expand Up @@ -81,8 +90,15 @@ private[security] class HadoopFSCredentialProvider
sparkConf.get(PRINCIPAL).flatMap { renewer =>
val creds = new Credentials()
hadoopFSsToAccess(hadoopConf, sparkConf).foreach { dst =>
val dstFs = dst.getFileSystem(hadoopConf)
dstFs.addDelegationTokens(renewer, creds)
try {
val dstFs = dst.getFileSystem(hadoopConf)
dstFs.addDelegationTokens(renewer, creds)
} catch {
case e: StandbyException =>
logWarning(s"Can't get token from ${dst} for it is in state standby", e)
case e: RemoteException =>
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'd suggested adding a handler for UnknownHostException too, but now I think that could hide problems with client config. Best to leave as is.

logWarning(s"Can't get token from ${dst}", e)
}
}

val renewIntervals = creds.getAllTokens.asScala.filter {
Expand Down