This is an automated email from the ASF dual-hosted git repository. jerryshao pushed a commit to branch branch-1.3 in repository https://gitbox.apache.org/repos/asf/gravitino.git
commit 531058d2e913dfa5b5e1f5fbb64fc9ac1d4d6a37 Author: MaSai <[email protected]> AuthorDate: Wed Sep 9 17:09:26 2026 +0800 [#9418] fix(iceberg): inject GCS FileIO token from gcs-service-account-file (#12962) Iceberg's `GCSFileIO` does not understand Gravitino's `gcs-service-account-file`. When that property is set, load the service account at catalog initialization and inject Iceberg `gcs.oauth2.token` / `gcs.oauth2.token-expires-at` so server-side FileIO can authenticate (same property-injection style as S3/OSS/ADLS key mapping). Also update GCS docs that previously required `GOOGLE_APPLICATION_CREDENTIALS` even when the catalog property was set. `gcs-service-account-file` reached the credential provider (vending worked) but not FileIO. Table create wrote metadata with Application Default Credentials and failed with 401 unless `GOOGLE_APPLICATION_CREDENTIALS` was set on the process. Fix: #9418 - Configuring `gcs-service-account-file` is sufficient for Iceberg GCS FileIO; `GOOGLE_APPLICATION_CREDENTIALS` is only a fallback when the property is unset. - No new public API or property keys. ``` ./gradlew :iceberg:iceberg-common:test --tests org.apache.gravitino.iceberg.common.utils.TestIcebergCatalogUtil -PskipITs ``` Made with [Cursor](https://cursor.com) --------- Co-authored-by: Cursor <[email protected]> --- .../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 + .../iceberg/common/utils/IcebergCatalogUtil.java | 92 ++++++++++++++++++ .../common/utils/TestIcebergCatalogUtil.java | 70 ++++++++++++++ .../service/IcebergCatalogWrapperManager.java | 107 ++++++++++++++++++--- .../TestIcebergCatalogWrapperManagerForREST.java | 65 +++++++++++++ 9 files changed, 348 insertions(+), 22 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 e7f97b6bdf..120cebb619 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 @@ -77,6 +77,20 @@ public class IcebergConstants { public static final String AZURE_CLIENT_SECRET_TOKEN_CREDENTIAL_PROVIDER = "org.apache.gravitino.iceberg.common.credential.AzureClientSecretTokenCredentialProvider"; + /** Iceberg GCSFileIO OAuth2 access token property. */ + public static final String ICEBERG_GCS_OAUTH2_TOKEN = "gcs.oauth2.token"; + + /** Iceberg GCSFileIO OAuth2 token expiry property (epoch millis). */ + public static final String ICEBERG_GCS_OAUTH2_TOKEN_EXPIRES_AT = "gcs.oauth2.token-expires-at"; + + /** + * Whether Iceberg GCSFileIO should refresh OAuth2 tokens via a credentials endpoint. Defaults to + * true in Iceberg; Gravitino disables it when minting a token from {@code + * gcs-service-account-file} because that path has no table credentials refresh endpoint. + */ + public static final String ICEBERG_GCS_OAUTH2_REFRESH_CREDENTIALS_ENABLED = + "gcs.oauth2.refresh-credentials-enabled"; + // Iceberg Table properties constants public static final String COMMENT = "comment"; diff --git a/docs/iceberg-rest-service.md b/docs/iceberg-rest-service.md index 2d07fc4f78..c424e39127 100644 --- a/docs/iceberg-rest-service.md +++ b/docs/iceberg-rest-service.md @@ -529,16 +529,17 @@ Please set the `gravitino.iceberg-rest.warehouse` parameter to `oss://{bucket_na Supports using static GCS credential file or generating GCS token to access GCS data. -| Configuration item | Description | Default value | Required | -|----------------------------------|------------------------------------------------------------------------------------------------------------------------------|-----------------------------------------|----------| -| `gravitino.iceberg-rest.io-impl` | The IO implementation for `FileIO` in Iceberg. Set it to `org.apache.iceberg.gcp.gcs.GCSFileIO` to explicitly use GCSFileIO. | `org.apache.iceberg.io.ResolvingFileIO` | No | +| Configuration item | Description | Default value | Required | +|---------------------------------------------------|------------------------------------------------------------------------------------------------------------------------------|-----------------------------------------|----------| +| `gravitino.iceberg-rest.io-impl` | The IO implementation for `FileIO` in Iceberg. Set it to `org.apache.iceberg.gcp.gcs.GCSFileIO` to explicitly use GCSFileIO. | `org.apache.iceberg.io.ResolvingFileIO` | No | +| `gravitino.iceberg-rest.gcs-service-account-file` | Path of the GCS service account JSON file. Used for server-side FileIO and for `gcs-token` credential vending. | GCS Application default credential. | No | For other Iceberg GCS properties not managed by Gravitino like `gcs.project-id`, you could config it directly by `gravitino.iceberg-rest.gcs.project-id`. Refer to [GCS credentials](./security/credential-vending.md#gcs-credentials) for credential related configurations. :::note -Ensure that the credential file is accessible by the Gravitino server. For example, the server may be running on a GCE machine, or you may set the environment variable `export GOOGLE_APPLICATION_CREDENTIALS=/xx/application_default_credentials.json` even when `gcs-service-account-file` is already configured. +When `gcs-service-account-file` is set, Gravitino loads it at catalog initialization and injects Iceberg `gcs.oauth2.token` for FileIO. The IRC catalog cache evicts that catalog before the token expires so the next request recreates the catalog and mints a fresh token. If unset, use Application Default Credentials (for example GCE metadata or `GOOGLE_APPLICATION_CREDENTIALS`). ::: :::info diff --git a/docs/lakehouse-iceberg-catalog.md b/docs/lakehouse-iceberg-catalog.md index d4c0179243..881d77699a 100644 --- a/docs/lakehouse-iceberg-catalog.md +++ b/docs/lakehouse-iceberg-catalog.md @@ -174,13 +174,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 5c2d077f8a..20b56966ad 100644 --- a/docs/security/credential-vending.md +++ b/docs/security/credential-vending.md @@ -159,7 +159,7 @@ An GCS token is a token credential with scoped privileges, by leveraging GCS [Cr | `gcs-service-account-file` | `gravitino.iceberg-rest.gcs-service-account-file` | The location of GCS credential file. | GCS Application default credential. | No | :::note -For the Gravitino Iceberg REST server, ensure that the credential file is accessible by the 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`). ::: ## Custom Credentials diff --git a/iceberg/iceberg-common/build.gradle.kts b/iceberg/iceberg-common/build.gradle.kts index c6275006d4..296db0d8e9 100644 --- a/iceberg/iceberg-common/build.gradle.kts +++ b/iceberg/iceberg-common/build.gradle.kts @@ -59,6 +59,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 f849b77ad5..74aeb2c54b 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,25 @@ package org.apache.gravitino.iceberg.common.utils; import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.HADOOP_SECURITY_AUTHENTICATION; import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.HADOOP_SECURITY_AUTHORIZATION; +import com.google.auth.oauth2.AccessToken; +import com.google.auth.oauth2.GoogleCredentials; import com.google.common.annotations.VisibleForTesting; import com.google.common.collect.Maps; import java.io.IOException; +import java.io.InputStream; +import java.io.UncheckedIOException; +import java.nio.file.Files; +import java.nio.file.NoSuchFileException; +import java.nio.file.Path; +import java.nio.file.Paths; import java.sql.SQLException; import java.util.Collections; +import java.util.Date; import java.util.HashMap; import java.util.Locale; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; +import org.apache.commons.lang3.StringUtils; import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergCatalogBackend; import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants; import org.apache.gravitino.exceptions.ConnectionFailedException; @@ -37,6 +47,7 @@ import org.apache.gravitino.iceberg.common.ClosableHiveCatalog; 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; @@ -58,6 +69,9 @@ public class IcebergCatalogUtil { private static final Logger LOG = LoggerFactory.getLogger(IcebergCatalogUtil.class); + private static final String GCS_CLOUD_PLATFORM_SCOPE = + "https://www.googleapis.com/auth/cloud-platform";; + private static final ConcurrentHashMap<String, InMemoryCatalog> MEMORY_CATALOGS = new ConcurrentHashMap<>(); @@ -208,6 +222,84 @@ public class IcebergCatalogUtil { @VisibleForTesting public static void applyDefaultResolvingFileIO(Map<String, String> properties) { properties.putIfAbsent(IcebergConstants.IO_IMPL, ResolvingFileIO.class.getName()); + applyGcsServiceAccountCredentials(properties); + } + + /** + * When {@code gcs-service-account-file} is set, mint an OAuth2 access token and inject Iceberg + * {@code gcs.oauth2.token} / {@code gcs.oauth2.token-expires-at} so the built-in {@code + * GCSFileIO} can authenticate. Iceberg's FileIO does not understand Gravitino's + * service-account-file property; S3/OSS/ADLS instead map static keys directly via {@link + * org.apache.gravitino.catalog.lakehouse.iceberg.IcebergPropertiesUtils}. + * + * <p>Skips injection when {@code gcs.oauth2.token} is already present. Disables Iceberg's + * credentials-endpoint refresh because that path is for vended table credentials, not catalog + * bootstrap from a service account file. + * + * @param properties Iceberg catalog properties, mutated in place + */ + @VisibleForTesting + static void applyGcsServiceAccountCredentials(Map<String, String> properties) { + String serviceAccountFile = properties.get(GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE); + if (StringUtils.isBlank(serviceAccountFile)) { + return; + } + if (StringUtils.isNotBlank(properties.get(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN))) { + return; + } + + AccessToken accessToken = loadAccessTokenFromFile(serviceAccountFile); + if (accessToken == null || StringUtils.isBlank(accessToken.getTokenValue())) { + throw new IllegalStateException( + "Failed to obtain GCS access token from service account file: " + serviceAccountFile); + } + + properties.put(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN, accessToken.getTokenValue()); + Date expirationTime = accessToken.getExpirationTime(); + if (expirationTime != null) { + properties.put( + IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN_EXPIRES_AT, + String.valueOf(expirationTime.toInstant().toEpochMilli())); + } + properties.put(IcebergConstants.ICEBERG_GCS_OAUTH2_REFRESH_CREDENTIALS_ENABLED, "false"); + LOG.info( + "Injected {} from {} for Iceberg GCSFileIO", + IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN, + GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE); + } + + /** + * Returns an {@link IcebergConfig} that includes a minted GCS OAuth2 token when {@code + * gcs-service-account-file} is configured. The returned config retains {@code + * gcs.oauth2.token-expires-at} so callers (for example the IRC catalog cache) can expire the + * catalog before the token becomes invalid. + * + * @param icebergConfig original catalog config + * @return the same instance when no token is injected; otherwise a new config with token fields + */ + public static IcebergConfig withGcsServiceAccountCredentials(IcebergConfig icebergConfig) { + Map<String, String> properties = new HashMap<>(icebergConfig.getAllConfig()); + applyGcsServiceAccountCredentials(properties); + if (properties.equals(icebergConfig.getAllConfig())) { + return icebergConfig; + } + return new IcebergConfig(properties); + } + + private static AccessToken loadAccessTokenFromFile(String serviceAccountFile) { + Path credentialsFilePath = Paths.get(serviceAccountFile); + try (InputStream inputStream = Files.newInputStream(credentialsFilePath)) { + GoogleCredentials credentials = + GoogleCredentials.fromStream(inputStream).createScoped(GCS_CLOUD_PLATFORM_SCOPE); + credentials.refreshIfExpired(); + return credentials.getAccessToken(); + } catch (NoSuchFileException e) { + throw new UncheckedIOException( + "GCS service account file does not exist: " + serviceAccountFile, e); + } catch (IOException e) { + throw new UncheckedIOException( + "Failed to load GCS service account file: " + serviceAccountFile, e); + } } @VisibleForTesting 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 89cf6f7897..337f8684da 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,14 @@ package org.apache.gravitino.iceberg.common.utils; +import java.io.UncheckedIOException; import java.nio.file.Path; import java.util.HashMap; import java.util.Map; import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergCatalogBackend; import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants; import org.apache.gravitino.iceberg.common.IcebergConfig; +import org.apache.gravitino.storage.GCSProperties; import org.apache.iceberg.CatalogProperties; import org.apache.iceberg.Schema; import org.apache.iceberg.catalog.Catalog; @@ -312,6 +314,74 @@ public class TestIcebergCatalogUtil { "org.apache.iceberg.aws.s3.S3FileIO", properties.get(IcebergConstants.IO_IMPL)); } + @Test + void testApplyGcsServiceAccountCredentialsSkipsWhenTokenAlreadyPresent() { + Map<String, String> properties = new HashMap<>(); + properties.put(GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE, "/tmp/gcs-key.json"); + properties.put(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN, "existing-token"); + + IcebergCatalogUtil.applyGcsServiceAccountCredentials(properties); + + Assertions.assertEquals( + "existing-token", properties.get(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN)); + Assertions.assertNull(properties.get(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN_EXPIRES_AT)); + } + + @Test + void testApplyGcsServiceAccountCredentialsNoOpWithoutServiceAccountFile() { + Map<String, String> properties = new HashMap<>(); + properties.put(IcebergConstants.IO_IMPL, "org.apache.iceberg.gcp.gcs.GCSFileIO"); + + IcebergCatalogUtil.applyGcsServiceAccountCredentials(properties); + + Assertions.assertNull(properties.get(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN)); + } + + @Test + void testApplyGcsServiceAccountCredentialsFailsWhenFileMissing() { + Map<String, String> properties = new HashMap<>(); + properties.put( + GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE, "/tmp/gravitino-missing-gcs-key.json"); + + UncheckedIOException thrown = + Assertions.assertThrows( + UncheckedIOException.class, + () -> IcebergCatalogUtil.applyGcsServiceAccountCredentials(properties)); + Assertions.assertTrue(thrown.getMessage().contains("does not exist")); + } + + @Test + void testWithGcsServiceAccountCredentialsReturnsSameConfigWhenNoServiceAccountFile() { + IcebergConfig config = new IcebergConfig(Map.of(IcebergConstants.CATALOG_BACKEND, "memory")); + Assertions.assertSame(config, IcebergCatalogUtil.withGcsServiceAccountCredentials(config)); + } + + @Test + void testWithGcsServiceAccountCredentialsReturnsSameConfigWhenTokenAlreadyPresent() { + Map<String, String> properties = new HashMap<>(); + properties.put(GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE, "/tmp/gcs-key.json"); + properties.put(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN, "existing-token"); + IcebergConfig config = new IcebergConfig(properties); + + Assertions.assertSame(config, IcebergCatalogUtil.withGcsServiceAccountCredentials(config)); + } + + @Test + void testApplyDefaultResolvingFileIOInjectsGcsToken() { + Map<String, String> properties = new HashMap<>(); + properties.put(IcebergConstants.WAREHOUSE, "gs://bucket/warehouse"); + properties.put(GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE, "/tmp/gcs-key.json"); + + // Pre-set token so applyDefaultResolvingFileIO skips loading a real service account file. + properties.put(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN, "pre-set"); + IcebergCatalogUtil.applyDefaultResolvingFileIO(properties); + + Assertions.assertEquals( + org.apache.iceberg.io.ResolvingFileIO.class.getName(), + properties.get(IcebergConstants.IO_IMPL)); + Assertions.assertEquals("pre-set", properties.get(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN)); + } + @Test void testApplyRestCatalogHttpTimeoutPropertiesUsesDefaults() { Map<String, String> properties = new HashMap<>(); 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 3a3c518f30..a960f175e0 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 @@ package org.apache.gravitino.iceberg.service; 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.Map; 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.AuthenticationConfig; 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<String, CatalogWrapperForREST> catalogWrapperCache; private final IcebergConfigProvider configProvider; @@ -57,17 +66,20 @@ public class IcebergCatalogWrapperManager implements AutoCloseable { 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; - LOG.info("Remove IcebergCatalogWrapper cache {}.", catalogName); - closeIcebergCatalogWrapper((IcebergCatalogWrapper) v); + (catalogName, catalogWrapper, cause) -> { + LOG.debug( + "Removing IcebergCatalogWrapper from cache: catalog={}, cause={}", + catalogName, + cause); + closeIcebergCatalogWrapper(catalogWrapper); }) .scheduler( Scheduler.forScheduledExecutorService( @@ -137,22 +149,26 @@ public class IcebergCatalogWrapperManager implements AutoCloseable { @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; @@ -166,8 +182,73 @@ public class IcebergCatalogWrapperManager implements AutoCloseable { } } + /** + * Computes how long a catalog wrapper may stay in the IRC cache. + * + * <p>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<String, CatalogWrapperForREST> { + + 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 45eb62f69e..3a5e8925e1 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.ImmutableMap; 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 class TestIcebergCatalogWrapperManagerForREST { } } + @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<String, String> config = Maps.newHashMap(); IcebergConfigProvider configProvider = IcebergConfigProviderFactory.create(config);
