[SPARK-28150][CORE] Log in user before getting delegation tokens.

This ensures that tokens are always created with an empty UGI, which
allows multiple contexts to be (sequentially) started from the same JVM.

Tested with code attached to the bug, and also usual kerberos tests.

Closes apache#24955 from vanzin/SPARK-28150.

Authored-by: Marcelo Vanzin <[email protected]>
Signed-off-by: Marcelo Vanzin <[email protected]>

Author: Marcelo Vanzin

core/src/main/scala/org/apache/spark/deploy/security/HadoopDelegationTokenManager.scala

@@ -136,13 +136,27 @@ private[spark] class HadoopDelegationTokenManager(
 
   /**
    * Fetch new delegation tokens for configured services, storing them in the given credentials.
-   * Tokens are fetched for the current logged in user.
    *
    * @param creds Credentials object where to store the delegation tokens.
-   * @return The time by which the tokens must be renewed.
    */
-  def obtainDelegationTokens(creds: Credentials): Long = {
-    delegationTokenProviders.values.flatMap { provider =>
+  def obtainDelegationTokens(creds: Credentials): Unit = {
+    val freshUGI = doLogin()
+    freshUGI.doAs(new PrivilegedExceptionAction[Unit]() {
+      override def run(): Unit = {
+        val (newTokens, _) = obtainDelegationTokens()
+        creds.addAll(newTokens)
+      }
+    })
+  }
+
+  /**
+   * Fetch new delegation tokens for configured services.
+   *
+   * @return 2-tuple (credentials with new tokens, time by which the tokens must be renewed)
+   */
+  private def obtainDelegationTokens(): (Credentials, Long) = {
+    val creds = new Credentials()
+    val nextRenewal = delegationTokenProviders.values.flatMap { provider =>
       if (provider.delegationTokensRequired(sparkConf, hadoopConf)) {
         provider.obtainDelegationTokens(hadoopConf, sparkConf, creds)
       } else {
@@ -151,6 +165,7 @@ private[spark] class HadoopDelegationTokenManager(
         None
       }
     }.foldLeft(Long.MaxValue)(math.min)
+    (creds, nextRenewal)
   }
 
   // Visible for testing.
@@ -228,8 +243,7 @@ private[spark] class HadoopDelegationTokenManager(
   private def obtainTokensAndScheduleRenewal(ugi: UserGroupInformation): Credentials = {
     ugi.doAs(new PrivilegedExceptionAction[Credentials]() {
       override def run(): Credentials = {
-        val creds = new Credentials()
-        val nextRenewal = obtainDelegationTokens(creds)
+        val (creds, nextRenewal) = obtainDelegationTokens()
 
         // Calculate the time when new credentials should be created, based on the configured
         // ratio.