[KYUUBI #2222] Refactor the log when failing to get hadoop fs delegat…

…ion token

### _Why are the changes needed?_

Refactor the log when failing to get hadoop fs delegation token

### _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

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

Closes #2222 from turboFei/part_fs_token.

Closes #2222

3f5c61a [Fei Wang] dedup
9b3d3e6 [Fei Wang] refactor prompt
618d6e6 [Fei Wang] save

Authored-by: Fei Wang <fwang12@ebay.com>
Signed-off-by: Fei Wang <fwang12@ebay.com>

kyuubi-server/src/main/scala/org/apache/kyuubi/credentials/HadoopFsDelegationTokenProvider.scala

@@ -29,7 +29,7 @@ import org.apache.hadoop.hdfs.HdfsConfiguration
 import org.apache.hadoop.security.{Credentials, SecurityUtil, UserGroupInformation}
 import org.apache.hadoop.security.UserGroupInformation.AuthenticationMethod
 
-import org.apache.kyuubi.Logging
+import org.apache.kyuubi.{KyuubiException, Logging}
 import org.apache.kyuubi.config.KyuubiConf
 import org.apache.kyuubi.credentials.HadoopFsDelegationTokenProvider.{doAsProxyUser, validatedFsUris}
 import org.apache.kyuubi.util.KyuubiHadoopUtils
@@ -63,19 +63,26 @@ class HadoopFsDelegationTokenProvider extends HadoopDelegationTokenProvider with
 
   override def obtainDelegationTokens(owner: String, creds: Credentials): Unit = {
     doAsProxyUser(owner) {
-      val fileSystems = fsUris.map(FileSystem.get(_, hadoopConf)).toSet
+      val fileSystems = fsUris.map { uri =>
+        FileSystem.get(uri, hadoopConf) -> uri
+      }.toMap
 
       try {
         // Renewer is not needed. But setting a renewer can avoid potential NPE.
         val renewer = UserGroupInformation.getCurrentUser.getUserName
-        fileSystems.foreach { fs =>
-          info(s"getting token owned by $owner for: $fs")
-          fs.addDelegationTokens(renewer, creds)
+        fileSystems.foreach { case (fs, uri) =>
+          info(s"getting token owned by $owner for: $uri")
+          try {
+            fs.addDelegationTokens(renewer, creds)
+          } catch {
+            case e: Exception =>
+              throw new KyuubiException(s"Failed to get token owned by $owner for: $uri", e)
+          }
         }
       } finally {
         // Token renewal interval is longer than FileSystems' underlying connections' max idle time.
         // Close FileSystems won't lose efficiency.
-        fileSystems.foreach(_.close())
+        fileSystems.keys.foreach(_.close())
       }
     }
   }