[fix][COM][hadoop-common] resolve kerberos tgt expiry issue for cached hdfs filesystem

linkis-commons/linkis-hadoop-common/src/main/java/org/apache/linkis/hadoop/common/utils/KerberosTgtUtils.java

@@ -0,0 +1,99 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.linkis.hadoop.common.utils;
+
+import org.apache.hadoop.security.UserGroupInformation;
+
+import javax.security.auth.Subject;
+import javax.security.auth.kerberos.KerberosTicket;
+
+import java.lang.reflect.Method;
+import java.util.Set;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class KerberosTgtUtils {
+
+  private static final Logger logger = LoggerFactory.getLogger(KerberosTgtUtils.class);
+
+  private static volatile Method getSubjectMethod = null;
+  private static volatile boolean getSubjectMethodResolved = false;
+
+  /**
+   * Get the Subject from UserGroupInformation via reflection, since getSubject() is protected in
+   * Hadoop 2.7.x.
+   */
+  private static Subject getSubjectFromUgi(UserGroupInformation ugi) {
+    if (!getSubjectMethodResolved) {
+      synchronized (KerberosTgtUtils.class) {
+        if (!getSubjectMethodResolved) {
+          try {
+            Method method = UserGroupInformation.class.getDeclaredMethod("getSubject");
+            method.setAccessible(true);
+            getSubjectMethod = method;
+          } catch (NoSuchMethodException e) {
+            logger.warn("Failed to find getSubject method on UserGroupInformation", e);
+            getSubjectMethod = null;
+          }
+          getSubjectMethodResolved = true;
+        }
+      }
+    }
+    if (getSubjectMethod == null) {
+      return null;
+    }
+    try {
+      return (Subject) getSubjectMethod.invoke(ugi);
+    } catch (Exception e) {
+      logger.warn("Failed to invoke getSubject on UserGroupInformation", e);
+      return null;
+    }
+  }
+
+  /**
+   * Check if the Kerberos TGT in the given UGI is still valid (not expired). For proxy UGI, the
+   * real user's TGT is checked.
+   *
+   * @param ugi the UserGroupInformation to check
+   * @return true if UGI is null, security not enabled, or TGT is still valid; false if TGT has
+   *     expired
+   */
+  public static boolean isTgtValid(UserGroupInformation ugi) {
+    if (ugi == null || !UserGroupInformation.isSecurityEnabled()) {
+      return true;
+    }
+    // For proxy UGI, check the real user's tickets
+    UserGroupInformation checkUgi = ugi.getRealUser() != null ? ugi.getRealUser() : ugi;
+    Subject subject = getSubjectFromUgi(checkUgi);
+    if (subject == null) {
+      return true;
+    }
+    Set<KerberosTicket> tickets = subject.getPrivateCredentials(KerberosTicket.class);
+    if (tickets == null || tickets.isEmpty()) {
+      return true;
+    }
+    long now = System.currentTimeMillis();
+    for (KerberosTicket ticket : tickets) {
+      if (ticket.getEndTime().getTime() <= now) {
+        return false;
+      }
+    }
+    return true;
+  }
+}

linkis-commons/linkis-hadoop-common/src/main/scala/org/apache/linkis/hadoop/common/conf/HadoopConf.scala

@@ -81,4 +81,13 @@ object HadoopConf {
    */
   val KEYTAB_TEMP_DIR = CommonVars("linkis.keytab.temp.dir", "/tmp/keytab")
 
+  /**
+   * Enable proactive Kerberos TGT refresh when retrieving cached HDFS FileSystem. When enabled,
+   * before returning a cached FileSystem, Hadoop's `reloginFromKeytab()` is called on the UGI to
+   * refresh the TGT if near expiry. If refresh fails (TGT expired and keytab unavailable), the
+   * cached entry is removed and a new FileSystem with fresh TGT is created.
+   */
+  val HDFS_TGT_PROACTIVE_CHECK_ENABLE =
+    CommonVars("linkis.hadoop.hdfs.tgt.proactive.check.enable", false)
+
 }

linkis-commons/linkis-hadoop-common/src/main/scala/org/apache/linkis/hadoop/common/entity/HDFSFileSystemContainer.scala

@@ -18,10 +18,17 @@
 package org.apache.linkis.hadoop.common.entity
 
 import org.apache.linkis.hadoop.common.conf.HadoopConf
