From 9fcf914e242f7b1927c52c568daa9f673d19b291 Mon Sep 17 00:00:00 2001 From: remilu <1175334135@qq.com> Date: Mon, 7 Sep 2026 18:09:02 +0800 Subject: [PATCH 01/11] [#9418] fix(iceberg): inject GCS FileIO token from gcs-service-account-file Iceberg's GCSFileIO ignores gcs-service-account-file; mint gcs.oauth2.token at catalog load so server-side FileIO works from catalog properties alone. Co-authored-by: Cursor --- .../lakehouse/iceberg/IcebergConstants.java | 14 ++++ docs/iceberg-rest-service.md | 9 ++- docs/lakehouse-iceberg-catalog.md | 9 ++- docs/security/credential-vending.md | 2 +- iceberg/iceberg-common/build.gradle.kts | 2 + .../common/utils/IcebergCatalogUtil.java | 81 +++++++++++++++++++ .../common/utils/TestIcebergCatalogUtil.java | 71 ++++++++++++++++ 7 files changed, 179 insertions(+), 9 deletions(-) diff --git a/catalogs/catalog-common/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergConstants.java b/catalogs/catalog-common/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergConstants.java index 2d29e9e754f..41c95dec25a 100644 --- a/catalogs/catalog-common/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergConstants.java +++ b/catalogs/catalog-common/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergConstants.java @@ -65,6 +65,20 @@ public class IcebergConstants { "adls.auth.shared-key.account.name"; public static final String ICEBERG_ADLS_STORAGE_ACCOUNT_KEY = "adls.auth.shared-key.account.key"; + /** 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"; diff --git a/docs/iceberg-rest-service.md b/docs/iceberg-rest-service.md index 8339d69268e..8dbfed06dc7 100644 --- a/docs/iceberg-rest-service.md +++ b/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. If unset, use Application Default Credentials (for example GCE metadata or `GOOGLE_APPLICATION_CREDENTIALS`). ::: :::info diff --git a/docs/lakehouse-iceberg-catalog.md b/docs/lakehouse-iceberg-catalog.md index 996c0427e9f..6f4741f15fc 100644 --- a/docs/lakehouse-iceberg-catalog.md +++ b/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/`. diff --git a/docs/security/credential-vending.md b/docs/security/credential-vending.md index 8e0c798af34..7b1aeb9a010 100755 --- a/docs/security/credential-vending.md +++ b/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 diff --git a/iceberg/iceberg-common/build.gradle.kts b/iceberg/iceberg-common/build.gradle.kts index 427df238ea6..c2373e59db0 100644 --- a/iceberg/iceberg-common/build.gradle.kts +++ b/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) diff --git a/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/utils/IcebergCatalogUtil.java b/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/utils/IcebergCatalogUtil.java index 745b6b1d4f6..80547235302 100644 --- a/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/utils/IcebergCatalogUtil.java +++ b/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/utils/IcebergCatalogUtil.java @@ -21,15 +21,26 @@ 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 java.util.function.Function; +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 +49,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 +77,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 MEMORY_CATALOGS = new ConcurrentHashMap<>(); @@ -269,6 +284,72 @@ private static Catalog loadCustomCatalog(IcebergConfig icebergConfig) { @VisibleForTesting public static void applyDefaultResolvingFileIO(Map 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}. + * + *

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 + public static void applyGcsServiceAccountCredentials(Map properties) { + applyGcsServiceAccountCredentials(properties, IcebergCatalogUtil::loadAccessTokenFromFile); + } + + @VisibleForTesting + static void applyGcsServiceAccountCredentials( + Map properties, Function tokenLoader) { + 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 = tokenLoader.apply(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); + } + + 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 diff --git a/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/utils/TestIcebergCatalogUtil.java b/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/utils/TestIcebergCatalogUtil.java index f5ee1b8cf01..deed5ddf022 100644 --- a/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/utils/TestIcebergCatalogUtil.java +++ b/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/utils/TestIcebergCatalogUtil.java @@ -19,9 +19,12 @@ package org.apache.gravitino.iceberg.common.utils; +import com.google.auth.oauth2.AccessToken; import java.nio.file.Path; import java.sql.SQLException; import java.sql.SQLSyntaxErrorException; +import java.time.Instant; +import java.util.Date; import java.util.HashMap; import java.util.Map; import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergCatalogBackend; @@ -29,6 +32,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 +341,73 @@ void testApplyDefaultResolvingFileIODoesNotOverrideExplicitIOImpl() { "org.apache.iceberg.aws.s3.S3FileIO", properties.get(IcebergConstants.IO_IMPL)); } + @Test + void testApplyGcsServiceAccountCredentialsInjectsOauth2Token() { + Map properties = new HashMap<>(); + properties.put(GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE, "/tmp/gcs-key.json"); + + IcebergCatalogUtil.applyGcsServiceAccountCredentials( + properties, + path -> new AccessToken("test-token", Date.from(Instant.ofEpochMilli(1_700_000_000_000L)))); + + Assertions.assertEquals( + "test-token", properties.get(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN)); + Assertions.assertEquals( + "1700000000000", properties.get(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN_EXPIRES_AT)); + Assertions.assertEquals( + "false", properties.get(IcebergConstants.ICEBERG_GCS_OAUTH2_REFRESH_CREDENTIALS_ENABLED)); + Assertions.assertEquals( + "/tmp/gcs-key.json", properties.get(GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE)); + } + + @Test + void testApplyGcsServiceAccountCredentialsSkipsWhenTokenAlreadyPresent() { + Map 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, + path -> { + throw new AssertionError("token loader should not be called"); + }); + + 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 properties = new HashMap<>(); + properties.put(IcebergConstants.IO_IMPL, "org.apache.iceberg.gcp.gcs.GCSFileIO"); + + IcebergCatalogUtil.applyGcsServiceAccountCredentials( + properties, + path -> { + throw new AssertionError("token loader should not be called"); + }); + + Assertions.assertNull(properties.get(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN)); + } + + @Test + void testApplyDefaultResolvingFileIOInjectsGcsToken() { + Map properties = new HashMap<>(); + properties.put(IcebergConstants.WAREHOUSE, "gs://bucket/warehouse"); + properties.put(GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE, "/tmp/gcs-key.json"); + + // Exercise the public entry that wires credential injection; use the overload via a + // pre-injected token path to avoid network calls in unit tests. + 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 properties = new HashMap<>(); From d0bee57afab25b8bd63c0ac10bc70de78fd66d82 Mon Sep 17 00:00:00 2001 From: remilu <1175334135@qq.com> Date: Mon, 7 Sep 2026 19:16:50 +0800 Subject: [PATCH 02/11] [#9418] docs: align GCS configuration table columns Co-authored-by: Cursor --- docs/iceberg-rest-service.md | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/docs/iceberg-rest-service.md b/docs/iceberg-rest-service.md index 8dbfed06dc7..1836f9423e5 100644 --- a/docs/iceberg-rest-service.md +++ b/docs/iceberg-rest-service.md @@ -577,10 +577,10 @@ 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 | -| `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 | +| 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`. From e9d3973d329fd4d81e51ae7685e18e1e5e2a0194 Mon Sep 17 00:00:00 2001 From: remilu <1175334135@qq.com> Date: Mon, 7 Sep 2026 19:19:13 +0800 Subject: [PATCH 03/11] [#9418] refactor(iceberg): drop test-only GCS token loader overload Co-authored-by: Cursor --- .../common/utils/IcebergCatalogUtil.java | 11 +--- .../common/utils/TestIcebergCatalogUtil.java | 51 +++++++------------ 2 files changed, 19 insertions(+), 43 deletions(-) diff --git a/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/utils/IcebergCatalogUtil.java b/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/utils/IcebergCatalogUtil.java index 80547235302..972577b516a 100644 --- a/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/utils/IcebergCatalogUtil.java +++ b/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/utils/IcebergCatalogUtil.java @@ -39,7 +39,6 @@ import java.util.Locale; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; -import java.util.function.Function; import org.apache.commons.lang3.StringUtils; import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergCatalogBackend; import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants; @@ -301,13 +300,7 @@ public static void applyDefaultResolvingFileIO(Map properties) { * @param properties Iceberg catalog properties, mutated in place */ @VisibleForTesting - public static void applyGcsServiceAccountCredentials(Map properties) { - applyGcsServiceAccountCredentials(properties, IcebergCatalogUtil::loadAccessTokenFromFile); - } - - @VisibleForTesting - static void applyGcsServiceAccountCredentials( - Map properties, Function tokenLoader) { + static void applyGcsServiceAccountCredentials(Map properties) { String serviceAccountFile = properties.get(GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE); if (StringUtils.isBlank(serviceAccountFile)) { return; @@ -316,7 +309,7 @@ static void applyGcsServiceAccountCredentials( return; } - AccessToken accessToken = tokenLoader.apply(serviceAccountFile); + 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); diff --git a/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/utils/TestIcebergCatalogUtil.java b/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/utils/TestIcebergCatalogUtil.java index deed5ddf022..069609a5915 100644 --- a/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/utils/TestIcebergCatalogUtil.java +++ b/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/utils/TestIcebergCatalogUtil.java @@ -19,12 +19,10 @@ package org.apache.gravitino.iceberg.common.utils; -import com.google.auth.oauth2.AccessToken; +import java.io.UncheckedIOException; import java.nio.file.Path; import java.sql.SQLException; import java.sql.SQLSyntaxErrorException; -import java.time.Instant; -import java.util.Date; import java.util.HashMap; import java.util.Map; import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergCatalogBackend; @@ -341,36 +339,13 @@ void testApplyDefaultResolvingFileIODoesNotOverrideExplicitIOImpl() { "org.apache.iceberg.aws.s3.S3FileIO", properties.get(IcebergConstants.IO_IMPL)); } - @Test - void testApplyGcsServiceAccountCredentialsInjectsOauth2Token() { - Map properties = new HashMap<>(); - properties.put(GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE, "/tmp/gcs-key.json"); - - IcebergCatalogUtil.applyGcsServiceAccountCredentials( - properties, - path -> new AccessToken("test-token", Date.from(Instant.ofEpochMilli(1_700_000_000_000L)))); - - Assertions.assertEquals( - "test-token", properties.get(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN)); - Assertions.assertEquals( - "1700000000000", properties.get(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN_EXPIRES_AT)); - Assertions.assertEquals( - "false", properties.get(IcebergConstants.ICEBERG_GCS_OAUTH2_REFRESH_CREDENTIALS_ENABLED)); - Assertions.assertEquals( - "/tmp/gcs-key.json", properties.get(GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE)); - } - @Test void testApplyGcsServiceAccountCredentialsSkipsWhenTokenAlreadyPresent() { Map 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, - path -> { - throw new AssertionError("token loader should not be called"); - }); + IcebergCatalogUtil.applyGcsServiceAccountCredentials(properties); Assertions.assertEquals( "existing-token", properties.get(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN)); @@ -382,23 +357,31 @@ void testApplyGcsServiceAccountCredentialsNoOpWithoutServiceAccountFile() { Map properties = new HashMap<>(); properties.put(IcebergConstants.IO_IMPL, "org.apache.iceberg.gcp.gcs.GCSFileIO"); - IcebergCatalogUtil.applyGcsServiceAccountCredentials( - properties, - path -> { - throw new AssertionError("token loader should not be called"); - }); + IcebergCatalogUtil.applyGcsServiceAccountCredentials(properties); Assertions.assertNull(properties.get(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN)); } + @Test + void testApplyGcsServiceAccountCredentialsFailsWhenFileMissing() { + Map 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 testApplyDefaultResolvingFileIOInjectsGcsToken() { Map properties = new HashMap<>(); properties.put(IcebergConstants.WAREHOUSE, "gs://bucket/warehouse"); properties.put(GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE, "/tmp/gcs-key.json"); - // Exercise the public entry that wires credential injection; use the overload via a - // pre-injected token path to avoid network calls in unit tests. + // Pre-set token so applyDefaultResolvingFileIO skips loading a real service account file. properties.put(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN, "pre-set"); IcebergCatalogUtil.applyDefaultResolvingFileIO(properties); From b0945460ff9987b220085979c548194fa88309b3 Mon Sep 17 00:00:00 2001 From: remilu <1175334135@qq.com> Date: Mon, 7 Sep 2026 19:49:54 +0800 Subject: [PATCH 04/11] [#9418] fix(iceberg): refresh GCS FileIO token via IRC catalog cache expiry Cap catalog-wrapper cache lifetime by gcs.oauth2.token-expires-at so hot catalogs recreate and remint OAuth2 tokens before they expire. Co-authored-by: Cursor --- docs/iceberg-rest-service.md | 2 +- .../common/utils/IcebergCatalogUtil.java | 18 ++++ .../common/utils/TestIcebergCatalogUtil.java | 16 +++ .../service/IcebergCatalogWrapperManager.java | 97 +++++++++++++++++-- ...stIcebergCatalogWrapperManagerForREST.java | 65 +++++++++++++ 5 files changed, 188 insertions(+), 10 deletions(-) diff --git a/docs/iceberg-rest-service.md b/docs/iceberg-rest-service.md index 1836f9423e5..f3c55cdafde 100644 --- a/docs/iceberg-rest-service.md +++ b/docs/iceberg-rest-service.md @@ -587,7 +587,7 @@ For other Iceberg GCS properties not managed by Gravitino like `gcs.project-id`, Refer to [GCS credentials](./security/credential-vending.md#gcs-credentials) for credential related configurations. :::note -When `gcs-service-account-file` is set, Gravitino loads it at catalog initialization and injects Iceberg `gcs.oauth2.token` for FileIO. If unset, use Application Default Credentials (for example GCE metadata or `GOOGLE_APPLICATION_CREDENTIALS`). +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 diff --git a/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/utils/IcebergCatalogUtil.java b/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/utils/IcebergCatalogUtil.java index 972577b516a..658786ea15b 100644 --- a/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/utils/IcebergCatalogUtil.java +++ b/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/utils/IcebergCatalogUtil.java @@ -329,6 +329,24 @@ static void applyGcsServiceAccountCredentials(Map properties) { 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 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)) { diff --git a/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/utils/TestIcebergCatalogUtil.java b/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/utils/TestIcebergCatalogUtil.java index 069609a5915..d8ce5cc18a4 100644 --- a/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/utils/TestIcebergCatalogUtil.java +++ b/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/utils/TestIcebergCatalogUtil.java @@ -375,6 +375,22 @@ void testApplyGcsServiceAccountCredentialsFailsWhenFileMissing() { 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 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 properties = new HashMap<>(); diff --git a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergCatalogWrapperManager.java b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergCatalogWrapperManager.java index 5d8fbf52f1e..90d8a8f4fe7 100644 --- a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergCatalogWrapperManager.java +++ b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergCatalogWrapperManager.java @@ -20,6 +20,7 @@ import com.github.benmanes.caffeine.cache.Cache; import com.github.benmanes.caffeine.cache.Caffeine; +import com.github.benmanes.caffeine.cache.Expiry; import com.github.benmanes.caffeine.cache.Scheduler; import com.google.common.annotations.VisibleForTesting; import com.google.common.util.concurrent.ThreadFactoryBuilder; @@ -28,6 +29,7 @@ import java.util.Optional; import java.util.concurrent.ScheduledThreadPoolExecutor; import java.util.concurrent.TimeUnit; +import org.apache.commons.lang3.StringUtils; import org.apache.gravitino.GravitinoEnv; import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergCatalogBackend; import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants; @@ -37,6 +39,7 @@ import org.apache.gravitino.iceberg.common.authentication.SupportsKerberos; import org.apache.gravitino.iceberg.common.ops.IcebergCatalogWrapper; import org.apache.gravitino.iceberg.common.ops.KerberosAwareIcebergCatalogProxy; +import org.apache.gravitino.iceberg.common.utils.IcebergCatalogUtil; import org.apache.gravitino.iceberg.service.authorization.IcebergRESTServerContext; import org.apache.gravitino.iceberg.service.provider.DynamicIcebergConfigProvider; import org.apache.gravitino.iceberg.service.provider.IcebergConfigProvider; @@ -47,6 +50,12 @@ public class IcebergCatalogWrapperManager implements AutoCloseable { public static final Logger LOG = LoggerFactory.getLogger(IcebergCatalogWrapperManager.class); + /** + * Evict a cached catalog this long before its minted GCS OAuth2 token expires, so the next + * request recreates the catalog and refreshes the token. + */ + @VisibleForTesting static final long GCS_TOKEN_REFRESH_BUFFER_MS = TimeUnit.MINUTES.toMillis(5); + private final Cache catalogWrapperCache; private final IcebergConfigProvider configProvider; @@ -57,12 +66,13 @@ public IcebergCatalogWrapperManager( boolean auxMode, String metalakeName) { this.configProvider = configProvider; + long accessEvictionNanos = + TimeUnit.MILLISECONDS.toNanos( + new IcebergConfig(properties) + .get(IcebergConfig.ICEBERG_REST_CATALOG_CACHE_EVICTION_INTERVAL)); this.catalogWrapperCache = Caffeine.newBuilder() - .expireAfterAccess( - (new IcebergConfig(properties)) - .get(IcebergConfig.ICEBERG_REST_CATALOG_CACHE_EVICTION_INTERVAL), - TimeUnit.MILLISECONDS) + .expireAfter(new CatalogWrapperExpiry(accessEvictionNanos)) .removalListener( (k, v, c) -> { String catalogName = (String) k; @@ -140,22 +150,26 @@ private CatalogWrapperForREST createCatalogWrapper(String catalogName) { @VisibleForTesting protected CatalogWrapperForREST createCatalogWrapper( String catalogName, IcebergConfig icebergConfig) { + // Mint GCS OAuth2 tokens into the config before constructing the wrapper so the IRC catalog + // cache can expire the entry before gcs.oauth2.token-expires-at. + IcebergConfig enrichedConfig = + IcebergCatalogUtil.withGcsServiceAccountCredentials(icebergConfig); // When the backend is a federated Iceberg REST catalog, use FederatedCatalogWrapper so // federation-aware behavior (FileIO property extraction, remote credential vending, remote // /v1/config defaults) is applied through polymorphic dispatch rather than scattered // instanceof checks. All other backends use the base CatalogWrapperForREST. IcebergCatalogBackend backend = IcebergCatalogBackend.valueOf( - icebergConfig.get(IcebergConfig.CATALOG_BACKEND).toUpperCase(Locale.ROOT)); + enrichedConfig.get(IcebergConfig.CATALOG_BACKEND).toUpperCase(Locale.ROOT)); CatalogWrapperForREST rest = backend == IcebergCatalogBackend.REST - ? new FederatedCatalogWrapper(catalogName, icebergConfig) - : new CatalogWrapperForREST(catalogName, icebergConfig); + ? new FederatedCatalogWrapper(catalogName, enrichedConfig) + : new CatalogWrapperForREST(catalogName, enrichedConfig); AuthenticationConfig authenticationConfig = - new AuthenticationConfig(icebergConfig.getAllConfig()); + new AuthenticationConfig(enrichedConfig.getAllConfig()); if (authenticationConfig.isKerberosAuth() && rest.getCatalog() instanceof SupportsKerberos) { return (CatalogWrapperForREST) - new KerberosAwareIcebergCatalogProxy(rest).getProxy(catalogName, icebergConfig); + new KerberosAwareIcebergCatalogProxy(rest).getProxy(catalogName, enrichedConfig); } return rest; @@ -169,8 +183,73 @@ private void closeIcebergCatalogWrapper(IcebergCatalogWrapper catalogWrapper) { } } + /** + * Computes how long a catalog wrapper may stay in the IRC cache. + * + *

Uses the configured access-based eviction interval, capped by the time until a minted GCS + * OAuth2 token should be refreshed ({@code gcs.oauth2.token-expires-at} minus {@link + * #GCS_TOKEN_REFRESH_BUFFER_MS}). When no token expiry is present, returns {@code + * accessEvictionNanos}. + * + * @param config catalog config that may contain {@code gcs.oauth2.token-expires-at} + * @param accessEvictionNanos default expire-after-access duration in nanoseconds + * @param nowEpochMillis current wall-clock time + * @return cache duration in nanoseconds; {@code 0} means expire immediately + */ + @VisibleForTesting + static long computeCacheDurationNanos( + IcebergConfig config, long accessEvictionNanos, long nowEpochMillis) { + String expiresAt = + config.getAllConfig().get(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN_EXPIRES_AT); + if (StringUtils.isBlank(expiresAt)) { + return accessEvictionNanos; + } + + long expiresAtMs; + try { + expiresAtMs = Long.parseLong(expiresAt); + } catch (NumberFormatException e) { + LOG.warn("Invalid {}: {}", IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN_EXPIRES_AT, expiresAt); + return accessEvictionNanos; + } + + long remainingMs = expiresAtMs - GCS_TOKEN_REFRESH_BUFFER_MS - nowEpochMillis; + if (remainingMs <= 0) { + return 0L; + } + return Math.min(accessEvictionNanos, TimeUnit.MILLISECONDS.toNanos(remainingMs)); + } + @Override public void close() throws Exception { catalogWrapperCache.invalidateAll(); } + + private static final class CatalogWrapperExpiry implements Expiry { + + private final long accessEvictionNanos; + + CatalogWrapperExpiry(long accessEvictionNanos) { + this.accessEvictionNanos = accessEvictionNanos; + } + + @Override + public long expireAfterCreate(String key, CatalogWrapperForREST value, long currentTime) { + return computeCacheDurationNanos( + value.getIcebergConfig(), accessEvictionNanos, System.currentTimeMillis()); + } + + @Override + public long expireAfterUpdate( + String key, CatalogWrapperForREST value, long currentTime, long currentDuration) { + return expireAfterCreate(key, value, currentTime); + } + + @Override + public long expireAfterRead( + String key, CatalogWrapperForREST value, long currentTime, long currentDuration) { + // Preserve expire-after-access, but never extend past the GCS token refresh deadline. + return expireAfterCreate(key, value, currentTime); + } + } } diff --git a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergCatalogWrapperManagerForREST.java b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergCatalogWrapperManagerForREST.java index 45eb62f69ee..3a5e8925e12 100644 --- a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergCatalogWrapperManagerForREST.java +++ b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergCatalogWrapperManagerForREST.java @@ -22,6 +22,7 @@ import com.google.common.collect.Maps; import java.util.Map; import java.util.Optional; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; import java.util.function.Consumer; import org.apache.commons.lang3.StringUtils; @@ -189,6 +190,70 @@ public void testDefaultCatalogAliasInvalidatedWhenCatalogRemoved() throws Except } } + @Test + public void testComputeCacheDurationNanosWithoutTokenExpiryUsesAccessEviction() { + IcebergConfig config = + new IcebergConfig(ImmutableMap.of(IcebergConstants.CATALOG_BACKEND, "memory")); + long accessEvictionNanos = TimeUnit.HOURS.toNanos(1); + Assertions.assertEquals( + accessEvictionNanos, + IcebergCatalogWrapperManager.computeCacheDurationNanos( + config, accessEvictionNanos, System.currentTimeMillis())); + } + + @Test + public void testComputeCacheDurationNanosCapsByGcsTokenExpiry() { + long now = 1_700_000_000_000L; + long expiresAt = now + TimeUnit.HOURS.toMillis(1); // token valid for 1h + IcebergConfig config = + new IcebergConfig( + ImmutableMap.of( + IcebergConstants.CATALOG_BACKEND, + "memory", + IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN_EXPIRES_AT, + String.valueOf(expiresAt))); + long accessEvictionNanos = TimeUnit.HOURS.toNanos(2); + long expected = + TimeUnit.MILLISECONDS.toNanos( + TimeUnit.HOURS.toMillis(1) - IcebergCatalogWrapperManager.GCS_TOKEN_REFRESH_BUFFER_MS); + Assertions.assertEquals( + expected, + IcebergCatalogWrapperManager.computeCacheDurationNanos(config, accessEvictionNanos, now)); + } + + @Test + public void testComputeCacheDurationNanosExpiresImmediatelyWhenPastRefreshDeadline() { + long now = 1_700_000_000_000L; + long expiresAt = now + TimeUnit.MINUTES.toMillis(2); // within 5-minute buffer + IcebergConfig config = + new IcebergConfig( + ImmutableMap.of( + IcebergConstants.CATALOG_BACKEND, + "memory", + IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN_EXPIRES_AT, + String.valueOf(expiresAt))); + Assertions.assertEquals( + 0L, + IcebergCatalogWrapperManager.computeCacheDurationNanos( + config, TimeUnit.HOURS.toNanos(1), now)); + } + + @Test + public void testComputeCacheDurationNanosIgnoresInvalidExpiresAt() { + IcebergConfig config = + new IcebergConfig( + ImmutableMap.of( + IcebergConstants.CATALOG_BACKEND, + "memory", + IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN_EXPIRES_AT, + "not-a-number")); + long accessEvictionNanos = TimeUnit.MINUTES.toNanos(30); + Assertions.assertEquals( + accessEvictionNanos, + IcebergCatalogWrapperManager.computeCacheDurationNanos( + config, accessEvictionNanos, System.currentTimeMillis())); + } + private static IcebergCatalogWrapperManager newManager() { Map config = Maps.newHashMap(); IcebergConfigProvider configProvider = IcebergConfigProviderFactory.create(config); From 4dbe53d82129566ae1079ed9c3b2aae7f38eef42 Mon Sep 17 00:00:00 2001 From: remilu <1175334135@qq.com> Date: Mon, 7 Sep 2026 20:12:41 +0800 Subject: [PATCH 05/11] [#9418] fix(iceberg): refresh GCS tokens via dedicated cache and FileIO Keep IRC catalog expire-after-access unchanged. Remint OAuth2 tokens in GcsAccessTokenCache and GravitinoGCSFileIO so credential lifetime is independent of catalog-wrapper eviction. Co-authored-by: Cursor --- docs/iceberg-rest-service.md | 2 +- docs/lakehouse-iceberg-catalog.md | 2 +- docs/security/credential-vending.md | 2 +- .../common/io/GcsAccessTokenCache.java | 131 +++++++++++++ .../iceberg/common/io/GravitinoGCSFileIO.java | 180 ++++++++++++++++++ .../common/utils/IcebergCatalogUtil.java | 94 ++------- .../common/io/TestGcsAccessTokenCache.java | 61 ++++++ .../common/utils/TestIcebergCatalogUtil.java | 64 +++---- .../service/CatalogWrapperForREST.java | 7 + .../service/IcebergCatalogWrapperManager.java | 97 +--------- ...stIcebergCatalogWrapperManagerForREST.java | 65 ------- 11 files changed, 441 insertions(+), 264 deletions(-) create mode 100644 iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/io/GcsAccessTokenCache.java create mode 100644 iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/io/GravitinoGCSFileIO.java create mode 100644 iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/io/TestGcsAccessTokenCache.java diff --git a/docs/iceberg-rest-service.md b/docs/iceberg-rest-service.md index f3c55cdafde..d6809bbd4de 100644 --- a/docs/iceberg-rest-service.md +++ b/docs/iceberg-rest-service.md @@ -587,7 +587,7 @@ For other Iceberg GCS properties not managed by Gravitino like `gcs.project-id`, Refer to [GCS credentials](./security/credential-vending.md#gcs-credentials) for credential related configurations. :::note -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`). +When `gcs-service-account-file` is set, Gravitino uses `GravitinoGCSFileIO`, which remints OAuth2 tokens from that file through a dedicated token cache. Token refresh is independent of the IRC catalog cache. If unset, use Application Default Credentials (for example GCE metadata or `GOOGLE_APPLICATION_CREDENTIALS`). ::: :::info diff --git a/docs/lakehouse-iceberg-catalog.md b/docs/lakehouse-iceberg-catalog.md index 6f4741f15fc..26e6b970e4a 100644 --- a/docs/lakehouse-iceberg-catalog.md +++ b/docs/lakehouse-iceberg-catalog.md @@ -182,7 +182,7 @@ Supports using google credential file to access GCS data. For other Iceberg GCS properties not managed by Gravitino like `gcs.project-id`, you could config it directly by `gravitino.bypass.gcs.project-id`. -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`. +When `gcs-service-account-file` is set, Gravitino uses `GravitinoGCSFileIO`, which remints OAuth2 tokens from that file through a dedicated token cache (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/`. diff --git a/docs/security/credential-vending.md b/docs/security/credential-vending.md index 7b1aeb9a010..6ae5c2125bd 100755 --- a/docs/security/credential-vending.md +++ b/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 | -`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`). +`gcs-service-account-file` is used both to vend downscoped tokens and to authenticate Iceberg FileIO on the server (via `GravitinoGCSFileIO` and a dedicated token cache, 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 diff --git a/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/io/GcsAccessTokenCache.java b/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/io/GcsAccessTokenCache.java new file mode 100644 index 00000000000..8b29828b0ff --- /dev/null +++ b/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/io/GcsAccessTokenCache.java @@ -0,0 +1,131 @@ +/* + * 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.gravitino.iceberg.common.io; + +import com.github.benmanes.caffeine.cache.Cache; +import com.github.benmanes.caffeine.cache.Caffeine; +import com.github.benmanes.caffeine.cache.Expiry; +import com.google.auth.oauth2.AccessToken; +import com.google.auth.oauth2.GoogleCredentials; +import com.google.common.annotations.VisibleForTesting; +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.util.concurrent.TimeUnit; + +/** + * Process-wide cache of GCS OAuth2 access tokens minted from service-account JSON files. + * + *

Token lifetime is independent of the IRC catalog-wrapper cache: entries expire shortly before + * the underlying Google access token expires, and the next lookup remints a fresh token from the + * service-account file. + */ +public final class GcsAccessTokenCache { + + private static final String GCS_CLOUD_PLATFORM_SCOPE = + "https://www.googleapis.com/auth/cloud-platform"; + + /** Evict this long before the Google access token expires so reminting happens early. */ + @VisibleForTesting static final long TOKEN_REFRESH_BUFFER_MS = TimeUnit.MINUTES.toMillis(5); + + private static final Cache CACHE = + Caffeine.newBuilder().expireAfter(new AccessTokenExpiry()).build(); + + private GcsAccessTokenCache() {} + + /** + * Returns a cached access token for {@code serviceAccountFile}, minting one on cache miss or + * after expiry. + * + * @param serviceAccountFile path to the GCS service-account JSON file + * @return a non-null access token + */ + public static AccessToken get(String serviceAccountFile) { + return CACHE.get(serviceAccountFile, GcsAccessTokenCache::loadAccessToken); + } + + @VisibleForTesting + static void invalidate(String serviceAccountFile) { + CACHE.invalidate(serviceAccountFile); + } + + @VisibleForTesting + static void invalidateAll() { + CACHE.invalidateAll(); + } + + @VisibleForTesting + static long durationUntilRefreshNanos(AccessToken accessToken, long nowEpochMillis) { + if (accessToken.getExpirationTime() == null) { + // No expiry from Google — keep the entry briefly so a bad token cannot live forever. + return TimeUnit.MINUTES.toNanos(30); + } + long remainingMs = + accessToken.getExpirationTime().toInstant().toEpochMilli() + - TOKEN_REFRESH_BUFFER_MS + - nowEpochMillis; + return remainingMs <= 0 ? 0L : TimeUnit.MILLISECONDS.toNanos(remainingMs); + } + + private static AccessToken loadAccessToken(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(); + AccessToken accessToken = credentials.getAccessToken(); + if (accessToken == null || accessToken.getTokenValue() == null) { + throw new IllegalStateException( + "Failed to obtain GCS access token from service account file: " + serviceAccountFile); + } + return accessToken; + } 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); + } + } + + private static final class AccessTokenExpiry implements Expiry { + + @Override + public long expireAfterCreate(String key, AccessToken value, long currentTime) { + return durationUntilRefreshNanos(value, System.currentTimeMillis()); + } + + @Override + public long expireAfterUpdate( + String key, AccessToken value, long currentTime, long currentDuration) { + return expireAfterCreate(key, value, currentTime); + } + + @Override + public long expireAfterRead( + String key, AccessToken value, long currentTime, long currentDuration) { + // Recalculate from wall clock so access does not stretch past token expiry. + return expireAfterCreate(key, value, currentTime); + } + } +} diff --git a/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/io/GravitinoGCSFileIO.java b/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/io/GravitinoGCSFileIO.java new file mode 100644 index 00000000000..ffaca8f93aa --- /dev/null +++ b/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/io/GravitinoGCSFileIO.java @@ -0,0 +1,180 @@ +/* + * 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.gravitino.iceberg.common.io; + +import com.google.auth.oauth2.AccessToken; +import java.util.Date; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import org.apache.commons.lang3.StringUtils; +import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants; +import org.apache.gravitino.storage.GCSProperties; +import org.apache.iceberg.gcp.gcs.GCSFileIO; +import org.apache.iceberg.io.BulkDeletionFailureException; +import org.apache.iceberg.io.DelegateFileIO; +import org.apache.iceberg.io.FileInfo; +import org.apache.iceberg.io.InputFile; +import org.apache.iceberg.io.OutputFile; +import org.apache.iceberg.io.StorageCredential; +import org.apache.iceberg.io.SupportsStorageCredentials; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * GCS {@link DelegateFileIO} that authenticates with {@code gcs-service-account-file}. + * + *

Iceberg's built-in {@link GCSFileIO} only understands a static {@code gcs.oauth2.token}. This + * wrapper keeps a delegate {@link GCSFileIO} in sync with {@link GcsAccessTokenCache}, recreating + * the delegate when the cached token is reminted. Token refresh does not depend on IRC catalog + * cache eviction. + */ +public class GravitinoGCSFileIO implements DelegateFileIO, SupportsStorageCredentials { + + private static final Logger LOG = LoggerFactory.getLogger(GravitinoGCSFileIO.class); + + private Map properties; + private String serviceAccountFile; + private volatile GCSFileIO delegate; + private volatile String boundToken; + + @Override + public void initialize(Map props) { + this.properties = new HashMap<>(props); + this.serviceAccountFile = props.get(GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE); + // Drop any stale delegate after re-initialize. + closeDelegate(); + currentDelegate(); + } + + @Override + public InputFile newInputFile(String path) { + return currentDelegate().newInputFile(path); + } + + @Override + public InputFile newInputFile(String path, long length) { + return currentDelegate().newInputFile(path, length); + } + + @Override + public OutputFile newOutputFile(String path) { + return currentDelegate().newOutputFile(path); + } + + @Override + public void deleteFile(String path) { + currentDelegate().deleteFile(path); + } + + @Override + public void deleteFiles(Iterable pathsToDelete) throws BulkDeletionFailureException { + currentDelegate().deleteFiles(pathsToDelete); + } + + @Override + public Iterable listPrefix(String prefix) { + return currentDelegate().listPrefix(prefix); + } + + @Override + public void deletePrefix(String prefix) { + currentDelegate().deletePrefix(prefix); + } + + @Override + public Map properties() { + return currentDelegate().properties(); + } + + @Override + public void setCredentials(List credentials) { + currentDelegate().setCredentials(credentials); + } + + @Override + public List credentials() { + return currentDelegate().credentials(); + } + + @Override + public void close() { + closeDelegate(); + } + + private GCSFileIO currentDelegate() { + if (StringUtils.isBlank(serviceAccountFile)) { + return ensureDelegate(null); + } + AccessToken accessToken = GcsAccessTokenCache.get(serviceAccountFile); + return ensureDelegate(accessToken); + } + + private GCSFileIO ensureDelegate(AccessToken accessToken) { + String tokenValue = accessToken == null ? null : accessToken.getTokenValue(); + GCSFileIO current = delegate; + if (current != null && java.util.Objects.equals(boundToken, tokenValue)) { + return current; + } + synchronized (this) { + current = delegate; + if (current != null && java.util.Objects.equals(boundToken, tokenValue)) { + return current; + } + closeDelegate(); + Map delegateProperties = new HashMap<>(properties); + if (accessToken != null) { + delegateProperties.put( + IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN, accessToken.getTokenValue()); + Date expirationTime = accessToken.getExpirationTime(); + if (expirationTime != null) { + delegateProperties.put( + IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN_EXPIRES_AT, + String.valueOf(expirationTime.toInstant().toEpochMilli())); + } + // Catalog bootstrap from a service-account file has no table credentials endpoint. + delegateProperties.put( + IcebergConstants.ICEBERG_GCS_OAUTH2_REFRESH_CREDENTIALS_ENABLED, "false"); + LOG.debug( + "Binding GCSFileIO delegate to token from {}", + GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE); + } + GCSFileIO created = new GCSFileIO(); + created.initialize(delegateProperties); + this.delegate = created; + this.boundToken = tokenValue; + return created; + } + } + + private void closeDelegate() { + GCSFileIO current = delegate; + if (current == null) { + return; + } + try { + current.close(); + } catch (RuntimeException e) { + LOG.warn("Failed to close GCSFileIO delegate", e); + } finally { + this.delegate = null; + this.boundToken = null; + } + } +} diff --git a/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/utils/IcebergCatalogUtil.java b/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/utils/IcebergCatalogUtil.java index 658786ea15b..87c781b2789 100644 --- a/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/utils/IcebergCatalogUtil.java +++ b/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/utils/IcebergCatalogUtil.java @@ -21,20 +21,10 @@ 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; @@ -47,12 +37,14 @@ 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.iceberg.common.io.GravitinoGCSFileIO; 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; import org.apache.iceberg.catalog.Catalog; +import org.apache.iceberg.gcp.gcs.GCSFileIO; import org.apache.iceberg.hive.HiveCatalog; import org.apache.iceberg.hive.HiveCatalogWithMetadataLocationSupport; import org.apache.iceberg.inmemory.InMemoryCatalog; @@ -76,9 +68,6 @@ 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 MEMORY_CATALOGS = new ConcurrentHashMap<>(); @@ -283,24 +272,23 @@ private static Catalog loadCustomCatalog(IcebergConfig icebergConfig) { @VisibleForTesting public static void applyDefaultResolvingFileIO(Map properties) { properties.putIfAbsent(IcebergConstants.IO_IMPL, ResolvingFileIO.class.getName()); - applyGcsServiceAccountCredentials(properties); + applyGcsServiceAccountFileIO(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}. + * When {@code gcs-service-account-file} is set and no explicit {@code gcs.oauth2.token} is + * present, switch the FileIO to {@link GravitinoGCSFileIO}. That FileIO remints tokens through + * {@link org.apache.gravitino.iceberg.common.io.GcsAccessTokenCache}, independently of the IRC + * catalog-wrapper cache. Iceberg's built-in {@link GCSFileIO} / {@link ResolvingFileIO} path does + * not understand Gravitino's service-account-file property. * - *

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. + *

Skips when {@code gcs.oauth2.token} is already present, or when the user set a custom {@code + * io-impl} other than {@link ResolvingFileIO} / {@link GCSFileIO}. * * @param properties Iceberg catalog properties, mutated in place */ @VisibleForTesting - static void applyGcsServiceAccountCredentials(Map properties) { + static void applyGcsServiceAccountFileIO(Map properties) { String serviceAccountFile = properties.get(GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE); if (StringUtils.isBlank(serviceAccountFile)) { return; @@ -309,57 +297,15 @@ static void applyGcsServiceAccountCredentials(Map properties) { 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 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); + String ioImpl = properties.get(IcebergConstants.IO_IMPL); + if (ioImpl == null + || ResolvingFileIO.class.getName().equals(ioImpl) + || GCSFileIO.class.getName().equals(ioImpl)) { + properties.put(IcebergConstants.IO_IMPL, GravitinoGCSFileIO.class.getName()); + LOG.info( + "Using {} for catalog FileIO because {} is set", + GravitinoGCSFileIO.class.getName(), + GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE); } } diff --git a/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/io/TestGcsAccessTokenCache.java b/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/io/TestGcsAccessTokenCache.java new file mode 100644 index 00000000000..0c926ce2545 --- /dev/null +++ b/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/io/TestGcsAccessTokenCache.java @@ -0,0 +1,61 @@ +/* + * 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.gravitino.iceberg.common.io; + +import com.google.auth.oauth2.AccessToken; +import java.io.UncheckedIOException; +import java.util.Date; +import java.util.concurrent.TimeUnit; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +public class TestGcsAccessTokenCache { + + @AfterEach + void tearDown() { + GcsAccessTokenCache.invalidateAll(); + } + + @Test + void testDurationUntilRefreshNanosUsesTokenExpiryMinusBuffer() { + long now = 1_700_000_000_000L; + AccessToken token = new AccessToken("token", new Date(now + TimeUnit.HOURS.toMillis(1))); + long expected = + TimeUnit.MILLISECONDS.toNanos( + TimeUnit.HOURS.toMillis(1) - GcsAccessTokenCache.TOKEN_REFRESH_BUFFER_MS); + Assertions.assertEquals(expected, GcsAccessTokenCache.durationUntilRefreshNanos(token, now)); + } + + @Test + void testDurationUntilRefreshNanosExpiresImmediatelyNearExpiry() { + long now = 1_700_000_000_000L; + AccessToken token = new AccessToken("token", new Date(now + TimeUnit.MINUTES.toMillis(2))); + Assertions.assertEquals(0L, GcsAccessTokenCache.durationUntilRefreshNanos(token, now)); + } + + @Test + void testGetFailsWhenServiceAccountFileMissing() { + UncheckedIOException thrown = + Assertions.assertThrows( + UncheckedIOException.class, + () -> GcsAccessTokenCache.get("/tmp/gravitino-missing-gcs-key.json")); + Assertions.assertTrue(thrown.getMessage().contains("does not exist")); + } +} diff --git a/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/utils/TestIcebergCatalogUtil.java b/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/utils/TestIcebergCatalogUtil.java index d8ce5cc18a4..3fad7d002ad 100644 --- a/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/utils/TestIcebergCatalogUtil.java +++ b/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/utils/TestIcebergCatalogUtil.java @@ -19,7 +19,6 @@ package org.apache.gravitino.iceberg.common.utils; -import java.io.UncheckedIOException; import java.nio.file.Path; import java.sql.SQLException; import java.sql.SQLSyntaxErrorException; @@ -340,71 +339,68 @@ void testApplyDefaultResolvingFileIODoesNotOverrideExplicitIOImpl() { } @Test - void testApplyGcsServiceAccountCredentialsSkipsWhenTokenAlreadyPresent() { + void testApplyGcsServiceAccountFileIOSwitchesFromResolvingFileIO() { Map properties = new HashMap<>(); + properties.put(IcebergConstants.WAREHOUSE, "gs://bucket/warehouse"); properties.put(GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE, "/tmp/gcs-key.json"); - properties.put(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN, "existing-token"); - IcebergCatalogUtil.applyGcsServiceAccountCredentials(properties); + IcebergCatalogUtil.applyDefaultResolvingFileIO(properties); Assertions.assertEquals( - "existing-token", properties.get(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN)); - Assertions.assertNull(properties.get(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN_EXPIRES_AT)); + org.apache.gravitino.iceberg.common.io.GravitinoGCSFileIO.class.getName(), + properties.get(IcebergConstants.IO_IMPL)); + Assertions.assertNull(properties.get(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN)); } @Test - void testApplyGcsServiceAccountCredentialsNoOpWithoutServiceAccountFile() { + void testApplyGcsServiceAccountFileIOSwitchesFromGcsFileIO() { Map properties = new HashMap<>(); properties.put(IcebergConstants.IO_IMPL, "org.apache.iceberg.gcp.gcs.GCSFileIO"); + properties.put(GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE, "/tmp/gcs-key.json"); - IcebergCatalogUtil.applyGcsServiceAccountCredentials(properties); + IcebergCatalogUtil.applyDefaultResolvingFileIO(properties); - Assertions.assertNull(properties.get(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN)); + Assertions.assertEquals( + org.apache.gravitino.iceberg.common.io.GravitinoGCSFileIO.class.getName(), + properties.get(IcebergConstants.IO_IMPL)); } @Test - void testApplyGcsServiceAccountCredentialsFailsWhenFileMissing() { + void testApplyGcsServiceAccountFileIOSkipsWhenTokenAlreadyPresent() { Map properties = new HashMap<>(); - properties.put( - GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE, "/tmp/gravitino-missing-gcs-key.json"); + properties.put(GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE, "/tmp/gcs-key.json"); + properties.put(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN, "existing-token"); - UncheckedIOException thrown = - Assertions.assertThrows( - UncheckedIOException.class, - () -> IcebergCatalogUtil.applyGcsServiceAccountCredentials(properties)); - Assertions.assertTrue(thrown.getMessage().contains("does not exist")); - } + IcebergCatalogUtil.applyDefaultResolvingFileIO(properties); - @Test - void testWithGcsServiceAccountCredentialsReturnsSameConfigWhenNoServiceAccountFile() { - IcebergConfig config = new IcebergConfig(Map.of(IcebergConstants.CATALOG_BACKEND, "memory")); - Assertions.assertSame(config, IcebergCatalogUtil.withGcsServiceAccountCredentials(config)); + Assertions.assertEquals( + org.apache.iceberg.io.ResolvingFileIO.class.getName(), + properties.get(IcebergConstants.IO_IMPL)); + Assertions.assertEquals( + "existing-token", properties.get(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN)); } @Test - void testWithGcsServiceAccountCredentialsReturnsSameConfigWhenTokenAlreadyPresent() { + void testApplyGcsServiceAccountFileIONoOpWithoutServiceAccountFile() { Map 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); + properties.put(IcebergConstants.IO_IMPL, "org.apache.iceberg.gcp.gcs.GCSFileIO"); + + IcebergCatalogUtil.applyDefaultResolvingFileIO(properties); - Assertions.assertSame(config, IcebergCatalogUtil.withGcsServiceAccountCredentials(config)); + Assertions.assertEquals( + "org.apache.iceberg.gcp.gcs.GCSFileIO", properties.get(IcebergConstants.IO_IMPL)); } @Test - void testApplyDefaultResolvingFileIOInjectsGcsToken() { + void testApplyGcsServiceAccountFileIODoesNotOverrideCustomIOImpl() { Map properties = new HashMap<>(); - properties.put(IcebergConstants.WAREHOUSE, "gs://bucket/warehouse"); + properties.put(IcebergConstants.IO_IMPL, "org.apache.iceberg.aws.s3.S3FileIO"); 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)); + "org.apache.iceberg.aws.s3.S3FileIO", properties.get(IcebergConstants.IO_IMPL)); } @Test diff --git a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java index c58547e4fc4..c58b30b7700 100644 --- a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java +++ b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java @@ -42,6 +42,7 @@ import org.apache.gravitino.credential.CredentialPropertyUtils; import org.apache.gravitino.credential.PathBasedCredentialContext; import org.apache.gravitino.iceberg.common.IcebergConfig; +import org.apache.gravitino.iceberg.common.io.GravitinoGCSFileIO; import org.apache.gravitino.iceberg.common.ops.IcebergCatalogWrapper; import org.apache.gravitino.iceberg.service.cache.ScanPlanCache; import org.apache.gravitino.iceberg.service.cache.ScanPlanCacheKey; @@ -60,6 +61,7 @@ import org.apache.iceberg.catalog.Namespace; import org.apache.iceberg.catalog.TableIdentifier; import org.apache.iceberg.exceptions.ServiceUnavailableException; +import org.apache.iceberg.gcp.gcs.GCSFileIO; import org.apache.iceberg.io.CloseableIterable; import org.apache.iceberg.rest.CatalogHandlers; import org.apache.iceberg.rest.PlanStatus; @@ -258,6 +260,11 @@ protected static Map filterCatalogConfigForClients( new HashMap<>( MapUtils.getFilteredMap( sourceProps, key -> catalogPropertiesToClientKeys.contains(key))); + // GravitinoGCSFileIO is server-only; clients should use Iceberg's GCSFileIO (plus vended + // credentials when enabled). + if (GravitinoGCSFileIO.class.getName().equals(filtered.get(IcebergConstants.IO_IMPL))) { + filtered.put(IcebergConstants.IO_IMPL, GCSFileIO.class.getName()); + } validateAndNormalizeDataAccessProperty(filtered); return Collections.unmodifiableMap(filtered); } diff --git a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergCatalogWrapperManager.java b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergCatalogWrapperManager.java index 90d8a8f4fe7..5d8fbf52f1e 100644 --- a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergCatalogWrapperManager.java +++ b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergCatalogWrapperManager.java @@ -20,7 +20,6 @@ import com.github.benmanes.caffeine.cache.Cache; import com.github.benmanes.caffeine.cache.Caffeine; -import com.github.benmanes.caffeine.cache.Expiry; import com.github.benmanes.caffeine.cache.Scheduler; import com.google.common.annotations.VisibleForTesting; import com.google.common.util.concurrent.ThreadFactoryBuilder; @@ -29,7 +28,6 @@ import java.util.Optional; import java.util.concurrent.ScheduledThreadPoolExecutor; import java.util.concurrent.TimeUnit; -import org.apache.commons.lang3.StringUtils; import org.apache.gravitino.GravitinoEnv; import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergCatalogBackend; import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants; @@ -39,7 +37,6 @@ import org.apache.gravitino.iceberg.common.authentication.SupportsKerberos; import org.apache.gravitino.iceberg.common.ops.IcebergCatalogWrapper; import org.apache.gravitino.iceberg.common.ops.KerberosAwareIcebergCatalogProxy; -import org.apache.gravitino.iceberg.common.utils.IcebergCatalogUtil; import org.apache.gravitino.iceberg.service.authorization.IcebergRESTServerContext; import org.apache.gravitino.iceberg.service.provider.DynamicIcebergConfigProvider; import org.apache.gravitino.iceberg.service.provider.IcebergConfigProvider; @@ -50,12 +47,6 @@ public class IcebergCatalogWrapperManager implements AutoCloseable { public static final Logger LOG = LoggerFactory.getLogger(IcebergCatalogWrapperManager.class); - /** - * Evict a cached catalog this long before its minted GCS OAuth2 token expires, so the next - * request recreates the catalog and refreshes the token. - */ - @VisibleForTesting static final long GCS_TOKEN_REFRESH_BUFFER_MS = TimeUnit.MINUTES.toMillis(5); - private final Cache catalogWrapperCache; private final IcebergConfigProvider configProvider; @@ -66,13 +57,12 @@ public IcebergCatalogWrapperManager( boolean auxMode, String metalakeName) { this.configProvider = configProvider; - long accessEvictionNanos = - TimeUnit.MILLISECONDS.toNanos( - new IcebergConfig(properties) - .get(IcebergConfig.ICEBERG_REST_CATALOG_CACHE_EVICTION_INTERVAL)); this.catalogWrapperCache = Caffeine.newBuilder() - .expireAfter(new CatalogWrapperExpiry(accessEvictionNanos)) + .expireAfterAccess( + (new IcebergConfig(properties)) + .get(IcebergConfig.ICEBERG_REST_CATALOG_CACHE_EVICTION_INTERVAL), + TimeUnit.MILLISECONDS) .removalListener( (k, v, c) -> { String catalogName = (String) k; @@ -150,26 +140,22 @@ private CatalogWrapperForREST createCatalogWrapper(String catalogName) { @VisibleForTesting protected CatalogWrapperForREST createCatalogWrapper( String catalogName, IcebergConfig icebergConfig) { - // Mint GCS OAuth2 tokens into the config before constructing the wrapper so the IRC catalog - // cache can expire the entry before gcs.oauth2.token-expires-at. - IcebergConfig enrichedConfig = - IcebergCatalogUtil.withGcsServiceAccountCredentials(icebergConfig); // When the backend is a federated Iceberg REST catalog, use FederatedCatalogWrapper so // federation-aware behavior (FileIO property extraction, remote credential vending, remote // /v1/config defaults) is applied through polymorphic dispatch rather than scattered // instanceof checks. All other backends use the base CatalogWrapperForREST. IcebergCatalogBackend backend = IcebergCatalogBackend.valueOf( - enrichedConfig.get(IcebergConfig.CATALOG_BACKEND).toUpperCase(Locale.ROOT)); + icebergConfig.get(IcebergConfig.CATALOG_BACKEND).toUpperCase(Locale.ROOT)); CatalogWrapperForREST rest = backend == IcebergCatalogBackend.REST - ? new FederatedCatalogWrapper(catalogName, enrichedConfig) - : new CatalogWrapperForREST(catalogName, enrichedConfig); + ? new FederatedCatalogWrapper(catalogName, icebergConfig) + : new CatalogWrapperForREST(catalogName, icebergConfig); AuthenticationConfig authenticationConfig = - new AuthenticationConfig(enrichedConfig.getAllConfig()); + new AuthenticationConfig(icebergConfig.getAllConfig()); if (authenticationConfig.isKerberosAuth() && rest.getCatalog() instanceof SupportsKerberos) { return (CatalogWrapperForREST) - new KerberosAwareIcebergCatalogProxy(rest).getProxy(catalogName, enrichedConfig); + new KerberosAwareIcebergCatalogProxy(rest).getProxy(catalogName, icebergConfig); } return rest; @@ -183,73 +169,8 @@ private void closeIcebergCatalogWrapper(IcebergCatalogWrapper catalogWrapper) { } } - /** - * Computes how long a catalog wrapper may stay in the IRC cache. - * - *

Uses the configured access-based eviction interval, capped by the time until a minted GCS - * OAuth2 token should be refreshed ({@code gcs.oauth2.token-expires-at} minus {@link - * #GCS_TOKEN_REFRESH_BUFFER_MS}). When no token expiry is present, returns {@code - * accessEvictionNanos}. - * - * @param config catalog config that may contain {@code gcs.oauth2.token-expires-at} - * @param accessEvictionNanos default expire-after-access duration in nanoseconds - * @param nowEpochMillis current wall-clock time - * @return cache duration in nanoseconds; {@code 0} means expire immediately - */ - @VisibleForTesting - static long computeCacheDurationNanos( - IcebergConfig config, long accessEvictionNanos, long nowEpochMillis) { - String expiresAt = - config.getAllConfig().get(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN_EXPIRES_AT); - if (StringUtils.isBlank(expiresAt)) { - return accessEvictionNanos; - } - - long expiresAtMs; - try { - expiresAtMs = Long.parseLong(expiresAt); - } catch (NumberFormatException e) { - LOG.warn("Invalid {}: {}", IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN_EXPIRES_AT, expiresAt); - return accessEvictionNanos; - } - - long remainingMs = expiresAtMs - GCS_TOKEN_REFRESH_BUFFER_MS - nowEpochMillis; - if (remainingMs <= 0) { - return 0L; - } - return Math.min(accessEvictionNanos, TimeUnit.MILLISECONDS.toNanos(remainingMs)); - } - @Override public void close() throws Exception { catalogWrapperCache.invalidateAll(); } - - private static final class CatalogWrapperExpiry implements Expiry { - - private final long accessEvictionNanos; - - CatalogWrapperExpiry(long accessEvictionNanos) { - this.accessEvictionNanos = accessEvictionNanos; - } - - @Override - public long expireAfterCreate(String key, CatalogWrapperForREST value, long currentTime) { - return computeCacheDurationNanos( - value.getIcebergConfig(), accessEvictionNanos, System.currentTimeMillis()); - } - - @Override - public long expireAfterUpdate( - String key, CatalogWrapperForREST value, long currentTime, long currentDuration) { - return expireAfterCreate(key, value, currentTime); - } - - @Override - public long expireAfterRead( - String key, CatalogWrapperForREST value, long currentTime, long currentDuration) { - // Preserve expire-after-access, but never extend past the GCS token refresh deadline. - return expireAfterCreate(key, value, currentTime); - } - } } diff --git a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergCatalogWrapperManagerForREST.java b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergCatalogWrapperManagerForREST.java index 3a5e8925e12..45eb62f69ee 100644 --- a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergCatalogWrapperManagerForREST.java +++ b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergCatalogWrapperManagerForREST.java @@ -22,7 +22,6 @@ import com.google.common.collect.Maps; import java.util.Map; import java.util.Optional; -import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; import java.util.function.Consumer; import org.apache.commons.lang3.StringUtils; @@ -190,70 +189,6 @@ public void testDefaultCatalogAliasInvalidatedWhenCatalogRemoved() throws Except } } - @Test - public void testComputeCacheDurationNanosWithoutTokenExpiryUsesAccessEviction() { - IcebergConfig config = - new IcebergConfig(ImmutableMap.of(IcebergConstants.CATALOG_BACKEND, "memory")); - long accessEvictionNanos = TimeUnit.HOURS.toNanos(1); - Assertions.assertEquals( - accessEvictionNanos, - IcebergCatalogWrapperManager.computeCacheDurationNanos( - config, accessEvictionNanos, System.currentTimeMillis())); - } - - @Test - public void testComputeCacheDurationNanosCapsByGcsTokenExpiry() { - long now = 1_700_000_000_000L; - long expiresAt = now + TimeUnit.HOURS.toMillis(1); // token valid for 1h - IcebergConfig config = - new IcebergConfig( - ImmutableMap.of( - IcebergConstants.CATALOG_BACKEND, - "memory", - IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN_EXPIRES_AT, - String.valueOf(expiresAt))); - long accessEvictionNanos = TimeUnit.HOURS.toNanos(2); - long expected = - TimeUnit.MILLISECONDS.toNanos( - TimeUnit.HOURS.toMillis(1) - IcebergCatalogWrapperManager.GCS_TOKEN_REFRESH_BUFFER_MS); - Assertions.assertEquals( - expected, - IcebergCatalogWrapperManager.computeCacheDurationNanos(config, accessEvictionNanos, now)); - } - - @Test - public void testComputeCacheDurationNanosExpiresImmediatelyWhenPastRefreshDeadline() { - long now = 1_700_000_000_000L; - long expiresAt = now + TimeUnit.MINUTES.toMillis(2); // within 5-minute buffer - IcebergConfig config = - new IcebergConfig( - ImmutableMap.of( - IcebergConstants.CATALOG_BACKEND, - "memory", - IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN_EXPIRES_AT, - String.valueOf(expiresAt))); - Assertions.assertEquals( - 0L, - IcebergCatalogWrapperManager.computeCacheDurationNanos( - config, TimeUnit.HOURS.toNanos(1), now)); - } - - @Test - public void testComputeCacheDurationNanosIgnoresInvalidExpiresAt() { - IcebergConfig config = - new IcebergConfig( - ImmutableMap.of( - IcebergConstants.CATALOG_BACKEND, - "memory", - IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN_EXPIRES_AT, - "not-a-number")); - long accessEvictionNanos = TimeUnit.MINUTES.toNanos(30); - Assertions.assertEquals( - accessEvictionNanos, - IcebergCatalogWrapperManager.computeCacheDurationNanos( - config, accessEvictionNanos, System.currentTimeMillis())); - } - private static IcebergCatalogWrapperManager newManager() { Map config = Maps.newHashMap(); IcebergConfigProvider configProvider = IcebergConfigProviderFactory.create(config); From 1d2ca6256a3564b756c464512baf382923cad93f Mon Sep 17 00:00:00 2001 From: remilu <1175334135@qq.com> Date: Mon, 7 Sep 2026 20:38:44 +0800 Subject: [PATCH 06/11] [#9418] fix(iceberg): restore IOException import in IcebergCatalogUtil Co-authored-by: Cursor --- .../gravitino/iceberg/common/utils/IcebergCatalogUtil.java | 1 + 1 file changed, 1 insertion(+) diff --git a/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/utils/IcebergCatalogUtil.java b/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/utils/IcebergCatalogUtil.java index 87c781b2789..3fa29c784c2 100644 --- a/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/utils/IcebergCatalogUtil.java +++ b/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/utils/IcebergCatalogUtil.java @@ -23,6 +23,7 @@ import com.google.common.annotations.VisibleForTesting; import com.google.common.collect.Maps; +import java.io.IOException; import java.sql.SQLException; import java.util.Collections; import java.util.HashMap; From 9aae30abfe1728bd145af413a7f38ebda8f223bf Mon Sep 17 00:00:00 2001 From: remilu <1175334135@qq.com> Date: Mon, 7 Sep 2026 20:48:35 +0800 Subject: [PATCH 07/11] [#9418] fix(iceberg): avoid iceberg-gcp compile dep in rest-server Rewrite GravitinoGCSFileIO to the Iceberg GCSFileIO class name via a string constant so iceberg-rest-server does not need iceberg-gcp at compile time. Co-authored-by: Cursor --- .../iceberg/service/CatalogWrapperForREST.java | 6 ++++-- .../iceberg/service/TestCatalogWrapperForREST.java | 11 +++++++++++ 2 files changed, 15 insertions(+), 2 deletions(-) diff --git a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java index c58b30b7700..9e18143a857 100644 --- a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java +++ b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java @@ -61,7 +61,6 @@ import org.apache.iceberg.catalog.Namespace; import org.apache.iceberg.catalog.TableIdentifier; import org.apache.iceberg.exceptions.ServiceUnavailableException; -import org.apache.iceberg.gcp.gcs.GCSFileIO; import org.apache.iceberg.io.CloseableIterable; import org.apache.iceberg.rest.CatalogHandlers; import org.apache.iceberg.rest.PlanStatus; @@ -87,6 +86,9 @@ public class CatalogWrapperForREST extends IcebergCatalogWrapper { private static final String DATA_ACCESS_VENDED_CREDENTIALS = "vended-credentials"; private static final String DATA_ACCESS_REMOTE_SIGNING = "remote-signing"; + /** Iceberg built-in GCS FileIO; used when rewriting server-only {@link GravitinoGCSFileIO}. */ + private static final String ICEBERG_GCS_FILE_IO = "org.apache.iceberg.gcp.gcs.GCSFileIO"; + /** * Client-facing catalog property keys retained when building the IRC {@code /v1/config} defaults * and when extracting FileIO-derived config in {@link FederatedCatalogWrapper}. @@ -263,7 +265,7 @@ protected static Map filterCatalogConfigForClients( // GravitinoGCSFileIO is server-only; clients should use Iceberg's GCSFileIO (plus vended // credentials when enabled). if (GravitinoGCSFileIO.class.getName().equals(filtered.get(IcebergConstants.IO_IMPL))) { - filtered.put(IcebergConstants.IO_IMPL, GCSFileIO.class.getName()); + filtered.put(IcebergConstants.IO_IMPL, ICEBERG_GCS_FILE_IO); } validateAndNormalizeDataAccessProperty(filtered); return Collections.unmodifiableMap(filtered); diff --git a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestCatalogWrapperForREST.java b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestCatalogWrapperForREST.java index 50a3f684d16..9ea4b1b67e7 100644 --- a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestCatalogWrapperForREST.java +++ b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestCatalogWrapperForREST.java @@ -907,6 +907,17 @@ void testCatalogConfigToClientsIncludesResolvingFileIO() { ResolvingFileIO.class.getName(), configToClients.get(IcebergConstants.IO_IMPL)); } + @Test + void testCatalogConfigRewritesGravitinoGcsFileIOForClients() { + Map configToClients = + CatalogWrapperForREST.filterCatalogConfigForClients( + ImmutableMap.of( + IcebergConstants.IO_IMPL, + org.apache.gravitino.iceberg.common.io.GravitinoGCSFileIO.class.getName())); + Assertions.assertEquals( + "org.apache.iceberg.gcp.gcs.GCSFileIO", configToClients.get(IcebergConstants.IO_IMPL)); + } + @Test void testNonRestCatalogClientConfig() { Map configToClients = From 51a2ea8ce452c986ded741f77d3f143d3be184a7 Mon Sep 17 00:00:00 2001 From: remilu <1175334135@qq.com> Date: Mon, 7 Sep 2026 20:53:19 +0800 Subject: [PATCH 08/11] [#9418] fix(iceberg): suppress JavaUtilDate in GCS token cache tests Co-authored-by: Cursor --- .../gravitino/iceberg/common/io/TestGcsAccessTokenCache.java | 1 + 1 file changed, 1 insertion(+) diff --git a/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/io/TestGcsAccessTokenCache.java b/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/io/TestGcsAccessTokenCache.java index 0c926ce2545..13c696b687b 100644 --- a/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/io/TestGcsAccessTokenCache.java +++ b/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/io/TestGcsAccessTokenCache.java @@ -26,6 +26,7 @@ import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; +@SuppressWarnings("JavaUtilDate") // AccessToken API uses java.util.Date public class TestGcsAccessTokenCache { @AfterEach From 8b93fe0b224c7ba5c2c0ba7ae9b5d6ca0af628f3 Mon Sep 17 00:00:00 2001 From: remilu <1175334135@qq.com> Date: Tue, 8 Sep 2026 20:21:57 +0800 Subject: [PATCH 09/11] [#9418] fix(iceberg): refresh GCS tokens via IRC catalog cache expiry Mint gcs.oauth2.token from gcs-service-account-file and expire IRC catalog wrappers before token expiry, instead of a dedicated GravitinoGCSFileIO cache. Co-authored-by: Cursor --- docs/iceberg-rest-service.md | 2 +- docs/lakehouse-iceberg-catalog.md | 2 +- docs/security/credential-vending.md | 2 +- .../common/io/GcsAccessTokenCache.java | 131 ------------- .../iceberg/common/io/GravitinoGCSFileIO.java | 180 ------------------ .../common/utils/IcebergCatalogUtil.java | 93 +++++++-- .../common/io/TestGcsAccessTokenCache.java | 62 ------ .../common/utils/TestIcebergCatalogUtil.java | 64 ++++--- .../service/CatalogWrapperForREST.java | 9 - .../service/IcebergCatalogWrapperManager.java | 104 ++++++++-- .../service/TestCatalogWrapperForREST.java | 11 -- ...stIcebergCatalogWrapperManagerForREST.java | 65 +++++++ 12 files changed, 266 insertions(+), 459 deletions(-) delete mode 100644 iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/io/GcsAccessTokenCache.java delete mode 100644 iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/io/GravitinoGCSFileIO.java delete mode 100644 iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/io/TestGcsAccessTokenCache.java diff --git a/docs/iceberg-rest-service.md b/docs/iceberg-rest-service.md index d6809bbd4de..f3c55cdafde 100644 --- a/docs/iceberg-rest-service.md +++ b/docs/iceberg-rest-service.md @@ -587,7 +587,7 @@ For other Iceberg GCS properties not managed by Gravitino like `gcs.project-id`, Refer to [GCS credentials](./security/credential-vending.md#gcs-credentials) for credential related configurations. :::note -When `gcs-service-account-file` is set, Gravitino uses `GravitinoGCSFileIO`, which remints OAuth2 tokens from that file through a dedicated token cache. Token refresh is independent of the IRC catalog cache. If unset, use Application Default Credentials (for example GCE metadata or `GOOGLE_APPLICATION_CREDENTIALS`). +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 diff --git a/docs/lakehouse-iceberg-catalog.md b/docs/lakehouse-iceberg-catalog.md index 26e6b970e4a..6f4741f15fc 100644 --- a/docs/lakehouse-iceberg-catalog.md +++ b/docs/lakehouse-iceberg-catalog.md @@ -182,7 +182,7 @@ Supports using google credential file to access GCS data. For other Iceberg GCS properties not managed by Gravitino like `gcs.project-id`, you could config it directly by `gravitino.bypass.gcs.project-id`. -When `gcs-service-account-file` is set, Gravitino uses `GravitinoGCSFileIO`, which remints OAuth2 tokens from that file through a dedicated token cache (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`. +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/`. diff --git a/docs/security/credential-vending.md b/docs/security/credential-vending.md index 6ae5c2125bd..7b1aeb9a010 100755 --- a/docs/security/credential-vending.md +++ b/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 | -`gcs-service-account-file` is used both to vend downscoped tokens and to authenticate Iceberg FileIO on the server (via `GravitinoGCSFileIO` and a dedicated token cache, 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`). +`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 diff --git a/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/io/GcsAccessTokenCache.java b/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/io/GcsAccessTokenCache.java deleted file mode 100644 index 8b29828b0ff..00000000000 --- a/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/io/GcsAccessTokenCache.java +++ /dev/null @@ -1,131 +0,0 @@ -/* - * 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.gravitino.iceberg.common.io; - -import com.github.benmanes.caffeine.cache.Cache; -import com.github.benmanes.caffeine.cache.Caffeine; -import com.github.benmanes.caffeine.cache.Expiry; -import com.google.auth.oauth2.AccessToken; -import com.google.auth.oauth2.GoogleCredentials; -import com.google.common.annotations.VisibleForTesting; -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.util.concurrent.TimeUnit; - -/** - * Process-wide cache of GCS OAuth2 access tokens minted from service-account JSON files. - * - *

Token lifetime is independent of the IRC catalog-wrapper cache: entries expire shortly before - * the underlying Google access token expires, and the next lookup remints a fresh token from the - * service-account file. - */ -public final class GcsAccessTokenCache { - - private static final String GCS_CLOUD_PLATFORM_SCOPE = - "https://www.googleapis.com/auth/cloud-platform"; - - /** Evict this long before the Google access token expires so reminting happens early. */ - @VisibleForTesting static final long TOKEN_REFRESH_BUFFER_MS = TimeUnit.MINUTES.toMillis(5); - - private static final Cache CACHE = - Caffeine.newBuilder().expireAfter(new AccessTokenExpiry()).build(); - - private GcsAccessTokenCache() {} - - /** - * Returns a cached access token for {@code serviceAccountFile}, minting one on cache miss or - * after expiry. - * - * @param serviceAccountFile path to the GCS service-account JSON file - * @return a non-null access token - */ - public static AccessToken get(String serviceAccountFile) { - return CACHE.get(serviceAccountFile, GcsAccessTokenCache::loadAccessToken); - } - - @VisibleForTesting - static void invalidate(String serviceAccountFile) { - CACHE.invalidate(serviceAccountFile); - } - - @VisibleForTesting - static void invalidateAll() { - CACHE.invalidateAll(); - } - - @VisibleForTesting - static long durationUntilRefreshNanos(AccessToken accessToken, long nowEpochMillis) { - if (accessToken.getExpirationTime() == null) { - // No expiry from Google — keep the entry briefly so a bad token cannot live forever. - return TimeUnit.MINUTES.toNanos(30); - } - long remainingMs = - accessToken.getExpirationTime().toInstant().toEpochMilli() - - TOKEN_REFRESH_BUFFER_MS - - nowEpochMillis; - return remainingMs <= 0 ? 0L : TimeUnit.MILLISECONDS.toNanos(remainingMs); - } - - private static AccessToken loadAccessToken(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(); - AccessToken accessToken = credentials.getAccessToken(); - if (accessToken == null || accessToken.getTokenValue() == null) { - throw new IllegalStateException( - "Failed to obtain GCS access token from service account file: " + serviceAccountFile); - } - return accessToken; - } 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); - } - } - - private static final class AccessTokenExpiry implements Expiry { - - @Override - public long expireAfterCreate(String key, AccessToken value, long currentTime) { - return durationUntilRefreshNanos(value, System.currentTimeMillis()); - } - - @Override - public long expireAfterUpdate( - String key, AccessToken value, long currentTime, long currentDuration) { - return expireAfterCreate(key, value, currentTime); - } - - @Override - public long expireAfterRead( - String key, AccessToken value, long currentTime, long currentDuration) { - // Recalculate from wall clock so access does not stretch past token expiry. - return expireAfterCreate(key, value, currentTime); - } - } -} diff --git a/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/io/GravitinoGCSFileIO.java b/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/io/GravitinoGCSFileIO.java deleted file mode 100644 index ffaca8f93aa..00000000000 --- a/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/io/GravitinoGCSFileIO.java +++ /dev/null @@ -1,180 +0,0 @@ -/* - * 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.gravitino.iceberg.common.io; - -import com.google.auth.oauth2.AccessToken; -import java.util.Date; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import org.apache.commons.lang3.StringUtils; -import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants; -import org.apache.gravitino.storage.GCSProperties; -import org.apache.iceberg.gcp.gcs.GCSFileIO; -import org.apache.iceberg.io.BulkDeletionFailureException; -import org.apache.iceberg.io.DelegateFileIO; -import org.apache.iceberg.io.FileInfo; -import org.apache.iceberg.io.InputFile; -import org.apache.iceberg.io.OutputFile; -import org.apache.iceberg.io.StorageCredential; -import org.apache.iceberg.io.SupportsStorageCredentials; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -/** - * GCS {@link DelegateFileIO} that authenticates with {@code gcs-service-account-file}. - * - *

Iceberg's built-in {@link GCSFileIO} only understands a static {@code gcs.oauth2.token}. This - * wrapper keeps a delegate {@link GCSFileIO} in sync with {@link GcsAccessTokenCache}, recreating - * the delegate when the cached token is reminted. Token refresh does not depend on IRC catalog - * cache eviction. - */ -public class GravitinoGCSFileIO implements DelegateFileIO, SupportsStorageCredentials { - - private static final Logger LOG = LoggerFactory.getLogger(GravitinoGCSFileIO.class); - - private Map properties; - private String serviceAccountFile; - private volatile GCSFileIO delegate; - private volatile String boundToken; - - @Override - public void initialize(Map props) { - this.properties = new HashMap<>(props); - this.serviceAccountFile = props.get(GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE); - // Drop any stale delegate after re-initialize. - closeDelegate(); - currentDelegate(); - } - - @Override - public InputFile newInputFile(String path) { - return currentDelegate().newInputFile(path); - } - - @Override - public InputFile newInputFile(String path, long length) { - return currentDelegate().newInputFile(path, length); - } - - @Override - public OutputFile newOutputFile(String path) { - return currentDelegate().newOutputFile(path); - } - - @Override - public void deleteFile(String path) { - currentDelegate().deleteFile(path); - } - - @Override - public void deleteFiles(Iterable pathsToDelete) throws BulkDeletionFailureException { - currentDelegate().deleteFiles(pathsToDelete); - } - - @Override - public Iterable listPrefix(String prefix) { - return currentDelegate().listPrefix(prefix); - } - - @Override - public void deletePrefix(String prefix) { - currentDelegate().deletePrefix(prefix); - } - - @Override - public Map properties() { - return currentDelegate().properties(); - } - - @Override - public void setCredentials(List credentials) { - currentDelegate().setCredentials(credentials); - } - - @Override - public List credentials() { - return currentDelegate().credentials(); - } - - @Override - public void close() { - closeDelegate(); - } - - private GCSFileIO currentDelegate() { - if (StringUtils.isBlank(serviceAccountFile)) { - return ensureDelegate(null); - } - AccessToken accessToken = GcsAccessTokenCache.get(serviceAccountFile); - return ensureDelegate(accessToken); - } - - private GCSFileIO ensureDelegate(AccessToken accessToken) { - String tokenValue = accessToken == null ? null : accessToken.getTokenValue(); - GCSFileIO current = delegate; - if (current != null && java.util.Objects.equals(boundToken, tokenValue)) { - return current; - } - synchronized (this) { - current = delegate; - if (current != null && java.util.Objects.equals(boundToken, tokenValue)) { - return current; - } - closeDelegate(); - Map delegateProperties = new HashMap<>(properties); - if (accessToken != null) { - delegateProperties.put( - IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN, accessToken.getTokenValue()); - Date expirationTime = accessToken.getExpirationTime(); - if (expirationTime != null) { - delegateProperties.put( - IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN_EXPIRES_AT, - String.valueOf(expirationTime.toInstant().toEpochMilli())); - } - // Catalog bootstrap from a service-account file has no table credentials endpoint. - delegateProperties.put( - IcebergConstants.ICEBERG_GCS_OAUTH2_REFRESH_CREDENTIALS_ENABLED, "false"); - LOG.debug( - "Binding GCSFileIO delegate to token from {}", - GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE); - } - GCSFileIO created = new GCSFileIO(); - created.initialize(delegateProperties); - this.delegate = created; - this.boundToken = tokenValue; - return created; - } - } - - private void closeDelegate() { - GCSFileIO current = delegate; - if (current == null) { - return; - } - try { - current.close(); - } catch (RuntimeException e) { - LOG.warn("Failed to close GCSFileIO delegate", e); - } finally { - this.delegate = null; - this.boundToken = null; - } - } -} diff --git a/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/utils/IcebergCatalogUtil.java b/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/utils/IcebergCatalogUtil.java index 3fa29c784c2..658786ea15b 100644 --- a/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/utils/IcebergCatalogUtil.java +++ b/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/utils/IcebergCatalogUtil.java @@ -21,11 +21,20 @@ 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; @@ -38,14 +47,12 @@ 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.iceberg.common.io.GravitinoGCSFileIO; 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; import org.apache.iceberg.catalog.Catalog; -import org.apache.iceberg.gcp.gcs.GCSFileIO; import org.apache.iceberg.hive.HiveCatalog; import org.apache.iceberg.hive.HiveCatalogWithMetadataLocationSupport; import org.apache.iceberg.inmemory.InMemoryCatalog; @@ -69,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 MEMORY_CATALOGS = new ConcurrentHashMap<>(); @@ -273,23 +283,24 @@ private static Catalog loadCustomCatalog(IcebergConfig icebergConfig) { @VisibleForTesting public static void applyDefaultResolvingFileIO(Map properties) { properties.putIfAbsent(IcebergConstants.IO_IMPL, ResolvingFileIO.class.getName()); - applyGcsServiceAccountFileIO(properties); + applyGcsServiceAccountCredentials(properties); } /** - * When {@code gcs-service-account-file} is set and no explicit {@code gcs.oauth2.token} is - * present, switch the FileIO to {@link GravitinoGCSFileIO}. That FileIO remints tokens through - * {@link org.apache.gravitino.iceberg.common.io.GcsAccessTokenCache}, independently of the IRC - * catalog-wrapper cache. Iceberg's built-in {@link GCSFileIO} / {@link ResolvingFileIO} path does - * not understand Gravitino's service-account-file property. + * 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}. * - *

Skips when {@code gcs.oauth2.token} is already present, or when the user set a custom {@code - * io-impl} other than {@link ResolvingFileIO} / {@link GCSFileIO}. + *

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 applyGcsServiceAccountFileIO(Map properties) { + static void applyGcsServiceAccountCredentials(Map properties) { String serviceAccountFile = properties.get(GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE); if (StringUtils.isBlank(serviceAccountFile)) { return; @@ -298,15 +309,57 @@ static void applyGcsServiceAccountFileIO(Map properties) { return; } - String ioImpl = properties.get(IcebergConstants.IO_IMPL); - if (ioImpl == null - || ResolvingFileIO.class.getName().equals(ioImpl) - || GCSFileIO.class.getName().equals(ioImpl)) { - properties.put(IcebergConstants.IO_IMPL, GravitinoGCSFileIO.class.getName()); - LOG.info( - "Using {} for catalog FileIO because {} is set", - GravitinoGCSFileIO.class.getName(), - GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE); + 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 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); } } diff --git a/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/io/TestGcsAccessTokenCache.java b/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/io/TestGcsAccessTokenCache.java deleted file mode 100644 index 13c696b687b..00000000000 --- a/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/io/TestGcsAccessTokenCache.java +++ /dev/null @@ -1,62 +0,0 @@ -/* - * 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.gravitino.iceberg.common.io; - -import com.google.auth.oauth2.AccessToken; -import java.io.UncheckedIOException; -import java.util.Date; -import java.util.concurrent.TimeUnit; -import org.junit.jupiter.api.AfterEach; -import org.junit.jupiter.api.Assertions; -import org.junit.jupiter.api.Test; - -@SuppressWarnings("JavaUtilDate") // AccessToken API uses java.util.Date -public class TestGcsAccessTokenCache { - - @AfterEach - void tearDown() { - GcsAccessTokenCache.invalidateAll(); - } - - @Test - void testDurationUntilRefreshNanosUsesTokenExpiryMinusBuffer() { - long now = 1_700_000_000_000L; - AccessToken token = new AccessToken("token", new Date(now + TimeUnit.HOURS.toMillis(1))); - long expected = - TimeUnit.MILLISECONDS.toNanos( - TimeUnit.HOURS.toMillis(1) - GcsAccessTokenCache.TOKEN_REFRESH_BUFFER_MS); - Assertions.assertEquals(expected, GcsAccessTokenCache.durationUntilRefreshNanos(token, now)); - } - - @Test - void testDurationUntilRefreshNanosExpiresImmediatelyNearExpiry() { - long now = 1_700_000_000_000L; - AccessToken token = new AccessToken("token", new Date(now + TimeUnit.MINUTES.toMillis(2))); - Assertions.assertEquals(0L, GcsAccessTokenCache.durationUntilRefreshNanos(token, now)); - } - - @Test - void testGetFailsWhenServiceAccountFileMissing() { - UncheckedIOException thrown = - Assertions.assertThrows( - UncheckedIOException.class, - () -> GcsAccessTokenCache.get("/tmp/gravitino-missing-gcs-key.json")); - Assertions.assertTrue(thrown.getMessage().contains("does not exist")); - } -} diff --git a/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/utils/TestIcebergCatalogUtil.java b/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/utils/TestIcebergCatalogUtil.java index 3fad7d002ad..d8ce5cc18a4 100644 --- a/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/utils/TestIcebergCatalogUtil.java +++ b/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; @@ -339,68 +340,71 @@ void testApplyDefaultResolvingFileIODoesNotOverrideExplicitIOImpl() { } @Test - void testApplyGcsServiceAccountFileIOSwitchesFromResolvingFileIO() { + void testApplyGcsServiceAccountCredentialsSkipsWhenTokenAlreadyPresent() { Map properties = new HashMap<>(); - properties.put(IcebergConstants.WAREHOUSE, "gs://bucket/warehouse"); properties.put(GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE, "/tmp/gcs-key.json"); + properties.put(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN, "existing-token"); - IcebergCatalogUtil.applyDefaultResolvingFileIO(properties); + IcebergCatalogUtil.applyGcsServiceAccountCredentials(properties); Assertions.assertEquals( - org.apache.gravitino.iceberg.common.io.GravitinoGCSFileIO.class.getName(), - properties.get(IcebergConstants.IO_IMPL)); - Assertions.assertNull(properties.get(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN)); + "existing-token", properties.get(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN)); + Assertions.assertNull(properties.get(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN_EXPIRES_AT)); } @Test - void testApplyGcsServiceAccountFileIOSwitchesFromGcsFileIO() { + void testApplyGcsServiceAccountCredentialsNoOpWithoutServiceAccountFile() { Map properties = new HashMap<>(); properties.put(IcebergConstants.IO_IMPL, "org.apache.iceberg.gcp.gcs.GCSFileIO"); - properties.put(GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE, "/tmp/gcs-key.json"); - IcebergCatalogUtil.applyDefaultResolvingFileIO(properties); + IcebergCatalogUtil.applyGcsServiceAccountCredentials(properties); - Assertions.assertEquals( - org.apache.gravitino.iceberg.common.io.GravitinoGCSFileIO.class.getName(), - properties.get(IcebergConstants.IO_IMPL)); + Assertions.assertNull(properties.get(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN)); } @Test - void testApplyGcsServiceAccountFileIOSkipsWhenTokenAlreadyPresent() { + void testApplyGcsServiceAccountCredentialsFailsWhenFileMissing() { Map properties = new HashMap<>(); - properties.put(GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE, "/tmp/gcs-key.json"); - properties.put(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN, "existing-token"); + properties.put( + GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE, "/tmp/gravitino-missing-gcs-key.json"); - IcebergCatalogUtil.applyDefaultResolvingFileIO(properties); + UncheckedIOException thrown = + Assertions.assertThrows( + UncheckedIOException.class, + () -> IcebergCatalogUtil.applyGcsServiceAccountCredentials(properties)); + Assertions.assertTrue(thrown.getMessage().contains("does not exist")); + } - Assertions.assertEquals( - org.apache.iceberg.io.ResolvingFileIO.class.getName(), - properties.get(IcebergConstants.IO_IMPL)); - Assertions.assertEquals( - "existing-token", properties.get(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN)); + @Test + void testWithGcsServiceAccountCredentialsReturnsSameConfigWhenNoServiceAccountFile() { + IcebergConfig config = new IcebergConfig(Map.of(IcebergConstants.CATALOG_BACKEND, "memory")); + Assertions.assertSame(config, IcebergCatalogUtil.withGcsServiceAccountCredentials(config)); } @Test - void testApplyGcsServiceAccountFileIONoOpWithoutServiceAccountFile() { + void testWithGcsServiceAccountCredentialsReturnsSameConfigWhenTokenAlreadyPresent() { Map properties = new HashMap<>(); - properties.put(IcebergConstants.IO_IMPL, "org.apache.iceberg.gcp.gcs.GCSFileIO"); - - IcebergCatalogUtil.applyDefaultResolvingFileIO(properties); + 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.assertEquals( - "org.apache.iceberg.gcp.gcs.GCSFileIO", properties.get(IcebergConstants.IO_IMPL)); + Assertions.assertSame(config, IcebergCatalogUtil.withGcsServiceAccountCredentials(config)); } @Test - void testApplyGcsServiceAccountFileIODoesNotOverrideCustomIOImpl() { + void testApplyDefaultResolvingFileIOInjectsGcsToken() { Map properties = new HashMap<>(); - properties.put(IcebergConstants.IO_IMPL, "org.apache.iceberg.aws.s3.S3FileIO"); + 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.aws.s3.S3FileIO", properties.get(IcebergConstants.IO_IMPL)); + org.apache.iceberg.io.ResolvingFileIO.class.getName(), + properties.get(IcebergConstants.IO_IMPL)); + Assertions.assertEquals("pre-set", properties.get(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN)); } @Test diff --git a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java index 9e18143a857..c58547e4fc4 100644 --- a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java +++ b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java @@ -42,7 +42,6 @@ import org.apache.gravitino.credential.CredentialPropertyUtils; import org.apache.gravitino.credential.PathBasedCredentialContext; import org.apache.gravitino.iceberg.common.IcebergConfig; -import org.apache.gravitino.iceberg.common.io.GravitinoGCSFileIO; import org.apache.gravitino.iceberg.common.ops.IcebergCatalogWrapper; import org.apache.gravitino.iceberg.service.cache.ScanPlanCache; import org.apache.gravitino.iceberg.service.cache.ScanPlanCacheKey; @@ -86,9 +85,6 @@ public class CatalogWrapperForREST extends IcebergCatalogWrapper { private static final String DATA_ACCESS_VENDED_CREDENTIALS = "vended-credentials"; private static final String DATA_ACCESS_REMOTE_SIGNING = "remote-signing"; - /** Iceberg built-in GCS FileIO; used when rewriting server-only {@link GravitinoGCSFileIO}. */ - private static final String ICEBERG_GCS_FILE_IO = "org.apache.iceberg.gcp.gcs.GCSFileIO"; - /** * Client-facing catalog property keys retained when building the IRC {@code /v1/config} defaults * and when extracting FileIO-derived config in {@link FederatedCatalogWrapper}. @@ -262,11 +258,6 @@ protected static Map filterCatalogConfigForClients( new HashMap<>( MapUtils.getFilteredMap( sourceProps, key -> catalogPropertiesToClientKeys.contains(key))); - // GravitinoGCSFileIO is server-only; clients should use Iceberg's GCSFileIO (plus vended - // credentials when enabled). - if (GravitinoGCSFileIO.class.getName().equals(filtered.get(IcebergConstants.IO_IMPL))) { - filtered.put(IcebergConstants.IO_IMPL, ICEBERG_GCS_FILE_IO); - } validateAndNormalizeDataAccessProperty(filtered); return Collections.unmodifiableMap(filtered); } diff --git a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergCatalogWrapperManager.java b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergCatalogWrapperManager.java index 5d8fbf52f1e..a960f175e0d 100644 --- a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergCatalogWrapperManager.java +++ b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergCatalogWrapperManager.java @@ -20,6 +20,7 @@ import com.github.benmanes.caffeine.cache.Cache; import com.github.benmanes.caffeine.cache.Caffeine; +import com.github.benmanes.caffeine.cache.Expiry; import com.github.benmanes.caffeine.cache.Scheduler; import com.google.common.annotations.VisibleForTesting; import com.google.common.util.concurrent.ThreadFactoryBuilder; @@ -28,6 +29,7 @@ import java.util.Optional; import java.util.concurrent.ScheduledThreadPoolExecutor; import java.util.concurrent.TimeUnit; +import org.apache.commons.lang3.StringUtils; import org.apache.gravitino.GravitinoEnv; import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergCatalogBackend; import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants; @@ -37,6 +39,7 @@ import org.apache.gravitino.iceberg.common.authentication.SupportsKerberos; import org.apache.gravitino.iceberg.common.ops.IcebergCatalogWrapper; import org.apache.gravitino.iceberg.common.ops.KerberosAwareIcebergCatalogProxy; +import org.apache.gravitino.iceberg.common.utils.IcebergCatalogUtil; import org.apache.gravitino.iceberg.service.authorization.IcebergRESTServerContext; import org.apache.gravitino.iceberg.service.provider.DynamicIcebergConfigProvider; import org.apache.gravitino.iceberg.service.provider.IcebergConfigProvider; @@ -47,6 +50,12 @@ public class IcebergCatalogWrapperManager implements AutoCloseable { public static final Logger LOG = LoggerFactory.getLogger(IcebergCatalogWrapperManager.class); + /** + * Evict a cached catalog this long before its minted GCS OAuth2 token expires, so the next + * request recreates the catalog and refreshes the token. + */ + @VisibleForTesting static final long GCS_TOKEN_REFRESH_BUFFER_MS = TimeUnit.MINUTES.toMillis(5); + private final Cache catalogWrapperCache; private final IcebergConfigProvider configProvider; @@ -57,20 +66,20 @@ public IcebergCatalogWrapperManager( boolean auxMode, String metalakeName) { this.configProvider = configProvider; + long accessEvictionNanos = + TimeUnit.MILLISECONDS.toNanos( + new IcebergConfig(properties) + .get(IcebergConfig.ICEBERG_REST_CATALOG_CACHE_EVICTION_INTERVAL)); this.catalogWrapperCache = Caffeine.newBuilder() - .expireAfterAccess( - (new IcebergConfig(properties)) - .get(IcebergConfig.ICEBERG_REST_CATALOG_CACHE_EVICTION_INTERVAL), - TimeUnit.MILLISECONDS) + .expireAfter(new CatalogWrapperExpiry(accessEvictionNanos)) .removalListener( - (k, v, c) -> { - String catalogName = (String) k; + (catalogName, catalogWrapper, cause) -> { LOG.debug( "Removing IcebergCatalogWrapper from cache: catalog={}, cause={}", catalogName, - c); - closeIcebergCatalogWrapper((IcebergCatalogWrapper) v); + cause); + closeIcebergCatalogWrapper(catalogWrapper); }) .scheduler( Scheduler.forScheduledExecutorService( @@ -140,22 +149,26 @@ private CatalogWrapperForREST createCatalogWrapper(String catalogName) { @VisibleForTesting protected CatalogWrapperForREST createCatalogWrapper( String catalogName, IcebergConfig icebergConfig) { + // Mint GCS OAuth2 tokens into the config before constructing the wrapper so the IRC catalog + // cache can expire the entry before gcs.oauth2.token-expires-at. + IcebergConfig enrichedConfig = + IcebergCatalogUtil.withGcsServiceAccountCredentials(icebergConfig); // When the backend is a federated Iceberg REST catalog, use FederatedCatalogWrapper so // federation-aware behavior (FileIO property extraction, remote credential vending, remote // /v1/config defaults) is applied through polymorphic dispatch rather than scattered // instanceof checks. All other backends use the base CatalogWrapperForREST. IcebergCatalogBackend backend = IcebergCatalogBackend.valueOf( - icebergConfig.get(IcebergConfig.CATALOG_BACKEND).toUpperCase(Locale.ROOT)); + enrichedConfig.get(IcebergConfig.CATALOG_BACKEND).toUpperCase(Locale.ROOT)); CatalogWrapperForREST rest = backend == IcebergCatalogBackend.REST - ? new FederatedCatalogWrapper(catalogName, icebergConfig) - : new CatalogWrapperForREST(catalogName, icebergConfig); + ? new FederatedCatalogWrapper(catalogName, enrichedConfig) + : new CatalogWrapperForREST(catalogName, enrichedConfig); AuthenticationConfig authenticationConfig = - new AuthenticationConfig(icebergConfig.getAllConfig()); + new AuthenticationConfig(enrichedConfig.getAllConfig()); if (authenticationConfig.isKerberosAuth() && rest.getCatalog() instanceof SupportsKerberos) { return (CatalogWrapperForREST) - new KerberosAwareIcebergCatalogProxy(rest).getProxy(catalogName, icebergConfig); + new KerberosAwareIcebergCatalogProxy(rest).getProxy(catalogName, enrichedConfig); } return rest; @@ -169,8 +182,73 @@ private void closeIcebergCatalogWrapper(IcebergCatalogWrapper catalogWrapper) { } } + /** + * Computes how long a catalog wrapper may stay in the IRC cache. + * + *

Uses the configured access-based eviction interval, capped by the time until a minted GCS + * OAuth2 token should be refreshed ({@code gcs.oauth2.token-expires-at} minus {@link + * #GCS_TOKEN_REFRESH_BUFFER_MS}). When no token expiry is present, returns {@code + * accessEvictionNanos}. + * + * @param config catalog config that may contain {@code gcs.oauth2.token-expires-at} + * @param accessEvictionNanos default expire-after-access duration in nanoseconds + * @param nowEpochMillis current wall-clock time + * @return cache duration in nanoseconds; {@code 0} means expire immediately + */ + @VisibleForTesting + static long computeCacheDurationNanos( + IcebergConfig config, long accessEvictionNanos, long nowEpochMillis) { + String expiresAt = + config.getAllConfig().get(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN_EXPIRES_AT); + if (StringUtils.isBlank(expiresAt)) { + return accessEvictionNanos; + } + + long expiresAtMs; + try { + expiresAtMs = Long.parseLong(expiresAt); + } catch (NumberFormatException e) { + LOG.warn("Invalid {}: {}", IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN_EXPIRES_AT, expiresAt); + return accessEvictionNanos; + } + + long remainingMs = expiresAtMs - GCS_TOKEN_REFRESH_BUFFER_MS - nowEpochMillis; + if (remainingMs <= 0) { + return 0L; + } + return Math.min(accessEvictionNanos, TimeUnit.MILLISECONDS.toNanos(remainingMs)); + } + @Override public void close() throws Exception { catalogWrapperCache.invalidateAll(); } + + private static final class CatalogWrapperExpiry implements Expiry { + + private final long accessEvictionNanos; + + CatalogWrapperExpiry(long accessEvictionNanos) { + this.accessEvictionNanos = accessEvictionNanos; + } + + @Override + public long expireAfterCreate(String key, CatalogWrapperForREST value, long currentTime) { + return computeCacheDurationNanos( + value.getIcebergConfig(), accessEvictionNanos, System.currentTimeMillis()); + } + + @Override + public long expireAfterUpdate( + String key, CatalogWrapperForREST value, long currentTime, long currentDuration) { + return expireAfterCreate(key, value, currentTime); + } + + @Override + public long expireAfterRead( + String key, CatalogWrapperForREST value, long currentTime, long currentDuration) { + // Preserve expire-after-access, but never extend past the GCS token refresh deadline. + return expireAfterCreate(key, value, currentTime); + } + } } diff --git a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestCatalogWrapperForREST.java b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestCatalogWrapperForREST.java index 9ea4b1b67e7..50a3f684d16 100644 --- a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestCatalogWrapperForREST.java +++ b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestCatalogWrapperForREST.java @@ -907,17 +907,6 @@ void testCatalogConfigToClientsIncludesResolvingFileIO() { ResolvingFileIO.class.getName(), configToClients.get(IcebergConstants.IO_IMPL)); } - @Test - void testCatalogConfigRewritesGravitinoGcsFileIOForClients() { - Map configToClients = - CatalogWrapperForREST.filterCatalogConfigForClients( - ImmutableMap.of( - IcebergConstants.IO_IMPL, - org.apache.gravitino.iceberg.common.io.GravitinoGCSFileIO.class.getName())); - Assertions.assertEquals( - "org.apache.iceberg.gcp.gcs.GCSFileIO", configToClients.get(IcebergConstants.IO_IMPL)); - } - @Test void testNonRestCatalogClientConfig() { Map configToClients = diff --git a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergCatalogWrapperManagerForREST.java b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergCatalogWrapperManagerForREST.java index 45eb62f69ee..3a5e8925e12 100644 --- a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergCatalogWrapperManagerForREST.java +++ b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergCatalogWrapperManagerForREST.java @@ -22,6 +22,7 @@ import com.google.common.collect.Maps; import java.util.Map; import java.util.Optional; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; import java.util.function.Consumer; import org.apache.commons.lang3.StringUtils; @@ -189,6 +190,70 @@ public void testDefaultCatalogAliasInvalidatedWhenCatalogRemoved() throws Except } } + @Test + public void testComputeCacheDurationNanosWithoutTokenExpiryUsesAccessEviction() { + IcebergConfig config = + new IcebergConfig(ImmutableMap.of(IcebergConstants.CATALOG_BACKEND, "memory")); + long accessEvictionNanos = TimeUnit.HOURS.toNanos(1); + Assertions.assertEquals( + accessEvictionNanos, + IcebergCatalogWrapperManager.computeCacheDurationNanos( + config, accessEvictionNanos, System.currentTimeMillis())); + } + + @Test + public void testComputeCacheDurationNanosCapsByGcsTokenExpiry() { + long now = 1_700_000_000_000L; + long expiresAt = now + TimeUnit.HOURS.toMillis(1); // token valid for 1h + IcebergConfig config = + new IcebergConfig( + ImmutableMap.of( + IcebergConstants.CATALOG_BACKEND, + "memory", + IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN_EXPIRES_AT, + String.valueOf(expiresAt))); + long accessEvictionNanos = TimeUnit.HOURS.toNanos(2); + long expected = + TimeUnit.MILLISECONDS.toNanos( + TimeUnit.HOURS.toMillis(1) - IcebergCatalogWrapperManager.GCS_TOKEN_REFRESH_BUFFER_MS); + Assertions.assertEquals( + expected, + IcebergCatalogWrapperManager.computeCacheDurationNanos(config, accessEvictionNanos, now)); + } + + @Test + public void testComputeCacheDurationNanosExpiresImmediatelyWhenPastRefreshDeadline() { + long now = 1_700_000_000_000L; + long expiresAt = now + TimeUnit.MINUTES.toMillis(2); // within 5-minute buffer + IcebergConfig config = + new IcebergConfig( + ImmutableMap.of( + IcebergConstants.CATALOG_BACKEND, + "memory", + IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN_EXPIRES_AT, + String.valueOf(expiresAt))); + Assertions.assertEquals( + 0L, + IcebergCatalogWrapperManager.computeCacheDurationNanos( + config, TimeUnit.HOURS.toNanos(1), now)); + } + + @Test + public void testComputeCacheDurationNanosIgnoresInvalidExpiresAt() { + IcebergConfig config = + new IcebergConfig( + ImmutableMap.of( + IcebergConstants.CATALOG_BACKEND, + "memory", + IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN_EXPIRES_AT, + "not-a-number")); + long accessEvictionNanos = TimeUnit.MINUTES.toNanos(30); + Assertions.assertEquals( + accessEvictionNanos, + IcebergCatalogWrapperManager.computeCacheDurationNanos( + config, accessEvictionNanos, System.currentTimeMillis())); + } + private static IcebergCatalogWrapperManager newManager() { Map config = Maps.newHashMap(); IcebergConfigProvider configProvider = IcebergConfigProviderFactory.create(config); From fdd7610e080284a553f1a0d866f2a25a47bd9595 Mon Sep 17 00:00:00 2001 From: remilu <1175334135@qq.com> Date: Wed, 9 Sep 2026 09:41:00 +0800 Subject: [PATCH 10/11] [#9418] test(core): wait for ClassLoader pool drain after manager close Caffeine may retire a missed cache entry asynchronously after close(), so asserting pool size immediately can flake with size 1. Co-authored-by: Cursor --- .../catalog/TestClassLoaderPoolIntegration.java | 14 +++++++++++++- 1 file changed, 13 insertions(+), 1 deletion(-) diff --git a/core/src/test/java/org/apache/gravitino/catalog/TestClassLoaderPoolIntegration.java b/core/src/test/java/org/apache/gravitino/catalog/TestClassLoaderPoolIntegration.java index 5589d41d53b..988281a9a6c 100644 --- a/core/src/test/java/org/apache/gravitino/catalog/TestClassLoaderPoolIntegration.java +++ b/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; } From 15f2d51a86c82459c40c746801a06b1c84856d22 Mon Sep 17 00:00:00 2001 From: remilu <1175334135@qq.com> Date: Wed, 9 Sep 2026 12:03:17 +0800 Subject: [PATCH 11/11] [#9418] chore: retrigger CI after BackendIT mysql timeout Co-authored-by: Cursor