[Fix-17036] Fix kerberos will not refresh in HdfsStorageOperator (#17039)

Author: ruanwenjun

dolphinscheduler-storage-plugin/dolphinscheduler-storage-hdfs/src/main/java/org/apache/dolphinscheduler/plugin/storage/hdfs/HdfsStorageOperator.java

@@ -17,7 +17,10 @@
 
 package org.apache.dolphinscheduler.plugin.storage.hdfs;
 
+import org.apache.dolphinscheduler.common.thread.ThreadUtils;
 import org.apache.dolphinscheduler.common.utils.FileUtils;
+import org.apache.dolphinscheduler.common.utils.PropertyUtils;
+import org.apache.dolphinscheduler.plugin.datasource.api.constants.DataSourceConstants;
 import org.apache.dolphinscheduler.plugin.datasource.api.utils.CommonUtils;
 import org.apache.dolphinscheduler.plugin.storage.api.AbstractStorageOperator;
 import org.apache.dolphinscheduler.plugin.storage.api.ResourceMetadata;
@@ -51,6 +54,7 @@
 import java.util.Date;
 import java.util.LinkedList;
 import java.util.List;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
 import lombok.SneakyThrows;
@@ -102,6 +106,18 @@ private void init() {
 
         if (CommonUtils.getKerberosStartupState()) {
             CommonUtils.loadKerberosConf(configuration);
+            final Long kerberosExpireTimeInHour = PropertyUtils.getLong(DataSourceConstants.KERBEROS_EXPIRE_TIME, -1L);
+            if (kerberosExpireTimeInHour > 0) {
+                ThreadUtils.newDaemonScheduledExecutorService("ds-hdfs-kerberos-refresh-%s", 1)
+                        .scheduleWithFixedDelay(() -> {
+                            try {
+                                UserGroupInformation.getLoginUser().checkTGTAndReloginFromKeytab();
+                                log.info("checkTGTAndReloginFromKeytab finished");
+                            } catch (Exception e) {
+                                log.error("checkTGTAndReloginFromKeytab Error", e);
+                            }
+                        }, kerberosExpireTimeInHour, kerberosExpireTimeInHour, TimeUnit.MINUTES);
+            }
             fs = FileSystem.get(configuration);
             log.info("Initialize HdfsStorageOperator with kerberos");
             return;