[KYUUBI #3219] Error renew delegation tokens: Unknown version of delegation token 8

### _Why are the changes needed?_
Spark SQL Engine failed to get issue date from some DelegationTokenIdentifiers, such as OzoneTokenIdentifier.

### _How was this patch tested?_
- [ ] Add some test cases that check the changes thoroughly including negative and positive cases if possible

- [ ] Add screenshots for manual tests if appropriate

- [x] [Run test](https://kyuubi.apache.org/docs/latest/develop_tools/testing.html#running-tests) locally before make a pull request

Closes #3225 from zhouyifan279/3219.

Closes #3219

0ffdcf9 [zhouyifan279] [KYUUBI #3219] Error renew delegation tokens: Unknown version of delegation token 8
c5a37a0 [zhouyifan279] [KYUUBI #3219] Error renew delegation tokens: Unknown version of delegation token 8
594c5a0 [zhouyifan279] [KYUUBI #3219] Error renew delegation tokens: Unknown version of delegation token 8
cbbafd5 [zhouyifan279] [KYUUBI #3219] Error renew delegation tokens: Unknown version of delegation token 8
394308e [zhouyifan279] [KYUUBI #3219] Error renew delegation tokens: Unknown version of delegation token 8

Authored-by: zhouyifan279 <zhouyifan279@gmail.com>
Signed-off-by: Cheng Pan <chengpan@apache.org>

Author: zhouyifan279

externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/SparkTBinaryFrontendService.scala

@@ -141,8 +141,7 @@ object SparkTBinaryFrontendService extends Logging {
         }
         .map(_._2)
       newToken.foreach { token =>
-        if (KyuubiHadoopUtils.getTokenIssueDate(token) >
-            KyuubiHadoopUtils.getTokenIssueDate(oldAliasAndToken.get._2)) {
+        if (compareIssueDate(token, oldAliasAndToken.get._2) > 0) {
           updateCreds.addToken(oldAliasAndToken.get._1, token)
         } else {
           warn(s"Ignore Hive token with earlier issue date: $token")
@@ -166,8 +165,7 @@ object SparkTBinaryFrontendService extends Logging {
     tokens.foreach { case (alias, newToken) =>
       val oldToken = oldCreds.getToken(alias)
       if (oldToken != null) {
-        if (KyuubiHadoopUtils.getTokenIssueDate(newToken) >
-            KyuubiHadoopUtils.getTokenIssueDate(oldToken)) {
+        if (compareIssueDate(newToken, oldToken) > 0) {
           updateCreds.addToken(alias, newToken)
         } else {
           warn(s"Ignore token with earlier issue date: $newToken")
@@ -177,4 +175,16 @@ object SparkTBinaryFrontendService extends Logging {
       }
     }
   }
+
+  private def compareIssueDate(
+      newToken: Token[_ <: TokenIdentifier],
+      oldToken: Token[_ <: TokenIdentifier]): Int = {
+    val newDate = KyuubiHadoopUtils.getTokenIssueDate(newToken)
+    val oldDate = KyuubiHadoopUtils.getTokenIssueDate(oldToken)
+    if (newDate.isDefined && oldDate.isDefined && newDate.get <= oldDate.get) {
+      -1
+    } else {
+      1
+    }
+  }
 }

kyuubi-common/src/main/scala/org/apache/kyuubi/util/KyuubiHadoopUtils.scala

@@ -21,17 +21,20 @@ import java.io.{ByteArrayInputStream, ByteArrayOutputStream, DataInputStream, Da
 import java.util.{Base64, Map => JMap}
 
 import scala.collection.JavaConverters._
+import scala.util.{Failure, Success, Try}
 
 import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.hdfs.security.token.delegation.DelegationTokenIdentifier
 import org.apache.hadoop.io.Text
 import org.apache.hadoop.security.{Credentials, SecurityUtil, UserGroupInformation}
 import org.apache.hadoop.security.token.{Token, TokenIdentifier}
+import org.apache.hadoop.security.token.delegation.AbstractDelegationTokenIdentifier
 import org.apache.hadoop.yarn.conf.YarnConfiguration
 
+import org.apache.kyuubi.Logging
 import org.apache.kyuubi.config.KyuubiConf
 
-object KyuubiHadoopUtils {
+object KyuubiHadoopUtils extends Logging {
 
   private val subjectField =
     classOf[UserGroupInformation].getDeclaredField("subject")
@@ -85,13 +88,25 @@ object KyuubiHadoopUtils {
       .toMap
   }
 
-  def getTokenIssueDate(token: Token[_ <: TokenIdentifier]): Long = {
-    // It is safe to deserialize any token identifier to hdfs `DelegationTokenIdentifier`
-    // as all token identifiers have the same binary format.
-    val tokenIdentifier = new DelegationTokenIdentifier
-    val buf = new ByteArrayInputStream(token.getIdentifier)
-    val in = new DataInputStream(buf)
-    tokenIdentifier.readFields(in)
-    tokenIdentifier.getIssueDate
+  def getTokenIssueDate(token: Token[_ <: TokenIdentifier]): Option[Long] = {
+    token.decodeIdentifier() match {
+      case tokenIdent: AbstractDelegationTokenIdentifier =>
+        Some(tokenIdent.getIssueDate)
+      case null =>
+        // TokenIdentifiers not found in ServiceLoader
+        val tokenIdentifier = new DelegationTokenIdentifier
+        val buf = new ByteArrayInputStream(token.getIdentifier)
+        val in = new DataInputStream(buf)
+        Try(tokenIdentifier.readFields(in)) match {
+          case Success(_) =>
+            Some(tokenIdentifier.getIssueDate)
+          case Failure(e) =>
+            warn(s"Can not decode identifier of token $token", e)
+            None
+        }
+      case tokenIdent =>
+        debug(s"Unsupported TokenIdentifier kind: ${tokenIdent.getKind}")
+        None
+    }
   }
 }

kyuubi-common/src/test/scala/org/apache/kyuubi/util/KyuubiHadoopUtilsSuite.scala

@@ -17,13 +17,17 @@
 
 package org.apache.kyuubi.util
 
+import java.io.{DataInput, DataOutput}
 import java.util.stream.StreamSupport
 
 import scala.util.Random
 
+import org.apache.hadoop.hdfs.security.token.block.BlockTokenIdentifier
+import org.apache.hadoop.hdfs.security.token.delegation.{DelegationTokenIdentifier => HDFSTokenIdent}
 import org.apache.hadoop.io.Text
 import org.apache.hadoop.security.Credentials
-import org.apache.hadoop.security.token.Token
+import org.apache.hadoop.security.token.{Token, TokenIdentifier}
+import org.apache.hadoop.security.token.delegation.AbstractDelegationTokenIdentifier
 
 import org.apache.kyuubi.KyuubiFunSuite
 import org.apache.kyuubi.config.KyuubiConf
@@ -75,4 +79,52 @@ class KyuubiHadoopUtilsSuite extends KyuubiFunSuite {
     assert(StreamSupport.stream(hadoopConf.spliterator(), false)
       .noneMatch(entry => entry.getKey.startsWith("hadoop") || entry.getKey.startsWith("fs")))
   }
+
+  test("get token issue date") {
+    val issueDate = System.currentTimeMillis()
+
+    def checkIssueDate(tokenIdent: TokenIdentifier, expected: Option[Long]): Unit = {
+      val hdfsToken = new Token[HDFSTokenIdent]()
+      hdfsToken.setKind(tokenIdent.getKind)
+      hdfsToken.setID(tokenIdent.getBytes)
+      assert(KyuubiHadoopUtils.getTokenIssueDate(hdfsToken) == expected)
+    }
+
+    // DelegationTokenIdentifier found in ServiceLoader
+    // Such as HDFS_DELEGATION_TOKEN, OzoneToken
+    val hdfsTokenIdent = new HDFSTokenIdent()
+    hdfsTokenIdent.setIssueDate(issueDate)
+    checkIssueDate(hdfsTokenIdent, Some(issueDate))
+
+    // TokenIdentifier with no issue date found in ServiceLoader
+    val blockTokenIdent = new BlockTokenIdentifier()
+    checkIssueDate(blockTokenIdent, None)
+
+    // DelegationTokenIdentifier not found in ServiceLoader
+    // Such as HIVE_DELEGATION_TOKEN
+    val testTokenIdent = new TestDelegationTokenIdentifier()
+    testTokenIdent.setIssueDate(issueDate)
+    checkIssueDate(testTokenIdent, Some(issueDate))
+
+    // DelegationTokenIdentifier with custom binary format and not found in ServiceLoader
+    val testTokenIdent2 = new TestDelegationTokenIdentifier2()
+    testTokenIdent2.setIssueDate(issueDate)
+    checkIssueDate(testTokenIdent2, None)
+  }
+}
+
+private class TestDelegationTokenIdentifier extends AbstractDelegationTokenIdentifier {
+  override def getKind: Text = new Text("KYUUBI_TOKEN_NOT_IN_SERVICE_LOADER")
+}
+
+private class TestDelegationTokenIdentifier2 extends AbstractDelegationTokenIdentifier {
+  override def getKind: Text = new Text("KYUUBI_TOKEN_OVERRIDE_WRITE")
+
+  override def write(out: DataOutput): Unit = {
+    out.writeLong(getIssueDate)
+  }
+
+  override def readFields(in: DataInput): Unit = {
+    setIssueDate(in.readLong())
+  }
 }