[#9418] fix(iceberg): inject GCS FileIO token from gcs-service-account-file

catalogs/catalog-common/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergConstants.java

@@ -77,6 +77,20 @@ public class IcebergConstants {
   public static final String AZURE_CLIENT_SECRET_TOKEN_CREDENTIAL_PROVIDER =
       "org.apache.gravitino.iceberg.common.credential.AzureClientSecretTokenCredentialProvider";
 
+  /** Iceberg GCSFileIO OAuth2 access token property. */
+  public static final String ICEBERG_GCS_OAUTH2_TOKEN = "gcs.oauth2.token";
+
+  /** Iceberg GCSFileIO OAuth2 token expiry property (epoch millis). */
+  public static final String ICEBERG_GCS_OAUTH2_TOKEN_EXPIRES_AT = "gcs.oauth2.token-expires-at";
+
+  /**
+   * Whether Iceberg GCSFileIO should refresh OAuth2 tokens via a credentials endpoint. Defaults to
+   * true in Iceberg; Gravitino disables it when minting a token from {@code
+   * gcs-service-account-file} because that path has no table credentials refresh endpoint.
+   */
+  public static final String ICEBERG_GCS_OAUTH2_REFRESH_CREDENTIALS_ENABLED =
+      "gcs.oauth2.refresh-credentials-enabled";
+
   // Iceberg Table properties constants
 
   public static final String COMMENT = "comment";

core/src/test/java/org/apache/gravitino/catalog/TestClassLoaderPoolIntegration.java

@@ -22,6 +22,7 @@
 import java.io.IOException;
 import java.time.Instant;
 import java.util.Map;
+import java.util.concurrent.TimeUnit;
 import org.apache.commons.lang3.reflect.FieldUtils;
 import org.apache.gravitino.Catalog;
 import org.apache.gravitino.Config;
@@ -352,7 +353,18 @@ public void testClassLoaderPoolCleanupOnManagerClose() throws IllegalAccessExcep
 
     catalogManager.close();
 
-    // After close, the pool should be empty
+    // close() retires cached wrappers synchronously, but Caffeine's asMap() is only weakly
+    // consistent: invalidateAll may still deliver an asynchronous removal/retire for an entry
+    // the snapshot missed, which keeps the pooled ClassLoader alive until that listener runs.
+    long deadlineNs = System.nanoTime() + TimeUnit.SECONDS.toNanos(2);
+    while (pool.size() > 0 && System.nanoTime() < deadlineNs) {
+      try {
+        Thread.sleep(10L);
+      } catch (InterruptedException e) {
+        Thread.currentThread().interrupt();
+        break;
+      }
+    }
     Assertions.assertEquals(0, pool.size());
     catalogManager = null;
   }

docs/iceberg-rest-service.md

@@ -577,16 +577,17 @@ Please set the `gravitino.iceberg-rest.warehouse` parameter to `oss://{bucket_na
 
 Supports using static GCS credential file or generating GCS token to access GCS data.
 
-| Configuration item               | Description                                                                                                                  | Default value                           | Required |
-|----------------------------------|------------------------------------------------------------------------------------------------------------------------------|-----------------------------------------|----------|
-| `gravitino.iceberg-rest.io-impl` | The IO implementation for `FileIO` in Iceberg. Set it to `org.apache.iceberg.gcp.gcs.GCSFileIO` to explicitly use GCSFileIO. | `org.apache.iceberg.io.ResolvingFileIO` | No       |
+| Configuration item                                | Description                                                                                                                  | Default value                           | Required |
+|---------------------------------------------------|------------------------------------------------------------------------------------------------------------------------------|-----------------------------------------|----------|
+| `gravitino.iceberg-rest.io-impl`                  | The IO implementation for `FileIO` in Iceberg. Set it to `org.apache.iceberg.gcp.gcs.GCSFileIO` to explicitly use GCSFileIO. | `org.apache.iceberg.io.ResolvingFileIO` | No       |
+| `gravitino.iceberg-rest.gcs-service-account-file` | Path of the GCS service account JSON file. Used for server-side FileIO and for `gcs-token` credential vending.               | GCS Application default credential.     | No       |
 
 For other Iceberg GCS properties not managed by Gravitino like `gcs.project-id`, you could config it directly by `gravitino.iceberg-rest.gcs.project-id`.
 
 Refer to [GCS credentials](./security/credential-vending.md#gcs-credentials) for credential related configurations.
 
 :::note
-Ensure that the credential file is accessible by the Gravitino server. For example, the server may be running on a GCE machine, or you may set the environment variable `export GOOGLE_APPLICATION_CREDENTIALS=/xx/application_default_credentials.json` even when `gcs-service-account-file` is already configured.
+When `gcs-service-account-file` is set, Gravitino loads it at catalog initialization and injects Iceberg `gcs.oauth2.token` for FileIO. The IRC catalog cache evicts that catalog before the token expires so the next request recreates the catalog and mints a fresh token. If unset, use Application Default Credentials (for example GCE metadata or `GOOGLE_APPLICATION_CREDENTIALS`).
 :::
 
 :::info

docs/lakehouse-iceberg-catalog.md

@@ -175,13 +175,14 @@ The Gravitino Iceberg aliyun bundle jar already includes the Iceberg aliyun nece
 
 Supports using google credential file to access GCS data.
 
-| Configuration item | Description                                                                                                                  | Default value                           | Required |
-|--------------------|------------------------------------------------------------------------------------------------------------------------------|-----------------------------------------|----------|
-| `io-impl`          | The IO implementation for `FileIO` in Iceberg. Set it to `org.apache.iceberg.gcp.gcs.GCSFileIO` to explicitly use GCSFileIO. | `org.apache.iceberg.io.ResolvingFileIO` | No       |
+| Configuration item         | Description                                                                                                                  | Default value                           | Required |
+|----------------------------|------------------------------------------------------------------------------------------------------------------------------|-----------------------------------------|----------|
+| `io-impl`                  | The IO implementation for `FileIO` in Iceberg. Set it to `org.apache.iceberg.gcp.gcs.GCSFileIO` to explicitly use GCSFileIO. | `org.apache.iceberg.io.ResolvingFileIO` | No       |
+| `gcs-service-account-file` | Path of the GCS service account JSON file. Used for server-side FileIO and for `gcs-token` credential vending.               | GCS Application default credential.     | No       |
 
 For other Iceberg GCS properties not managed by Gravitino like `gcs.project-id`, you could config it directly by `gravitino.bypass.gcs.project-id`.
 
-Please make sure the credential file is accessible by Gravitino, like using `export GOOGLE_APPLICATION_CREDENTIALS=/xx/application_default_credentials.json` before Gravitino server is started.
+When `gcs-service-account-file` is set, Gravitino loads it at catalog initialization and injects Iceberg `gcs.oauth2.token` for FileIO (Iceberg's `GCSFileIO` has no service-account-file property). If that property is unset, fall back to Application Default Credentials, for example `export GOOGLE_APPLICATION_CREDENTIALS=/xx/application_default_credentials.json`.
 
 :::info
 Please set `warehouse` to `gs://{bucket_name}/${prefix_name}`, and download [Gravitino Iceberg GCP bundle jar](https://mvnrepository.com/artifact/org.apache.gravitino/gravitino-iceberg-gcp-bundle) and place it to `catalogs/lakehouse-iceberg/libs/`.

docs/security/credential-vending.md

@@ -403,7 +403,7 @@ There is no role to assume. The identity is the service account in `gcs-service-
 |----------------------------|------------------------------------------|-------------------------------------|----------|
 | `gcs-service-account-file` | The location of the GCS credential file. | GCS Application default credential. | No       |
 
-For the IRC, ensure that the credential file is accessible by that server. For example, the server may be running on a GCE machine, or you may set the environment variable `export GOOGLE_APPLICATION_CREDENTIALS=/xx/application_default_credentials.json` even when `gcs-service-account-file` is already configured.
+`gcs-service-account-file` is used both to vend downscoped tokens and to authenticate Iceberg `GCSFileIO` on the server (Gravitino injects `gcs.oauth2.token` at catalog load because Iceberg has no service-account-file property). Ensure the file is readable by the server process. If the property is unset, FileIO and token vending fall back to Application Default Credentials (for example GCE metadata or `GOOGLE_APPLICATION_CREDENTIALS`).
 
 ## Requesting Vended Credentials
 

iceberg/iceberg-common/build.gradle.kts

@@ -62,6 +62,8 @@ dependencies {
   implementation(libs.iceberg.azure)
   implementation(libs.iceberg.hive.metastore)
   implementation(libs.iceberg.gcp)
+  // Load gcs-service-account-file into Iceberg GCSFileIO properties (gcs.oauth2.token).
+  implementation(libs.google.auth.http)
   // Upgrade to Hadoop 3.3+ for Iceberg 1.10 compatibility
   // Iceberg 1.10 requires Hadoop 3.3+ APIs like FileSystem.openFile() and FsTracer.get()
   implementation(libs.hadoop3.client.api)

iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/utils/IcebergCatalogUtil.java

@@ -21,15 +21,25 @@
 import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.HADOOP_SECURITY_AUTHENTICATION;
 import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.HADOOP_SECURITY_AUTHORIZATION;
 
+import com.google.auth.oauth2.AccessToken;
+import com.google.auth.oauth2.GoogleCredentials;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.Maps;
 import java.io.IOException;
+import java.io.InputStream;
+import java.io.UncheckedIOException;
+import java.nio.file.Files;
+import java.nio.file.NoSuchFileException;
+import java.nio.file.Path;
+import java.nio.file.Paths;
 import java.sql.SQLException;
 import java.util.Collections;
+import java.util.Date;
 import java.util.HashMap;
 import java.util.Locale;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergCatalogBackend;
 import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants;
 import org.apache.gravitino.exceptions.ConnectionFailedException;
@@ -38,6 +48,7 @@
 import org.apache.gravitino.iceberg.common.IcebergConfig;
 import org.apache.gravitino.iceberg.common.authentication.AuthenticationConfig;
 import org.apache.gravitino.iceberg.common.rest.auth.UserPrincipalForwardingAuthManager;
+import org.apache.gravitino.storage.GCSProperties;
 import org.apache.hadoop.hdfs.HdfsConfiguration;
 import org.apache.iceberg.CatalogProperties;
 import org.apache.iceberg.CatalogUtil;
@@ -65,6 +76,9 @@ public class IcebergCatalogUtil {
    */
   private static final String ICEBERG_TYPE_COLUMN = "iceberg_type";
 
+  private static final String GCS_CLOUD_PLATFORM_SCOPE =
+      "https://www.googleapis.com/auth/cloud-platform";
+
   private static final ConcurrentHashMap<String, InMemoryCatalog> MEMORY_CATALOGS =
       new ConcurrentHashMap<>();
 
@@ -269,6 +283,84 @@ private static Catalog loadCustomCatalog(IcebergConfig icebergConfig) {
   @VisibleForTesting
   public static void applyDefaultResolvingFileIO(Map<String, String> properties) {
     properties.putIfAbsent(IcebergConstants.IO_IMPL, ResolvingFileIO.class.getName());
+    applyGcsServiceAccountCredentials(properties);
+  }
+
+  /**
+   * When {@code gcs-service-account-file} is set, mint an OAuth2 access token and inject Iceberg
+   * {@code gcs.oauth2.token} / {@code gcs.oauth2.token-expires-at} so the built-in {@code
+   * GCSFileIO} can authenticate. Iceberg's FileIO does not understand Gravitino's
+   * service-account-file property; S3/OSS/ADLS instead map static keys directly via {@link
+   * org.apache.gravitino.catalog.lakehouse.iceberg.IcebergPropertiesUtils}.
+   *
+   * <p>Skips injection when {@code gcs.oauth2.token} is already present. Disables Iceberg's
+   * credentials-endpoint refresh because that path is for vended table credentials, not catalog
+   * bootstrap from a service account file.
+   *
+   * @param properties Iceberg catalog properties, mutated in place
+   */
+  @VisibleForTesting
+  static void applyGcsServiceAccountCredentials(Map<String, String> properties) {
+    String serviceAccountFile = properties.get(GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE);
+    if (StringUtils.isBlank(serviceAccountFile)) {
+      return;
+    }
+    if (StringUtils.isNotBlank(properties.get(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN))) {
+      return;
+    }
+
+    AccessToken accessToken = loadAccessTokenFromFile(serviceAccountFile);
+    if (accessToken == null || StringUtils.isBlank(accessToken.getTokenValue())) {
+      throw new IllegalStateException(
+          "Failed to obtain GCS access token from service account file: " + serviceAccountFile);
+    }
+
+    properties.put(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN, accessToken.getTokenValue());
+    Date expirationTime = accessToken.getExpirationTime();
+    if (expirationTime != null) {
+      properties.put(
+          IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN_EXPIRES_AT,
+          String.valueOf(expirationTime.toInstant().toEpochMilli()));
+    }
+    properties.put(IcebergConstants.ICEBERG_GCS_OAUTH2_REFRESH_CREDENTIALS_ENABLED, "false");
+    LOG.info(
+        "Injected {} from {} for Iceberg GCSFileIO",
+        IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN,
+        GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE);
+  }
+
+  /**
+   * Returns an {@link IcebergConfig} that includes a minted GCS OAuth2 token when {@code
+   * gcs-service-account-file} is configured. The returned config retains {@code
+   * gcs.oauth2.token-expires-at} so callers (for example the IRC catalog cache) can expire the
+   * catalog before the token becomes invalid.
+   *
+   * @param icebergConfig original catalog config
+   * @return the same instance when no token is injected; otherwise a new config with token fields
+   */
+  public static IcebergConfig withGcsServiceAccountCredentials(IcebergConfig icebergConfig) {
+    Map<String, String> properties = new HashMap<>(icebergConfig.getAllConfig());
+    applyGcsServiceAccountCredentials(properties);
+    if (properties.equals(icebergConfig.getAllConfig())) {
+      return icebergConfig;
+    }
+    return new IcebergConfig(properties);
+  }
+
+  private static AccessToken loadAccessTokenFromFile(String serviceAccountFile) {
+    Path credentialsFilePath = Paths.get(serviceAccountFile);
+    try (InputStream inputStream = Files.newInputStream(credentialsFilePath)) {
+      GoogleCredentials credentials =
+          GoogleCredentials.fromStream(inputStream).createScoped(GCS_CLOUD_PLATFORM_SCOPE);
+      credentials.refreshIfExpired();
+      return credentials.getAccessToken();
+    } catch (NoSuchFileException e) {
+      throw new UncheckedIOException(
+          "GCS service account file does not exist: " + serviceAccountFile, e);
+    } catch (IOException e) {
+      throw new UncheckedIOException(
+          "Failed to load GCS service account file: " + serviceAccountFile, e);
+    }
   }
 
   @VisibleForTesting

iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/utils/TestIcebergCatalogUtil.java

@@ -19,6 +19,7 @@
 
 package org.apache.gravitino.iceberg.common.utils;
 
+import java.io.UncheckedIOException;
 import java.nio.file.Path;
 import java.sql.SQLException;
 import java.sql.SQLSyntaxErrorException;
@@ -29,6 +30,7 @@
 import org.apache.gravitino.iceberg.common.ClosableJdbcCatalog;
 import org.apache.gravitino.iceberg.common.IcebergConfig;
 import org.apache.gravitino.iceberg.common.authentication.AuthenticationConfig;
+import org.apache.gravitino.storage.GCSProperties;
 import org.apache.iceberg.CatalogProperties;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.catalog.Catalog;
@@ -337,6 +339,74 @@ void testApplyDefaultResolvingFileIODoesNotOverrideExplicitIOImpl() {
         "org.apache.iceberg.aws.s3.S3FileIO", properties.get(IcebergConstants.IO_IMPL));
   }
 
+  @Test
+  void testApplyGcsServiceAccountCredentialsSkipsWhenTokenAlreadyPresent() {
+    Map<String, String> properties = new HashMap<>();
+    properties.put(GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE, "/tmp/gcs-key.json");
+    properties.put(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN, "existing-token");
+
+    IcebergCatalogUtil.applyGcsServiceAccountCredentials(properties);
+
+    Assertions.assertEquals(
+        "existing-token", properties.get(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN));
+    Assertions.assertNull(properties.get(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN_EXPIRES_AT));
+  }
+
+  @Test
+  void testApplyGcsServiceAccountCredentialsNoOpWithoutServiceAccountFile() {
+    Map<String, String> properties = new HashMap<>();
+    properties.put(IcebergConstants.IO_IMPL, "org.apache.iceberg.gcp.gcs.GCSFileIO");
+
+    IcebergCatalogUtil.applyGcsServiceAccountCredentials(properties);
+
+    Assertions.assertNull(properties.get(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN));
+  }
+
+  @Test
+  void testApplyGcsServiceAccountCredentialsFailsWhenFileMissing() {
+    Map<String, String> properties = new HashMap<>();
+    properties.put(
+        GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE, "/tmp/gravitino-missing-gcs-key.json");
+
+    UncheckedIOException thrown =
+        Assertions.assertThrows(
+            UncheckedIOException.class,
+            () -> IcebergCatalogUtil.applyGcsServiceAccountCredentials(properties));
+    Assertions.assertTrue(thrown.getMessage().contains("does not exist"));
+  }
+
+  @Test
+  void testWithGcsServiceAccountCredentialsReturnsSameConfigWhenNoServiceAccountFile() {
+    IcebergConfig config = new IcebergConfig(Map.of(IcebergConstants.CATALOG_BACKEND, "memory"));
+    Assertions.assertSame(config, IcebergCatalogUtil.withGcsServiceAccountCredentials(config));
+  }
+
+  @Test
+  void testWithGcsServiceAccountCredentialsReturnsSameConfigWhenTokenAlreadyPresent() {
+    Map<String, String> properties = new HashMap<>();
+    properties.put(GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE, "/tmp/gcs-key.json");
+    properties.put(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN, "existing-token");
+    IcebergConfig config = new IcebergConfig(properties);
+
+    Assertions.assertSame(config, IcebergCatalogUtil.withGcsServiceAccountCredentials(config));
+  }
+
+  @Test
+  void testApplyDefaultResolvingFileIOInjectsGcsToken() {
+    Map<String, String> properties = new HashMap<>();
+    properties.put(IcebergConstants.WAREHOUSE, "gs://bucket/warehouse");
+    properties.put(GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE, "/tmp/gcs-key.json");
+
+    // Pre-set token so applyDefaultResolvingFileIO skips loading a real service account file.
+    properties.put(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN, "pre-set");
+    IcebergCatalogUtil.applyDefaultResolvingFileIO(properties);
+
+    Assertions.assertEquals(
+        org.apache.iceberg.io.ResolvingFileIO.class.getName(),
+        properties.get(IcebergConstants.IO_IMPL));
+    Assertions.assertEquals("pre-set", properties.get(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN));
+  }
+
   @Test
   void testApplyRestCatalogHttpTimeoutPropertiesUsesDefaults() {
     Map<String, String> properties = new HashMap<>();