This is an automated email from the ASF dual-hosted git repository.
roryqi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new 5c8f5bf9d2 [#9418] fix(iceberg): inject GCS FileIO token from
gcs-service-account-file (#12962)
5c8f5bf9d2 is described below
commit 5c8f5bf9d210b49d135896dd3c0fa64ebb065e24
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)
### What changes were proposed in this pull request?
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.
### Why are the changes needed?
`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
### Does this PR introduce _any_ user-facing change?
- 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.
### How was this patch tested?
```
./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 +++
.../catalog/TestClassLoaderPoolIntegration.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 | 104 ++++++++++++++++++---
.../TestIcebergCatalogWrapperManagerForREST.java | 65 +++++++++++++
10 files changed, 358 insertions(+), 23 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/core/src/test/java/org/apache/gravitino/catalog/TestClassLoaderPoolIntegration.java
b/core/src/test/java/org/apache/gravitino/catalog/TestClassLoaderPoolIntegration.java
index 5589d41d53..988281a9a6 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 com.google.common.collect.ImmutableMap;
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 class TestClassLoaderPoolIntegration {
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 8339d69268..f3c55cdafd 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 996c0427e9..6f4741f15f 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 8e0c798af3..7b1aeb9a01 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 938fea8e31..364e5cf29c 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 745b6b1d4f..658786ea15 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;
@@ -38,6 +48,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.iceberg.common.rest.auth.UserPrincipalForwardingAuthManager;
+import org.apache.gravitino.storage.GCSProperties;
import org.apache.hadoop.hdfs.HdfsConfiguration;
import org.apache.iceberg.CatalogProperties;
import org.apache.iceberg.CatalogUtil;
@@ -65,6 +76,9 @@ public class IcebergCatalogUtil {
*/
private static final String ICEBERG_TYPE_COLUMN = "iceberg_type";
+ private static final String GCS_CLOUD_PLATFORM_SCOPE =
+ "https://www.googleapis.com/auth/cloud-platform";;
+
private static final ConcurrentHashMap<String, InMemoryCatalog>
MEMORY_CATALOGS =
new ConcurrentHashMap<>();
@@ -269,6 +283,84 @@ 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 f5ee1b8cf0..d8ce5cc18a 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.catalog.lakehouse.iceberg.IcebergConstants;
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 @@ 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 5d8fbf52f1..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,20 +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;
+ (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 @@ 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;
@@ -169,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);
