[fix](filesystem) replace throttled kerberos relogin with proactive TGT refresh to fix flaky GSS failure (#64394)

### What problem does this PR solve?

Issue Number: close #xxx

Related PR: #62023

Problem Summary:

The regression case `test_two_hive_kerberos` fails intermittently with:

```
GSSException: No valid credentials provided
  ...
Caused by: javax.security.auth.login.LoginException: Cannot read from System.in
```

Root cause: the `KerberosHadoopAuthenticator` in `fe-filesystem-hdfs`
(introduced in #62023) relies on Hadoop's
`UserGroupInformation.checkTGTAndReloginFromKeytab()`, which has a
hard-coded 60s relogin throttle (`hasSufficientTimeElapsed`). When the
TGT expires inside that throttle window, the relogin is silently
skipped, the subsequent SASL handshake finds no valid ticket, and the
JAAS fallback path tries to prompt on `System.in` — which fails in a
server process.

Fix: replace the throttled reactive relogin with trino's proactive TGT
refresh model (ported 1:1 from trino 435 `KerberosAuthentication` /
`CachingKerberosHadoopAuthentication` / `KerberosTicketUtils`):

- Login eagerly at construction with explicit JAAS options
(`doNotPrompt=true`, `useKeyTab=true`, `storeKey=true`,
`isInitiator=true`) — `doNotPrompt=true` makes the `System.in` prompt
path impossible.
- Compute the next refresh time as ticket start + 80% of ticket
lifetime.
- On each `doAs`, if past the refresh point, perform a fresh keytab
login (no throttle) and swap the new credentials into the existing
`Subject` in place, so the cached UGI stays valid.
- A failed refresh does not advance the refresh time, so the next call
retries immediately; the failure is wrapped as `IOException` to keep the
SPI's checked-exception contract.

The vendored `KerberosTicketUtils` is placed in `fe-foundation`
(`org.apache.doris.foundation.security`) and provided to filesystem
plugins through `fe-filesystem-spi`, which also lets `fe-common`'s
`HadoopKerberosAuthenticator` drop its `io.trino` import.

### Release note

None

### Check List (For Author)

- Test <!-- At least one of them must be included. -->
    - [ ] Regression test
    - [x] Unit Test
    - [ ] Manual test (add detailed scripts or steps below)
    - [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
        - [ ] Previous test can cover this change.
        - [ ] No code files have been changed.
        - [ ] Other reason <!-- Add your reason?  -->

- Behavior changed:
    - [ ] No.
- [x] Yes. <!-- Explain the behavior change --> Kerberos keytab relogin
in fe-filesystem-hdfs is now proactive (refresh at 80% of TGT lifetime,
unthrottled) instead of Hadoop's reactive relogin with a 60s throttle.

fe/fe-common/pom.xml

@@ -67,6 +67,11 @@ under the License.
         </profile>
     </profiles>
     <dependencies>
+        <dependency>
+            <groupId>org.apache.doris</groupId>
+            <artifactId>fe-foundation</artifactId>
+            <version>${revision}</version>
+        </dependency>
         <dependency>
             <groupId>com.google.guava</groupId>
             <artifactId>guava</artifactId>

fe/fe-common/src/main/java/org/apache/doris/common/security/authentication/HadoopKerberosAuthenticator.java

@@ -17,10 +17,11 @@
 
 package org.apache.doris.common.security.authentication;
 
+import org.apache.doris.foundation.security.KerberosTicketUtils;
+
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
-import io.trino.plugin.base.authentication.KerberosTicketUtils;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
 import org.apache.hadoop.security.UserGroupInformation;

fe/fe-filesystem/fe-filesystem-hdfs/src/main/java/org/apache/doris/filesystem/hdfs/KerberosHadoopAuthenticator.java

@@ -14,11 +14,16 @@
 // KIND, either express or implied.  See the License for the
 // specific language governing permissions and limitations
 // under the License.
+// This file is based on code available under the Apache license here:
+// https://github.com/trinodb/trino/blob/435/lib/trino-plugin-toolkit/src/main/java/io/trino/plugin/base/authentication/KerberosAuthentication.java
+// https://github.com/trinodb/trino/blob/435/lib/trino-hdfs/src/main/java/io/trino/hdfs/authentication/CachingKerberosHadoopAuthentication.java
+// and modified by Doris
 
 package org.apache.doris.filesystem.hdfs;
 
 import org.apache.doris.filesystem.spi.HadoopAuthenticator;
 import org.apache.doris.filesystem.spi.IOCallable;
+import org.apache.doris.foundation.security.KerberosTicketUtils;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.security.SecurityUtil;
@@ -29,10 +34,29 @@ import org.apache.logging.log4j.Logger;
 
 import java.io.IOException;
 import java.security.PrivilegedExceptionAction;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import javax.security.auth.Subject;
+import javax.security.auth.kerberos.KerberosPrincipal;
+import javax.security.auth.login.AppConfigurationEntry;
+import javax.security.auth.login.LoginContext;
+import javax.security.auth.login.LoginException;
 
 /**
  * Kerberos-based implementation of {@link HadoopAuthenticator}.
- * Logs in from a keytab and executes actions as the Kerberos principal via UGI.doAs().
+ *
+ * <p>Logs in from a keytab via an explicit JAAS {@code Krb5LoginModule} configuration
+ * ({@code doNotPrompt=true}) and proactively refreshes the TGT once it passes 80% of
+ * its lifetime, swapping the new credentials into the existing Subject in place. This
+ * is a port of trino's {@code KerberosAuthentication} +
+ * {@code CachingKerberosHadoopAuthentication}. It deliberately does NOT use
+ * {@code UserGroupInformation.checkTGTAndReloginFromKeytab()}: its hard-coded
+ * 60-second relogin throttle can leave an expired TGT in the Subject, after which the
+ * SASL/GSS layer falls back to the JVM-default interactive JAAS login and fails with
+ * "LoginException: Cannot read from System.in".
  *
  * <p>Note: {@link UserGroupInformation#setConfiguration(Configuration)} mutates
  * JVM-global state — all UGI instances in the process share the same authentication
@@ -51,7 +75,14 @@ public class KerberosHadoopAuthenticator implements HadoopAuthenticator {
 
     private final String principal;
     private final String keytab;
-    private volatile UserGroupInformation ugi;
+
+    // The Subject/UGI pair is created once and never replaced: refreshes swap new
+    // Kerberos credentials into this same Subject so Hadoop code that caches the
+    // UGI (e.g. DFSClient) transparently sees the new ticket.
+    private final Subject subject;
+    private final UserGroupInformation ugi;
+    // Guarded by "this" (only touched in the constructor and synchronized getUGI()).
+    private long nextRefreshTime;
 
     public KerberosHadoopAuthenticator(String principal, String keytab, Configuration conf) {
         this.principal = principal;
@@ -62,10 +93,14 @@ public class KerberosHadoopAuthenticator implements HadoopAuthenticator {
                 if (!shouldSkipSetConfiguration(desired)) {
                     UserGroupInformation.setConfiguration(conf);
                 }
-                this.ugi = UserGroupInformation.loginUserFromKeytabAndReturnUGI(principal, keytab);
+                this.subject = loginSubject();
+                this.ugi = Objects.requireNonNull(UserGroupInformation.getUGIFromSubject(subject),
+                        "getUGIFromSubject returned null");
             }
+            this.nextRefreshTime = KerberosTicketUtils.getRefreshTime(
+                    KerberosTicketUtils.getTicketGrantingTicket(subject));
             LOG.info("Kerberos login succeeded for principal={}", principal);
-        } catch (IOException e) {
+        } catch (IOException | RuntimeException e) {
             throw new RuntimeException("Failed to login with Kerberos principal=" + principal
                     + ", keytab=" + keytab, e);
         }
@@ -98,20 +133,99 @@ public class KerberosHadoopAuthenticator implements HadoopAuthenticator {
 
     @Override
     public <T> T doAs(IOCallable<T> action) throws IOException {
-        // Refresh the Kerberos TGT from the keytab if it's close to expiry. This is
-        // a no-op when the ticket is still valid, so it's safe and cheap to call on
-        // every request and avoids long-lived FE processes failing with
-        // "GSSException: No valid credentials" after the initial ticket expires.
+        UserGroupInformation currentUgi;
         try {
-            ugi.checkTGTAndReloginFromKeytab();
-        } catch (IOException e) {
+            currentUgi = getUGI();
+        } catch (IOException | RuntimeException e) {
+            // Keep the SPI's checked-IOException contract: a relogin failure (unchecked
+            // RuntimeException from the JAAS login) must not escape doAs unchecked.
             throw new IOException("Kerberos relogin failed for principal=" + principal, e);
         }
         try {
-            return ugi.doAs((PrivilegedExceptionAction<T>) action::call);
+            return currentUgi.doAs((PrivilegedExceptionAction<T>) action::call);
         } catch (InterruptedException e) {
             Thread.currentThread().interrupt();
             throw new IOException("Kerberos doAs interrupted for principal=" + principal, e);
         }
     }
+
+    /**
+     * Returns the cached UGI, first refreshing the TGT if it is past 80% of its
+     * lifetime. Ported from trino's
+     * {@code CachingKerberosHadoopAuthentication.getUserGroupInformation()} — note
+     * there is intentionally no relogin throttle.
+     * If the refresh login fails, {@code nextRefreshTime} is not advanced, so each
+     * subsequent call retries the login until the KDC recovers (no backoff) — same
+     * trade-off as trino.
+     */
+    private synchronized UserGroupInformation getUGI() throws IOException {
+        if (nextRefreshTime < System.currentTimeMillis()) {
+            Subject newSubject = loginSubject();
+            Objects.requireNonNull(UserGroupInformation.getUGIFromSubject(newSubject),
+                    "getUGIFromSubject returned null");
+            // We modify the existing UGI's credentials in-place instead of returning a new UGI
+            // because some parts of Hadoop code reuse UGI (e.g. DFSClient).
+            // We also need to clear the old credentials because the JDK assumes that the first
+            // credential is the TGT, which is not always true.
+            subject.getPrincipals().addAll(newSubject.getPrincipals());
+            Set<Object> privateCredentials = subject.getPrivateCredentials();
+            synchronized (privateCredentials) {
+                privateCredentials.clear();
+                privateCredentials.addAll(newSubject.getPrivateCredentials());
+            }
+            Set<Object> publicCredentials = subject.getPublicCredentials();
+            synchronized (publicCredentials) {
+                publicCredentials.clear();
+                publicCredentials.addAll(newSubject.getPublicCredentials());
+            }
+            nextRefreshTime = KerberosTicketUtils.getRefreshTime(
+                    KerberosTicketUtils.getTicketGrantingTicket(newSubject));
+            LOG.info("Kerberos ticket refreshed for principal={}, next refresh time={}",
+                    principal, nextRefreshTime);
+        }
+        return ugi;
+    }
+
+    /**
+     * Performs the JAAS keytab login and returns the logged-in Subject. This is
+     * trino's {@code KerberosAuthentication.getSubject()} delegate boundary, kept
+     * as a package-private method so tests can substitute fabricated Subjects.
+     */
+    Subject loginSubject() {
+        return getSubject(keytab, principal);
+    }
+
+    private static Subject getSubject(String keytab, String principal) {
+        Subject subject = new Subject(false, Collections.singleton(new KerberosPrincipal(principal)),
+                Collections.emptySet(), Collections.emptySet());
+        javax.security.auth.login.Configuration conf = getConfiguration(keytab, principal);
+        try {
+            LoginContext loginContext = new LoginContext("", subject, null, conf);
+            loginContext.login();
+            return loginContext.getSubject();
+        } catch (LoginException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    private static javax.security.auth.login.Configuration getConfiguration(String keytab, String principal) {
+        Map<String, String> optionsBuilder = new LinkedHashMap<>();
+        optionsBuilder.put("doNotPrompt", "true");
+        optionsBuilder.put("isInitiator", "true");
+        optionsBuilder.put("principal", principal);
+        optionsBuilder.put("useKeyTab", "true");
+        optionsBuilder.put("storeKey", "true");
+        optionsBuilder.put("keyTab", keytab);
+        Map<String, String> options = Collections.unmodifiableMap(optionsBuilder);
+        return new javax.security.auth.login.Configuration() {
+            @Override
+            public AppConfigurationEntry[] getAppConfigurationEntry(String name) {
+                return new AppConfigurationEntry[] {
+                        new AppConfigurationEntry(
+                                "com.sun.security.auth.module.Krb5LoginModule",
+                                AppConfigurationEntry.LoginModuleControlFlag.REQUIRED,
+                                options)};
+            }
+        };
+    }
 }

fe/fe-filesystem/fe-filesystem-hdfs/src/test/java/org/apache/doris/filesystem/hdfs/KerberosHadoopAuthenticatorTest.java

@@ -0,0 +1,165 @@
+// 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.doris.filesystem.hdfs;
+
+import org.apache.doris.foundation.security.KerberosTicketUtils;
+
+import org.apache.hadoop.conf.Configuration;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.ArrayDeque;
+import java.util.Date;
+import java.util.Deque;
+import javax.security.auth.Subject;
+import javax.security.auth.kerberos.KerberosPrincipal;
+import javax.security.auth.kerberos.KerberosTicket;
+
+/**
+ * Unit tests for the proactive TGT refresh logic. No KDC: the JAAS keytab login
+ * (package-private {@code loginSubject()} seam, trino's KerberosAuthentication
+ * boundary) is replaced with fabricated Subjects holding real KerberosTicket
+ * objects whose start/end times steer the 80%-lifetime refresh point.
+ */
+class KerberosHadoopAuthenticatorTest {
+
+    private static final KerberosPrincipal CLIENT = new KerberosPrincipal("doris/test@EXAMPLE.COM");
+    private static final KerberosPrincipal TGS = new KerberosPrincipal("krbtgt/EXAMPLE.COM@EXAMPLE.COM");
+
+    /** Canned-login subclass; static state because loginSubject() is called from the super constructor. */
+    private static final class FakeLoginAuthenticator extends KerberosHadoopAuthenticator {
+        static final Deque<Subject> LOGINS = new ArrayDeque<>();
+        static int loginCount = 0;
+        static RuntimeException nextLoginFailure = null;
+
+        FakeLoginAuthenticator() {
+            super("doris/test@EXAMPLE.COM", "/path/to/doris.keytab", new Configuration());
+        }
+
+        @Override
+        Subject loginSubject() {
+            loginCount++;
+            if (nextLoginFailure != null) {
+                RuntimeException failure = nextLoginFailure;
+                nextLoginFailure = null;
+                throw failure;
+            }
+            return LOGINS.removeFirst();
+        }
+    }
+
+    private static Subject subjectWithTgt(long startMillis, long endMillis) {
+        Subject subject = new Subject();
+        subject.getPrincipals().add(CLIENT);
+        subject.getPrivateCredentials().add(new KerberosTicket(new byte[] {1}, CLIENT, TGS,
+                new byte[] {1}, 1, null, new Date(startMillis), new Date(startMillis),
+                new Date(endMillis), null, null));
+        return subject;
+    }
+
+    @BeforeEach
+    void resetFakeLogins() {
+        FakeLoginAuthenticator.LOGINS.clear();
+        FakeLoginAuthenticator.loginCount = 0;
+        FakeLoginAuthenticator.nextLoginFailure = null;
+    }
+
+    @Test
+    void constructorLogsInEagerlyAndFreshTicketIsNotRefreshed() throws IOException {
+        long now = System.currentTimeMillis();
+        // refresh point = now + 0.8 * 600s = now + 480s, far in the future
+        FakeLoginAuthenticator.LOGINS.add(subjectWithTgt(now, now + 600_000));
+
+        KerberosHadoopAuthenticator auth = new FakeLoginAuthenticator();
+        Assertions.assertEquals(1, FakeLoginAuthenticator.loginCount);
+
+        Assertions.assertEquals("ok", auth.doAs(() -> "ok"));
+        auth.doAs(() -> "ok");
+        Assertions.assertEquals(1, FakeLoginAuthenticator.loginCount);
+    }
+
+    @Test
+    void staleTicketIsRefreshedInPlaceWithoutThrottle() throws IOException {
+        long now = System.currentTimeMillis();
+        // initial TGT is past its 80%-lifetime refresh point:
+        // refresh point = (now-100s) + 0.8 * 101s = now - 19.2s < now
+        Subject initial = subjectWithTgt(now - 100_000, now + 1_000);
+        Subject renewed = subjectWithTgt(now, now + 600_000);
+        FakeLoginAuthenticator.LOGINS.add(initial);
+        FakeLoginAuthenticator.LOGINS.add(renewed);
+
+        KerberosHadoopAuthenticator auth = new FakeLoginAuthenticator();
+        Assertions.assertEquals(1, FakeLoginAuthenticator.loginCount);
+
+        // constructor login happened seconds ago; a 60s-throttled implementation
+        // (checkTGTAndReloginFromKeytab) would skip this refresh — ours must not
+        Assertions.assertEquals("ok", auth.doAs(() -> "ok"));
+        Assertions.assertEquals(2, FakeLoginAuthenticator.loginCount);
+
+        // in-place swap: the ORIGINAL Subject now holds the renewed TGT
+        KerberosTicket current = KerberosTicketUtils.getTicketGrantingTicket(initial);
+        Assertions.assertEquals(now + 600_000, current.getEndTime().getTime());
+
+        // renewed ticket is fresh → no further login
+        auth.doAs(() -> "ok");
+        Assertions.assertEquals(2, FakeLoginAuthenticator.loginCount);
+    }
+
+    @Test
+    void doAsPropagatesIOException() {
+        long now = System.currentTimeMillis();
+        FakeLoginAuthenticator.LOGINS.add(subjectWithTgt(now, now + 600_000));
+
+        KerberosHadoopAuthenticator auth = new FakeLoginAuthenticator();
+        IOException thrown = Assertions.assertThrows(IOException.class, () -> auth.doAs(() -> {
+            throw new IOException("intentional");
+        }));
+        Assertions.assertEquals("intentional", thrown.getMessage());
+    }
+
+    @Test
+    void constructorWrapsLoginFailureWithPrincipalAndKeytab() {
+        FakeLoginAuthenticator.nextLoginFailure =
+                new RuntimeException(new javax.security.auth.login.LoginException("no keytab"));
+        RuntimeException thrown = Assertions.assertThrows(RuntimeException.class,
+                FakeLoginAuthenticator::new);
+        Assertions.assertTrue(thrown.getMessage().contains("doris/test@EXAMPLE.COM"));
+        Assertions.assertTrue(thrown.getMessage().contains("/path/to/doris.keytab"));
+    }
+
+    @Test
+    void doAsWrapsRefreshLoginFailureAsIOException() throws IOException {
+        long now = System.currentTimeMillis();
+        // TGT already past its 80% refresh point, so the first doAs attempts a relogin
+        FakeLoginAuthenticator.LOGINS.add(subjectWithTgt(now - 100_000, now + 1_000));
+
+        KerberosHadoopAuthenticator auth = new FakeLoginAuthenticator();
+        FakeLoginAuthenticator.nextLoginFailure =
+                new RuntimeException(new javax.security.auth.login.LoginException("kdc down"));
+
+        IOException thrown = Assertions.assertThrows(IOException.class, () -> auth.doAs(() -> "ok"));
+        Assertions.assertTrue(thrown.getMessage().contains("Kerberos relogin failed"));
+        Assertions.assertTrue(thrown.getMessage().contains("doris/test@EXAMPLE.COM"));
+
+        // a later doAs retries the login (nextRefreshTime was not advanced) and succeeds
+        FakeLoginAuthenticator.LOGINS.add(subjectWithTgt(now, now + 600_000));
+        Assertions.assertEquals("ok", auth.doAs(() -> "ok"));
+    }
+}

fe/fe-filesystem/fe-filesystem-spi/pom.xml

@@ -36,8 +36,8 @@ under the License.
         Service Provider Interface (SPI) for Doris FE filesystem abstraction.
         Contains FileSystemProvider (ServiceLoader SPI entry point), the ObjStorage SPI layer
         for object-storage backends, HadoopAuthenticator, and their supporting value types.
-        Zero third-party external dependencies — only JDK and Doris internal SPI interfaces.
-        This is the ONLY filesystem artifact compiled into fe-core.
+        Depends only on JDK and Doris internal modules (fe-filesystem-api, fe-extension-spi,
+        fe-foundation). This is the ONLY filesystem artifact compiled into fe-core.
     </description>
 
     <dependencies>
@@ -56,6 +56,14 @@ under the License.
             <artifactId>fe-extension-spi</artifactId>
             <version>${project.version}</version>
         </dependency>
+        <!-- fe-foundation: bottom-layer shared utilities (e.g. KerberosTicketUtils) provided
+             to all filesystem provider plugins through the SPI platform, so individual
+             plugins do not declare it themselves. -->
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>fe-foundation</artifactId>
+            <version>${project.version}</version>
+        </dependency>
         <dependency>
             <groupId>org.junit.jupiter</groupId>
             <artifactId>junit-jupiter</artifactId>

fe/fe-foundation/src/main/java/org/apache/doris/foundation/security/KerberosTicketUtils.java

@@ -0,0 +1,75 @@
+// 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.
+// This file is copied from
+// https://github.com/trinodb/trino/blob/435/lib/trino-plugin-toolkit/src/main/java/io/trino/plugin/base/authentication/KerberosTicketUtils.java
+// and modified by Doris
+
+package org.apache.doris.foundation.security;
+
+import java.util.Set;
+import javax.security.auth.Subject;
+import javax.security.auth.kerberos.KerberosPrincipal;
+import javax.security.auth.kerberos.KerberosTicket;
+
+public final class KerberosTicketUtils {
+    private static final float TICKET_RENEW_WINDOW = 0.80f;
+
+    private KerberosTicketUtils() {
+    }
+
+    public static KerberosTicket getTicketGrantingTicket(Subject subject) {
+        Set<KerberosTicket> tickets = subject.getPrivateCredentials(KerberosTicket.class);
+        for (KerberosTicket ticket : tickets) {
+            if (isOriginalTicketGrantingTicket(ticket)) {
+                return ticket;
+            }
+        }
+        throw new IllegalArgumentException("kerberos ticket not found in " + subject);
+    }
+
+    public static long getRefreshTime(KerberosTicket ticket) {
+        long start = ticket.getStartTime().getTime();
+        long end = ticket.getEndTime().getTime();
+        return start + (long) ((end - start) * TICKET_RENEW_WINDOW);
+    }
+
+    /**
+     * Check whether the server principal is the TGS's principal
+     *
+     * @param ticket the original TGT (the ticket that is obtained when a
+     * kinit is done)
+     * @return true or false
+     */
+    public static boolean isOriginalTicketGrantingTicket(KerberosTicket ticket) {
+        return isTicketGrantingServerPrincipal(ticket.getServer());
+    }
+
+    /**
+     * TGS must have the server principal of the form "krbtgt/FOO@FOO".
+     *
+     * @return true or false
+     */
+    private static boolean isTicketGrantingServerPrincipal(KerberosPrincipal principal) {
+        if (principal == null) {
+            return false;
+        }
+        if (principal.getName().equals("krbtgt/" + principal.getRealm() + "@" + principal.getRealm())) {
+            return true;
+        }
+        return false;
+    }
+}

fe/fe-foundation/src/test/java/org/apache/doris/foundation/security/KerberosTicketUtilsTest.java

@@ -0,0 +1,72 @@
+// 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.doris.foundation.security;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.Date;
+import javax.security.auth.Subject;
+import javax.security.auth.kerberos.KerberosPrincipal;
+import javax.security.auth.kerberos.KerberosTicket;
+
+class KerberosTicketUtilsTest {
+
+    private static final KerberosPrincipal CLIENT = new KerberosPrincipal("doris/test@EXAMPLE.COM");
+    private static final KerberosPrincipal TGS = new KerberosPrincipal("krbtgt/EXAMPLE.COM@EXAMPLE.COM");
+    private static final KerberosPrincipal SERVICE = new KerberosPrincipal("hive/host@EXAMPLE.COM");
+
+    static KerberosTicket ticket(KerberosPrincipal server, long startMillis, long endMillis) {
+        return new KerberosTicket(new byte[] {1}, CLIENT, server, new byte[] {1}, 1, null,
+                new Date(startMillis), new Date(startMillis), new Date(endMillis), null, null);
+    }
+
+    @Test
+    void getRefreshTimeIsAt80PercentOfTicketLifetime() {
+        long start = 1_000_000L;
+        long end = start + 100_000L;
+        Assertions.assertEquals(start + 80_000L,
+                KerberosTicketUtils.getRefreshTime(ticket(TGS, start, end)));
+    }
+
+    @Test
+    void getTicketGrantingTicketSelectsTgtAmongOtherTickets() {
+        KerberosTicket serviceTicket = ticket(SERVICE, 0, 100_000);
+        KerberosTicket tgt = ticket(TGS, 0, 100_000);
+        Subject subject = new Subject();
+        subject.getPrivateCredentials().add(serviceTicket);
+        subject.getPrivateCredentials().add(tgt);
+        Assertions.assertSame(tgt, KerberosTicketUtils.getTicketGrantingTicket(subject));
+    }
+
+    @Test
+    void getTicketGrantingTicketThrowsWhenAbsent() {
+        Subject subject = new Subject();
+        subject.getPrivateCredentials().add(ticket(SERVICE, 0, 100_000));
+        Assertions.assertThrows(IllegalArgumentException.class,
+                () -> KerberosTicketUtils.getTicketGrantingTicket(subject));
+    }
+
+    @Test
+    void isOriginalTicketGrantingTicketChecksServerPrincipalForm() {
+        Assertions.assertTrue(
+                KerberosTicketUtils.isOriginalTicketGrantingTicket(ticket(TGS, 0, 100_000)));
+        Assertions.assertFalse(
+                KerberosTicketUtils.isOriginalTicketGrantingTicket(ticket(SERVICE, 0, 100_000)));
+    }
+}

fs_brokers/cdc_client/build.sh

@@ -27,7 +27,7 @@ export CDC_CLIENT_HOME="${ROOT}"
 
 bash "${DORIS_HOME}"/generated-source.sh noclean
 cd "${DORIS_HOME}/fe"
-"${MVN_CMD}" -Pflatten install -pl fe-common -Dskip.doc=true -DskipTests -Dmaven.build.cache.enabled=false
+"${MVN_CMD}" -Pflatten install -pl fe-common -am -Dskip.doc=true -DskipTests -Dmaven.build.cache.enabled=false
 
 echo "Install cdc client..."
 cd "${CDC_CLIENT_HOME}"