[LIVY-1063] Fix resource leak in Kubernetes service account token refresh

server/src/main/scala/org/apache/livy/utils/SparkKubernetesApp.scala

@@ -77,26 +77,36 @@ object SparkKubernetesApp extends Logging {
   }
 
   val RefreshServiceAccountTokenThread = new Thread() {
+    // Reuse the KubernetesClient created at startup; refresh its OAuth token in place
+    private val tokenRefreshIntervalMs = 300000L
+
     override def run(): Unit = {
-      while (true) {
-        var currentContext = new Context()
-        var currentContextName = new String
-        val config = kubernetesClient.getConfiguration
-        if (config.getCurrentContext != null) {
-          currentContext = config.getCurrentContext.getContext
-          currentContextName = config.getCurrentContext.getName
+      while (!Thread.currentThread().isInterrupted) {
+        try {
+          val config = kubernetesClient.getConfiguration
+          val contextName =
+            Option(config.getCurrentContext).map(_.getName).getOrElse("")
+          val newestConfig = Config.autoConfigure(contextName)
+          config.setOauthToken(newestConfig.getOauthToken)
+          info("Refreshed Kubernetes service account token.")
+        } catch {
+          case e: InterruptedException =>
+            Thread.currentThread().interrupt()
+            warn("RefreshServiceAccountTokenThread was interrupted, exiting.", e)
+            return
+          case NonFatal(e) =>
+            warn("Failed to refresh Kubernetes service account token, will retry.", e)
         }
 
-        var newAccessToken = new String
-        val newestConfig = Config.autoConfigure(currentContextName)
-        newAccessToken = newestConfig.getOauthToken
-        info("Refreshed Kubernetes service account token.")
-
-        config.setOauthToken(newAccessToken)
-        kubernetesClient = new DefaultKubernetesClient(config)
-
-        // Token will expire 1 hour default, community recommend to update every 5 minutes
-        Thread.sleep(300000)
+        // Sleep runs after both success and failure to avoid a hot retry loop.
+        try {
+          Thread.sleep(tokenRefreshIntervalMs)
+        } catch {
+          case e: InterruptedException =>
+            Thread.currentThread().interrupt()
+            warn("RefreshServiceAccountTokenThread was interrupted, exiting.", e)
+            return
+        }
       }
     }
   }