+import org.apache.linkis.hadoop.common.utils.KerberosTgtUtils
 
 import org.apache.hadoop.fs.FileSystem
+import org.apache.hadoop.security.UserGroupInformation
 
-class HDFSFileSystemContainer(fs: FileSystem, user: String, label: String) {
+class HDFSFileSystemContainer(
+    fs: FileSystem,
+    user: String,
+    label: String,
+    ugi: UserGroupInformation = null
+) {
 
   private var lastAccessTime: Long = System.currentTimeMillis()
 
@@ -33,6 +40,8 @@ class HDFSFileSystemContainer(fs: FileSystem, user: String, label: String) {
 
   def getLabel: String = this.label
 
+  def getUgi: UserGroupInformation = this.ugi
+
   def getLastAccessTime: Long = this.lastAccessTime
 
   def updateLastAccessTime: Unit = {
@@ -51,4 +60,14 @@ class HDFSFileSystemContainer(fs: FileSystem, user: String, label: String) {
     idleTime > HadoopConf.HDFS_ENABLE_CACHE_MAX_TIME || ((idleTime > HadoopConf.HDFS_ENABLE_CACHE_IDLE_TIME) && count <= 0)
   }
 
+  /**
+   * Check if the Kerberos TGT in this container's UGI is still valid. Delegates to
+   * KerberosTgtUtils.isTgtValid() which checks the KerberosTicket end time against current time.
+   *
+   * @return
+   *   true if TGT is valid (or UGI is null / non-Kerberos); false if TGT has expired → caller
+   *   should remove from cache and recreate FileSystem
+   */
+  def isTgtValid(): Boolean = KerberosTgtUtils.isTgtValid(ugi)
+
 }

linkis-commons/linkis-hadoop-common/src/main/scala/org/apache/linkis/hadoop/common/utils/HDFSUtils.scala

@@ -233,9 +233,23 @@ object HDFSUtils extends Logging {
       locker.intern().synchronized {
         if (fileSystemCache.containsKey(cacheKey)) {
           val hdfsFileSystemContainer = fileSystemCache.get(cacheKey)
-          hdfsFileSystemContainer.addAccessCount()
-          hdfsFileSystemContainer.updateLastAccessTime
-          hdfsFileSystemContainer.getFileSystem
+          // Proactive TGT expiry check: if expired, remove from cache and recreate
+          if (
+              HadoopConf.HDFS_TGT_PROACTIVE_CHECK_ENABLE.getValue && !hdfsFileSystemContainer
+                .isTgtValid()
+          ) {
+            logger.info(
+              s"Cached HDFS FileSystem TGT expired - user: $userName, label: $cacheLabel, removing from cache and recreating"
+            )
+            fileSystemCache.remove(cacheKey)
+            IOUtils.closeQuietly(hdfsFileSystemContainer.getFileSystem)
+            // Fall through to create new FileSystem with fresh TGT
+            getHDFSUserFileSystem(userName, label, getConfigurationByLabel(userName, label))
+          } else {
+            hdfsFileSystemContainer.addAccessCount()
+            hdfsFileSystemContainer.updateLastAccessTime
+            hdfsFileSystemContainer.getFileSystem
+          }
         } else {
           getHDFSUserFileSystem(userName, label, getConfigurationByLabel(userName, label))
         }
@@ -259,25 +273,33 @@ object HDFSUtils extends Logging {
       val cacheLabel = if (label == null) DEFAULT_CACHE_LABEL else label
       val cacheKey = userName + JOINT + cacheLabel
       locker.intern().synchronized {
-        val hdfsFileSystemContainer = if (fileSystemCache.containsKey(cacheKey)) {
-          fileSystemCache.get(cacheKey)
-        } else {
-          // we use cacheLabel to create HDFSFileSystemContainer, and in the rest part of HDFSUtils, we consistently
-          // use the same cacheLabel to operate HDFSFileSystemContainer, like close or remove.
-          // At the same time, we don't want to change the behavior of createFileSystem which is out of HDFSUtils,
-          // so we continue to use the original label to createFileSystem.
-          val newHDFSFileSystemContainer =
-            new HDFSFileSystemContainer(
-              createFileSystem(userName, label, conf),
-              userName,
-              cacheLabel
+        if (fileSystemCache.containsKey(cacheKey)) {
+          val hdfsFileSystemContainer = fileSystemCache.get(cacheKey)
+          // Proactive TGT expiry check: if expired, remove from cache and recreate
+          if (
+              HadoopConf.HDFS_TGT_PROACTIVE_CHECK_ENABLE.getValue && !hdfsFileSystemContainer
+                .isTgtValid()
+          ) {
+            logger.info(
+              s"Cached HDFS FileSystem TGT expired - user: $userName, label: $cacheLabel, removing from cache and recreating"
             )
-          fileSystemCache.put(cacheKey, newHDFSFileSystemContainer)
-          newHDFSFileSystemContainer
+            fileSystemCache.remove(cacheKey)
+            IOUtils.closeQuietly(hdfsFileSystemContainer.getFileSystem)
+            // Fall through to create new FileSystem with fresh TGT
+          } else {
+            hdfsFileSystemContainer.addAccessCount()
+            hdfsFileSystemContainer.updateLastAccessTime
+            return hdfsFileSystemContainer.getFileSystem
+          }
         }
-        hdfsFileSystemContainer.addAccessCount()
-        hdfsFileSystemContainer.updateLastAccessTime
-        hdfsFileSystemContainer.getFileSystem
+        // Cache miss or TGT expired → create new FileSystem with UGI
+        val (newFs, newUgi) = createFileSystemWithUgi(userName, label, conf)
+        val newHDFSFileSystemContainer =
+          new HDFSFileSystemContainer(newFs, userName, cacheLabel, newUgi)
+        fileSystemCache.put(cacheKey, newHDFSFileSystemContainer)
+        newHDFSFileSystemContainer.addAccessCount()
+        newHDFSFileSystemContainer.updateLastAccessTime
+        newHDFSFileSystemContainer.getFileSystem
       }
     } else {
       createFileSystem(userName, label, conf)
@@ -287,6 +309,45 @@ object HDFSUtils extends Logging {
   def createFileSystem(userName: String, conf: org.apache.hadoop.conf.Configuration): FileSystem =
     createFileSystem(userName, null, conf)
 
+  /**
+   * Create a FileSystem and also return the UserGroupInformation used to create it. The UGI is
+   * needed for proactive TGT validity checking in the cache layer.
+   */
+  private def createFileSystemWithUgi(
+      userName: String,
+      label: String,
+      conf: org.apache.hadoop.conf.Configuration
+  ): (FileSystem, UserGroupInformation) = {
+    val createCount = count.getAndIncrement()
+    val startTime = System.currentTimeMillis()
+    val labelInfo = if (label == null) "default" else label
+    logger.info(
+      s"Creating Hadoop FileSystem - user: $userName, label: $labelInfo, createCount: $createCount"
+    )
+    try {
+      val ugi = getUserGroupInformation(userName, label)
+      val fs = ugi
+        .doAs(new PrivilegedExceptionAction[FileSystem] {
+          def run: FileSystem = FileSystem.newInstance(conf)
+        })
+      val duration = System.currentTimeMillis() - startTime
+      logger.info(
+        s"Hadoop FileSystem created successfully - user: $userName, label: $labelInfo, duration: ${ByteTimeUtils
+          .msDurationToString(duration)}, createCount: $createCount"
+      )
+      (fs, ugi)
+    } catch {
+      case e: Exception =>
+        val duration = System.currentTimeMillis() - startTime
+        logger.error(
+          s"Failed to create Hadoop FileSystem - user: $userName, label: $labelInfo, duration: ${ByteTimeUtils
+            .msDurationToString(duration)}, createCount: $createCount",
+          e
+        )
+        throw e
+    }
+  }
+
   def createFileSystem(
       userName: String,
       label: String,

linkis-commons/linkis-hadoop-common/src/test/scala/org/apache/linkis/hadoop/common/conf/HadoopConfTest.scala

@@ -38,6 +38,7 @@ class HadoopConfTest {
     )
     Assertions.assertFalse(HadoopConf.HDFS_ENABLE_CACHE)
     Assertions.assertTrue(180000 == HadoopConf.HDFS_ENABLE_CACHE_IDLE_TIME)
+    Assertions.assertFalse(HadoopConf.HDFS_TGT_PROACTIVE_CHECK_ENABLE.getValue)
 
   }