[#9564] feat(flink): Flink connector supports user authentication (#9565)

### What changes were proposed in this pull request?

Flink connector supports user authentication.

### Why are the changes needed?

Fix: #9564 

### Does this PR introduce _any_ user-facing change?

Add the documents.

### How was this patch tested?

Add integration tests.

---------

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Author: roryqi

clients/client-java/src/main/java/org/apache/gravitino/client/KerberosTokenProvider.java

@@ -24,10 +24,14 @@
 import java.io.File;
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
+import java.time.Instant;
 import java.util.Base64;
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.Callable;
+import javax.security.auth.Subject;
+import javax.security.auth.kerberos.KerberosKey;
+import javax.security.auth.kerberos.KerberosPrincipal;
 import javax.security.auth.kerberos.KerberosTicket;
 import javax.security.auth.login.LoginContext;
 import javax.security.auth.login.LoginException;
@@ -49,7 +53,7 @@ public final class KerberosTokenProvider implements AuthDataProvider {
   private String clientPrincipal;
   private String keytabFile;
   private String host = "localhost";
-  private LoginContext loginContext;
+  private SubjectProvider subjectProvider;
 
   private KerberosTokenProvider() {}
 
@@ -82,22 +86,11 @@ public byte[] getTokenData() {
   private byte[] getTokenInternal() throws Exception {
     @SuppressWarnings("null")
     List<String> principalComponents = Splitter.on('@').splitToList(clientPrincipal);
-    // Gravitino server's principal must start with HTTP. This restriction follows
-    // the style of Apache Hadoop.
     String serverPrincipal = "HTTP/" + host + "@" + principalComponents.get(1);
-
-    synchronized (this) {
-      if (loginContext == null) {
-        loginContext = KerberosUtils.login(clientPrincipal, keytabFile);
-      } else if (isLoginTicketExpired() && keytabFile != null) {
-        // We only support use keytab to re-login context
-        loginContext.logout();
-        loginContext = KerberosUtils.login(clientPrincipal, keytabFile);
-      }
-    }
+    Subject currentSubject = subjectProvider.get();
 
     return KerberosUtils.doAs(
-        loginContext.getSubject(),
+        currentSubject,
         new Callable<byte[]>() {
           @Override
           public byte[] call() throws Exception {
@@ -127,24 +120,11 @@ public byte[] call() throws Exception {
         });
   }
 
-  @SuppressWarnings("JavaUtilDate")
-  private boolean isLoginTicketExpired() {
-    Set<KerberosTicket> tickets =
-        loginContext.getSubject().getPrivateCredentials(KerberosTicket.class);
-
-    if (tickets.isEmpty()) {
-      return false;
-    }
-
-    return tickets.iterator().next().getEndTime().getTime() < System.currentTimeMillis();
-  }
-
-  /** Closes the KerberosTokenProvider and releases any underlying resources. */
   @Override
   public void close() throws IOException {
     try {
-      if (loginContext != null) {
-        loginContext.logout();
+      if (subjectProvider != null) {
+        subjectProvider.close();
       }
     } catch (LoginException le) {
       throw new IOException("Fail to close login context", le);
@@ -155,6 +135,113 @@ void setHost(String host) {
     this.host = host;
   }
 
+  /**
+   * Strategy interface for providing Kerberos Subject credentials.
+   *
+   * <p>There are two strategies:
+   *
+   * <ul>
+   *   <li>{@link ExistingSubjectProvider} - Reuses credentials maintained by the framework (e.g.,
+   *       Flink)
+   *   <li>{@link LoginSubjectProvider} - Creates and manages new credentials using keytab and
+   *       principal
+   * </ul>
+   */
+  private interface SubjectProvider {
+    Subject get() throws LoginException;
+
+    void close() throws LoginException;
+  }
+
+  /**
+   * SubjectProvider that reuses existing Kerberos credentials from the framework.
+   *
+   * <p>When Flink (or another framework) has already logged in using keytab and principal, this
+   * provider reuses those credentials instead of creating new ones. This approach:
+   *
+   * <ul>
+   *   <li>Avoids redundant Kerberos logins
+   *   <li>Lets the framework manage credential lifecycle (renewal, expiration)
+   *   <li>Reduces complexity - the client doesn't need to maintain credentials
+   * </ul>
+   *
+   * <p>The Subject is obtained from the current AccessControlContext and contains KerberosKey or
+   * KerberosTicket credentials managed by the framework.
+   */
+  private static final class ExistingSubjectProvider implements SubjectProvider {
+    private final Subject subject;
+
+    ExistingSubjectProvider(Subject subject) {
+      this.subject = subject;
+    }
+
+    @Override
+    public Subject get() {
+      return subject;
+    }
+
+    @Override
+    public void close() {
+      // no-op: The framework owns the Subject and is responsible for its lifecycle
+    }
+  }
+
+  /**
+   * SubjectProvider that performs Kerberos login using keytab and principal.
+   *
+   * <p>This provider is used when no existing Kerberos credentials are available from the
+   * framework. It:
+   *
+   * <ul>
+   *   <li>Performs initial login using the provided keytab file and principal
+   *   <li>Manages the LoginContext lifecycle
+   *   <li>Automatically re-authenticates when the TGT (Ticket Granting Ticket) expires
+   * </ul>
+   *
+   * <p>This provider is responsible for credential lifecycle management including renewal and
+   * cleanup.
+   */
+  private static final class LoginSubjectProvider implements SubjectProvider {
+    private final String principal;
+    private final String keytabFile;
+    private LoginContext loginContext;
+
+    LoginSubjectProvider(String principal, String keytabFile) {
+      this.principal = principal;
+      this.keytabFile = keytabFile;
+    }
+
+    @Override
+    public synchronized Subject get() throws LoginException {
+      // Perform initial login if not already logged in
+      if (loginContext == null) {
+        loginContext = KerberosUtils.login(principal, keytabFile);
+      } else if (keytabFile != null && isLoginTicketExpired(loginContext)) {
+        // If the TGT (Ticket Granting Ticket) has expired, logout and re-authenticate
+        // This ensures we always have valid credentials for GSS context negotiation
+        loginContext.logout();
+        loginContext = KerberosUtils.login(principal, keytabFile);
+      }
+      return loginContext.getSubject();
+    }
+
+    @Override
+    public void close() throws LoginException {
+      if (loginContext != null) {
+        loginContext.logout();
+      }
+    }
+
+    private boolean isLoginTicketExpired(LoginContext ctx) {
+      Set<KerberosTicket> tickets = ctx.getSubject().getPrivateCredentials(KerberosTicket.class);
+      if (tickets.isEmpty()) {
+        return false;
+      }
+      // For one principal, there should be only one TGT ticket
+      return tickets.iterator().next().getEndTime().toInstant().isBefore(Instant.now());
+    }
+  }
+
   /**
    * Creates a new instance of the KerberosTokenProvider.Builder
    *
@@ -194,19 +281,41 @@ public Builder withKeyTabFile(File file) {
     /**
      * Builds the instance of the KerberosTokenProvider.
      *
+     * <p>This method determines the authentication strategy:
+     *
+     * <ul>
+     *   <li>If Kerberos credentials already exist in the current context (e.g., Flink has logged
+     *       in), use {@link ExistingSubjectProvider} to reuse those credentials
+     *   <li>Otherwise, use {@link LoginSubjectProvider} to perform a new Kerberos login using the
+     *       provided keytab and principal
+     * </ul>
+     *
      * @return The built KerberosTokenProvider instance.
      */
-    @SuppressWarnings("null")
+    @SuppressWarnings("removal")
     public KerberosTokenProvider build() {
       KerberosTokenProvider provider = new KerberosTokenProvider();
 
-      Preconditions.checkArgument(
-          StringUtils.isNotBlank(clientPrincipal),
-          "KerberosTokenProvider must set clientPrincipal");
-      Preconditions.checkArgument(
-          Splitter.on('@').splitToList(clientPrincipal).size() == 2,
-          "Principal has the wrong format");
-      provider.clientPrincipal = clientPrincipal;
+      // Check if the framework (e.g., Flink) has already established Kerberos credentials
+      java.security.AccessControlContext context = java.security.AccessController.getContext();
+      Subject subject = Subject.getSubject(context);
+
+      // If credentials exist (KerberosKey or KerberosTicket), reuse them
+      // This avoids redundant logins when Flink has already authenticated with Kerberos
+      if (subject != null
+          && (!subject.getPrivateCredentials(KerberosKey.class).isEmpty()
+              || !subject.getPrivateCredentials(KerberosTicket.class).isEmpty())) {
+        // Use ExistingSubjectProvider: framework manages the credentials
+        provider.subjectProvider = new ExistingSubjectProvider(subject);
+
+        extractPrincipalFromSubject(subject);
+        setProviderClientPrincipal(provider);
+
+        return provider;
+      }
+
+      // No existing credentials found - we need to login ourselves
+      setProviderClientPrincipal(provider);
 
       if (keyTabFile != null) {
         Preconditions.checkArgument(
@@ -216,7 +325,29 @@ public KerberosTokenProvider build() {
         provider.keytabFile = keyTabFile.getAbsolutePath();
       }
 
+      // Use LoginSubjectProvider: client manages the credentials
+      // This provider will login using keytab/principal and handle ticket renewal
+      provider.subjectProvider =
+          new LoginSubjectProvider(provider.clientPrincipal, provider.keytabFile);
       return provider;
     }
+
+    private void setProviderClientPrincipal(KerberosTokenProvider provider) {
+      Preconditions.checkArgument(
+          StringUtils.isNotBlank(clientPrincipal),
+          "KerberosTokenProvider must set clientPrincipal");
+      Preconditions.checkArgument(
+          Splitter.on('@').splitToList(clientPrincipal).size() == 2,
+          "Principal has the wrong format");
+      provider.clientPrincipal = clientPrincipal;
+    }
+
+    private void extractPrincipalFromSubject(Subject subject) {
+      clientPrincipal =
+          subject.getPrincipals(KerberosPrincipal.class).stream()
+              .findFirst()
+              .map(Object::toString)
+              .orElse(null);
+    }
   }
 }

docs/flink-connector/flink-authentication-with-gravitino.md

@@ -0,0 +1,50 @@
+---
+title: "Flink authentication with Gravitino server"
+slug: /flink-connector/flink-authentication
+keyword: flink connector authentication oauth2 kerberos
+license: "This software is licensed under the Apache License version 2."
+---
+
+## Overview
+
+Flink connector supports `simple`, `oauth2`, and `kerberos` authentication when accessing the Gravitino server.
+
+| Property                                                  | Type   | Default Value | Description                                                                                                                                | Required | Since Version |
+|-----------------------------------------------------------|--------|---------------|--------------------------------------------------------------------------------------------------------------------------------------------|----------|---------------|
+| table.catalog-store.gravitino.gravitino.client.auth.type  | string | (none)        | When explicitly set, only `oauth` is supported. If unset, Flink selects Kerberos or simple authentication based on its security settings.  | No       | 1.2.0         |
+
+## Simple mode
+
+In simple mode, the username originates from Flink. The resolution order is:
+1. `HADOOP_USER_NAME` environment variable
+2. The logged-in OS user
+
+## OAuth2 mode
+
+In OAuth2 mode, configure the following settings to fetch an OAuth2 token to access the Gravitino server:
+
+| Property                                                              | Type   | Default Value | Description                                      | Required                       | Since Version |
+|-----------------------------------------------------------------------|--------|---------------|--------------------------------------------------|--------------------------------|---------------|
+| table.catalog-store.gravitino.gravitino.client.oauth2.serverUri       | string | (none)        | The OAuth2 server URI.                           | Yes, for OAuth2 mode           | 1.2.0         |
+| table.catalog-store.gravitino.gravitino.client.oauth2.tokenPath       | string | (none)        | The token endpoint path on the OAuth2 server.    | Yes, for OAuth2 mode           | 1.2.0         |
+| table.catalog-store.gravitino.gravitino.client.oauth2.credential      | string | (none)        | The credential used to request the OAuth2 token. | Yes, for OAuth2 mode           | 1.2.0         |
+| table.catalog-store.gravitino.gravitino.client.oauth2.scope           | string | (none)        | The scope used to request the OAuth2 token.      | Yes, for OAuth2 mode           | 1.2.0         |
+
+### OAuth2 Configuration Example
+
+```yaml
+table.catalog-store.kind: gravitino
+table.catalog-store.gravitino.gravitino.uri: http://localhost:8090
+table.catalog-store.gravitino.gravitino.metalake: my_metalake
+table.catalog-store.gravitino.gravitino.client.auth.type: oauth2
+table.catalog-store.gravitino.gravitino.client.oauth2.serverUri: https://oauth-server.example.com
+table.catalog-store.gravitino.gravitino.client.oauth2.tokenPath: /oauth/token
+table.catalog-store.gravitino.gravitino.client.oauth2.credential: your-client-credentials
+table.catalog-store.gravitino.gravitino.client.oauth2.scope: your-scope
+```
+
+## Kerberos mode
+
+In Kerberos mode, use Flink security configurations to obtain a Kerberos ticket for accessing the Gravitino server. Configure `security.kerberos.login.principal` and `security.kerberos.login.keytab` for the Kerberos principal and keytab.
+
+The Gravitino server principal follows the pattern `HTTP/$host@$realm`; ensure `$host` matches the host specified in the Gravitino server URI. Ensure `krb5.conf` is available to Flink, for example via `-Djava.security.krb5.conf=/path/to/krb5.conf` in Flink JVM options.

flink-connector/flink/build.gradle.kts

@@ -96,6 +96,7 @@ dependencies {
   testImplementation(libs.testcontainers.mysql)
   testImplementation(libs.metrics.core)
   testImplementation(libs.flinkjdbc)
+  testImplementation(libs.minikdc)
 
   testImplementation("org.apache.iceberg:iceberg-core:$icebergVersion")
   testImplementation("org.apache.iceberg:iceberg-hive-metastore:$icebergVersion")