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 e7f97b6bdf6..120cebb6196 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/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; } diff --git a/docs/iceberg-rest-service.md b/docs/iceberg-rest-service.md index 8339d69268e..f3c55cdafde 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. 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 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 938fea8e314..364e5cf29c7 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..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,15 +21,25 @@ import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.HADOOP_SECURITY_AUTHENTICATION; import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.HADOOP_SECURITY_AUTHORIZATION; +import com.google.auth.oauth2.AccessToken; +import com.google.auth.oauth2.GoogleCredentials; import com.google.common.annotations.VisibleForTesting; import com.google.common.collect.Maps; import java.io.IOException; +import java.io.InputStream; +import java.io.UncheckedIOException; +import java.nio.file.Files; +import java.nio.file.NoSuchFileException; +import java.nio.file.Path; +import java.nio.file.Paths; import java.sql.SQLException; import java.util.Collections; +import java.util.Date; import java.util.HashMap; import java.util.Locale; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; +import org.apache.commons.lang3.StringUtils; import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergCatalogBackend; import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants; import org.apache.gravitino.exceptions.ConnectionFailedException; @@ -38,6 +48,7 @@ import org.apache.gravitino.iceberg.common.IcebergConfig; import org.apache.gravitino.iceberg.common.authentication.AuthenticationConfig; import org.apache.gravitino.iceberg.common.rest.auth.UserPrincipalForwardingAuthManager; +import org.apache.gravitino.storage.GCSProperties; import org.apache.hadoop.hdfs.HdfsConfiguration; import org.apache.iceberg.CatalogProperties; import org.apache.iceberg.CatalogUtil; @@ -65,6 +76,9 @@ public class IcebergCatalogUtil { */ private static final String ICEBERG_TYPE_COLUMN = "iceberg_type"; + private static final String GCS_CLOUD_PLATFORM_SCOPE = + "https://www.googleapis.com/auth/cloud-platform"; + private static final ConcurrentHashMap MEMORY_CATALOGS = new ConcurrentHashMap<>(); @@ -269,6 +283,84 @@ 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 + static void applyGcsServiceAccountCredentials(Map 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 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 f5ee1b8cf01..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; @@ -29,6 +30,7 @@ import org.apache.gravitino.iceberg.common.ClosableJdbcCatalog; import org.apache.gravitino.iceberg.common.IcebergConfig; import org.apache.gravitino.iceberg.common.authentication.AuthenticationConfig; +import org.apache.gravitino.storage.GCSProperties; import org.apache.iceberg.CatalogProperties; import org.apache.iceberg.Schema; import org.apache.iceberg.catalog.Catalog; @@ -337,6 +339,74 @@ void testApplyDefaultResolvingFileIODoesNotOverrideExplicitIOImpl() { "org.apache.iceberg.aws.s3.S3FileIO", properties.get(IcebergConstants.IO_IMPL)); } + @Test + void testApplyGcsServiceAccountCredentialsSkipsWhenTokenAlreadyPresent() { + Map 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 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 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 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<>(); + 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 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..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/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);