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 91ed7a2e75 [#10990] feat(iceberg-rest): Vended credential refresh for
S3, GCS, OSS, and ADLS (#11294)
91ed7a2e75 is described below
commit 91ed7a2e750537276d8ac2f579b81dec162ff307
Author: MaSai <[email protected]>
AuthorDate: Fri Jun 5 14:17:12 2026 +0800
[#10990] feat(iceberg-rest): Vended credential refresh for S3, GCS, OSS,
and ADLS (#11294)
### What changes were proposed in this pull request?
This PR adds **Iceberg 1.11 vended credential refresh** for object
storage in the Iceberg REST Catalog (IRC), and includes the
**server-side Iceberg 1.11.0 upgrade** required for the feature.
**Vended credential refresh (IRC)**
- Add `IcebergRESTUtils.toRESTCredential(...)` to map Gravitino storage
tokens to Iceberg 1.11 client properties, table-scoped refresh paths
(`GET .../tables/{table}/credentials`), and storage `prefix` from table
location.
- Extend `CredentialPropertyUtils.toIcebergProperties` with token expiry
metadata for S3/IRSA, OSS, and ADLS (GCS expiry already supported).
- Wire credential refresh into IRC load-table, credentials API, and
scan-plan responses in `CatalogWrapperForREST`.
- For `catalog-backend=rest`, proxy `/credentials` to the upstream IRC
via `getRESTTableCredentials()` (uses
`UserPrincipalForwardingAuthManager` from `RESTCatalog` properties);
other backends use local `getLocalTableCredentials()` via
`catalogCredentialManager`.
**Iceberg 1.11 upgrade (included in this branch)**
- Upgrade server-side Iceberg to **1.11.0** for IRC and lakehouse
catalog.
- Adapt IRC REST handlers, plan-scan response building, and related
Spark/Flink IT classpath expectations.
### Why are the changes needed?
Iceberg 1.11 clients can refresh vended object-storage credentials
before they expire during long-running jobs. Gravitino IRC must return
expiry fields and table-scoped refresh endpoints for S3, GCS, OSS, and
ADLS when temporary credentials are enabled.
When IRC proxies another IRC (`catalog-backend=rest`), `/credentials`
must be fetched from the upstream catalog instead of being generated
locally on the proxy.
The server must run Iceberg 1.11 to align with the REST spec and client
refresh behavior.
Fix: #10990
### Does this PR introduce _any_ user-facing change?
- When `vended-credentials` is enabled, load-table / credentials /
scan-plan responses may include cloud-specific expiry fields and
refresh-endpoint properties (for example
`client.refresh-credentials-endpoint`,
`gcs.oauth2.refresh-credentials-endpoint`,
`adls.refresh-credentials-endpoint`).
- Refresh paths use the relative form without a leading slash:
`v1/{catalog}/namespaces/.../tables/.../credentials`.
- IRC and lakehouse Iceberg catalog documentation reflect **Iceberg
1.11.0** server support.
### How was this patch tested?
- [x] `./gradlew :common:test --tests
org.apache.gravitino.credential.TestCredentialPropertiesUtils -PskipITs`
- [x] `./gradlew :iceberg:iceberg-rest-server:test --tests
org.apache.gravitino.iceberg.service.TestIcebergRESTUtils --tests
org.apache.gravitino.iceberg.service.TestCatalogWrapperForREST
-PskipITs`
- [x] `./gradlew :iceberg:iceberg-rest-server:compileJava
:iceberg:iceberg-rest-server:compileTestJava -PskipITs`
---
.../gravitino/credential/ADLSTokenCredential.java | 2 +-
.../gravitino/credential/AwsIrsaCredential.java | 9 +-
.../gravitino/credential/GCSTokenCredential.java | 2 +-
.../gravitino/credential/JdbcCredential.java | 2 +-
.../credential/OSSSecretKeyCredential.java | 2 +-
.../gravitino/credential/OSSTokenCredential.java | 2 +-
.../credential/S3SecretKeyCredential.java | 2 +-
.../gravitino/credential/S3TokenCredential.java | 2 +-
.../s3/credential/AwsIrsaCredentialGenerator.java | 10 +-
.../credential/CredentialPropertyUtils.java | 134 +++++--
.../credential/TestCredentialFactory.java | 24 ++
.../credential/TestCredentialPropertiesUtils.java | 75 +++-
.../credential/CatalogCredentialManager.java | 9 +
.../iceberg/service/CatalogWrapperForREST.java | 159 ++++++--
.../iceberg/service/IcebergRESTUtils.java | 124 ++++++
.../iceberg/service/TestCatalogWrapperForREST.java | 416 ++++++++++++++++++++-
.../iceberg/service/TestIcebergRESTUtils.java | 231 ++++++++++++
17 files changed, 1121 insertions(+), 84 deletions(-)
diff --git
a/api/src/main/java/org/apache/gravitino/credential/ADLSTokenCredential.java
b/api/src/main/java/org/apache/gravitino/credential/ADLSTokenCredential.java
index 6f1c463033..d18aa7f115 100644
--- a/api/src/main/java/org/apache/gravitino/credential/ADLSTokenCredential.java
+++ b/api/src/main/java/org/apache/gravitino/credential/ADLSTokenCredential.java
@@ -112,6 +112,6 @@ public class ADLSTokenCredential implements Credential {
Preconditions.checkArgument(
StringUtils.isNotBlank(sasToken), "ADLS SAS token should not be
empty.");
Preconditions.checkArgument(
- expireTimeInMS > 0, "The expire time of ADLSTokenCredential should
great than 0");
+ expireTimeInMS > 0, "The expiration time of ADLSTokenCredential should
be greater than 0");
}
}
diff --git
a/api/src/main/java/org/apache/gravitino/credential/AwsIrsaCredential.java
b/api/src/main/java/org/apache/gravitino/credential/AwsIrsaCredential.java
index 0c1888cffc..4fecf9ada1 100644
--- a/api/src/main/java/org/apache/gravitino/credential/AwsIrsaCredential.java
+++ b/api/src/main/java/org/apache/gravitino/credential/AwsIrsaCredential.java
@@ -50,7 +50,7 @@ public class AwsIrsaCredential implements Credential {
*/
public AwsIrsaCredential(
String accessKeyId, String secretAccessKey, String sessionToken, long
expireTimeInMs) {
- validate(accessKeyId, secretAccessKey, sessionToken);
+ validate(accessKeyId, secretAccessKey, sessionToken, expireTimeInMs);
this.accessKeyId = accessKeyId;
this.secretAccessKey = secretAccessKey;
this.sessionToken = sessionToken;
@@ -84,7 +84,7 @@ public class AwsIrsaCredential implements Credential {
String accessKeyId = credentialInfo.get(ACCESS_KEY_ID);
String secretAccessKey = credentialInfo.get(SECRET_ACCESS_KEY);
String sessionToken = credentialInfo.get(SESSION_TOKEN);
- validate(accessKeyId, secretAccessKey, sessionToken);
+ validate(accessKeyId, secretAccessKey, sessionToken, expireTimeInMs);
this.accessKeyId = accessKeyId;
this.secretAccessKey = secretAccessKey;
this.sessionToken = sessionToken;
@@ -118,12 +118,15 @@ public class AwsIrsaCredential implements Credential {
return sessionToken;
}
- private void validate(String accessKeyId, String secretAccessKey, String
sessionToken) {
+ private void validate(
+ String accessKeyId, String secretAccessKey, String sessionToken, long
expireTimeInMs) {
Preconditions.checkArgument(
StringUtils.isNotBlank(accessKeyId), "Access key Id should not be
empty");
Preconditions.checkArgument(
StringUtils.isNotBlank(secretAccessKey), "Secret access key should not
be empty");
Preconditions.checkArgument(
StringUtils.isNotBlank(sessionToken), "Session token should not be
empty");
+ Preconditions.checkArgument(
+ expireTimeInMs > 0, "The expiration time of AwsIrsaCredential should
be greater than 0");
}
}
diff --git
a/api/src/main/java/org/apache/gravitino/credential/GCSTokenCredential.java
b/api/src/main/java/org/apache/gravitino/credential/GCSTokenCredential.java
index d55799c52a..a3230645fe 100644
--- a/api/src/main/java/org/apache/gravitino/credential/GCSTokenCredential.java
+++ b/api/src/main/java/org/apache/gravitino/credential/GCSTokenCredential.java
@@ -88,6 +88,6 @@ public class GCSTokenCredential implements Credential {
Preconditions.checkArgument(
StringUtils.isNotBlank(token), "GCS session token should not be
empty");
Preconditions.checkArgument(
- expireTimeInMs > 0, "The expire time of GcsTokenCredential should
great than 0");
+ expireTimeInMs > 0, "The expiration time of GcsTokenCredential should
be greater than 0");
}
}
diff --git
a/api/src/main/java/org/apache/gravitino/credential/JdbcCredential.java
b/api/src/main/java/org/apache/gravitino/credential/JdbcCredential.java
index eeaccc2620..af19110d7e 100644
--- a/api/src/main/java/org/apache/gravitino/credential/JdbcCredential.java
+++ b/api/src/main/java/org/apache/gravitino/credential/JdbcCredential.java
@@ -111,6 +111,6 @@ public class JdbcCredential implements Credential {
StringUtils.isNotBlank(jdbcPassword), "JDBC password should not be
empty");
// JDBC credentials are static (no server-issued expiry). expireTimeInMs
must always be 0.
Preconditions.checkArgument(
- expireTimeInMs == 0, "The expire time of JdbcCredential should be 0");
+ expireTimeInMs == 0, "The expiration time of JdbcCredential should be
0");
}
}
diff --git
a/api/src/main/java/org/apache/gravitino/credential/OSSSecretKeyCredential.java
b/api/src/main/java/org/apache/gravitino/credential/OSSSecretKeyCredential.java
index fd98a3bfd3..34baa51707 100644
---
a/api/src/main/java/org/apache/gravitino/credential/OSSSecretKeyCredential.java
+++
b/api/src/main/java/org/apache/gravitino/credential/OSSSecretKeyCredential.java
@@ -115,6 +115,6 @@ public class OSSSecretKeyCredential implements Credential {
Preconditions.checkArgument(
StringUtils.isNotBlank(secretAccessKey), "OSS secret access key should
not empty");
Preconditions.checkArgument(
- expireTimeInMs == 0, "The expire time of OSSSecretKeyCredential is not
0");
+ expireTimeInMs == 0, "The expiration time of OSSSecretKeyCredential is
not 0");
}
}
diff --git
a/api/src/main/java/org/apache/gravitino/credential/OSSTokenCredential.java
b/api/src/main/java/org/apache/gravitino/credential/OSSTokenCredential.java
index 70e8839489..4f1a871501 100644
--- a/api/src/main/java/org/apache/gravitino/credential/OSSTokenCredential.java
+++ b/api/src/main/java/org/apache/gravitino/credential/OSSTokenCredential.java
@@ -133,6 +133,6 @@ public class OSSTokenCredential implements Credential {
Preconditions.checkArgument(
StringUtils.isNotBlank(sessionToken), "OSS session token should not be
empty");
Preconditions.checkArgument(
- expireTimeInMs > 0, "The expire time of OSSTokenCredential should
great than 0");
+ expireTimeInMs > 0, "The expiration time of OSSTokenCredential should
be greater than 0");
}
}
diff --git
a/api/src/main/java/org/apache/gravitino/credential/S3SecretKeyCredential.java
b/api/src/main/java/org/apache/gravitino/credential/S3SecretKeyCredential.java
index d9b2b60925..db6de05854 100644
---
a/api/src/main/java/org/apache/gravitino/credential/S3SecretKeyCredential.java
+++
b/api/src/main/java/org/apache/gravitino/credential/S3SecretKeyCredential.java
@@ -107,6 +107,6 @@ public class S3SecretKeyCredential implements Credential {
Preconditions.checkArgument(
StringUtils.isNotBlank(secretAccessKey), "S3 secret access key should
not empty");
Preconditions.checkArgument(
- expireTimeInMs == 0, "The expire time of S3SecretKeyCredential is not
0");
+ expireTimeInMs == 0, "The expiration time of S3SecretKeyCredential is
not 0");
}
}
diff --git
a/api/src/main/java/org/apache/gravitino/credential/S3TokenCredential.java
b/api/src/main/java/org/apache/gravitino/credential/S3TokenCredential.java
index 89d7de760f..ce69ab3b45 100644
--- a/api/src/main/java/org/apache/gravitino/credential/S3TokenCredential.java
+++ b/api/src/main/java/org/apache/gravitino/credential/S3TokenCredential.java
@@ -131,6 +131,6 @@ public class S3TokenCredential implements Credential {
Preconditions.checkArgument(
StringUtils.isNotBlank(sessionToken), "S3 session token should not be
empty");
Preconditions.checkArgument(
- expireTimeInMs > 0, "The expire time of S3TokenCredential should great
than 0");
+ expireTimeInMs > 0, "The expiration time of S3TokenCredential should
be greater than 0");
}
}
diff --git
a/bundles/aws/src/main/java/org/apache/gravitino/s3/credential/AwsIrsaCredentialGenerator.java
b/bundles/aws/src/main/java/org/apache/gravitino/s3/credential/AwsIrsaCredentialGenerator.java
index cd5ccad8c5..3c1641506e 100644
---
a/bundles/aws/src/main/java/org/apache/gravitino/s3/credential/AwsIrsaCredentialGenerator.java
+++
b/bundles/aws/src/main/java/org/apache/gravitino/s3/credential/AwsIrsaCredentialGenerator.java
@@ -81,10 +81,12 @@ public class AwsIrsaCredentialGenerator implements
CredentialGenerator<AwsIrsaCr
AwsCredentials creds = baseCredentialsProvider.resolveCredentials();
if (creds instanceof AwsSessionCredentials) {
AwsSessionCredentials sessionCreds = (AwsSessionCredentials) creds;
- long expiration =
- sessionCreds.expirationTime().isPresent()
- ? sessionCreds.expirationTime().get().toEpochMilli()
- : 0L;
+ if (!sessionCreds.expirationTime().isPresent()) {
+ throw new IllegalStateException(
+ "AWS IRSA session credentials must include an expiration time
for vended credential"
+ + " refresh");
+ }
+ long expiration = sessionCreds.expirationTime().get().toEpochMilli();
return new AwsIrsaCredential(
sessionCreds.accessKeyId(),
sessionCreds.secretAccessKey(),
diff --git
a/common/src/main/java/org/apache/gravitino/credential/CredentialPropertyUtils.java
b/common/src/main/java/org/apache/gravitino/credential/CredentialPropertyUtils.java
index b66b650178..d9543b93d3 100644
---
a/common/src/main/java/org/apache/gravitino/credential/CredentialPropertyUtils.java
+++
b/common/src/main/java/org/apache/gravitino/credential/CredentialPropertyUtils.java
@@ -23,6 +23,7 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
+import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
@@ -35,12 +36,22 @@ public class CredentialPropertyUtils {
@VisibleForTesting static final String ICEBERG_S3_ACCESS_KEY_ID =
"s3.access-key-id";
@VisibleForTesting static final String ICEBERG_S3_SECRET_ACCESS_KEY =
"s3.secret-access-key";
@VisibleForTesting static final String ICEBERG_S3_TOKEN = "s3.session-token";
- @VisibleForTesting static final String ICEBERG_GCS_TOKEN =
"gcs.oauth2.token";
+
+ @VisibleForTesting
+ static final String ICEBERG_S3_TOKEN_EXPIRES_AT_MS =
"s3.session-token-expires-at-ms";
+
+ @VisibleForTesting
+ static final String ICEBERG_CLIENT_REFRESH_CREDENTIALS_ENDPOINT =
+ "client.refresh-credentials-endpoint";
@VisibleForTesting static final String ICEBERG_OSS_ACCESS_KEY_ID =
"client.access-key-id";
@VisibleForTesting static final String ICEBERG_OSS_ACCESS_KEY_SECRET =
"client.access-key-secret";
@VisibleForTesting static final String ICEBERG_OSS_SECURITY_TOKEN =
"client.security-token";
+ @VisibleForTesting
+ static final String ICEBERG_OSS_SECURITY_TOKEN_EXPIRES_AT_MS =
+ "client.security-token-expires-at-ms";
+
@VisibleForTesting static final String ICEBERG_ADLS_TOKEN = "adls.sas-token";
@VisibleForTesting
@@ -49,7 +60,21 @@ public class CredentialPropertyUtils {
@VisibleForTesting
static final String ICEBERG_ADLS_ACCOUNT_KEY =
"adls.auth.shared-key.account.key";
- private static final String GCS_OAUTH_2_TOKEN_EXPIRES_AT =
"gcs.oauth2.token-expires-at";
+ @VisibleForTesting
+ static final String ICEBERG_ADLS_SAS_TOKEN_EXPIRES_AT_MS_PREFIX =
"adls.sas-token-expires-at-ms.";
+
+ @VisibleForTesting
+ static final String ICEBERG_ADLS_REFRESH_CREDENTIALS_ENDPOINT =
+ "adls.refresh-credentials-endpoint";
+
+ @VisibleForTesting static final String ICEBERG_GCS_TOKEN =
"gcs.oauth2.token";
+
+ @VisibleForTesting
+ static final String ICEBERG_GCS_TOKEN_EXPIRES_AT =
"gcs.oauth2.token-expires-at";
+
+ @VisibleForTesting
+ static final String ICEBERG_GCS_OAUTH2_REFRESH_CREDENTIALS_ENDPOINT =
+ "gcs.oauth2.refresh-credentials-endpoint";
private static Map<String, String> icebergCredentialPropertyMap =
ImmutableMap.<String, String>builder()
@@ -82,36 +107,40 @@ public class CredentialPropertyUtils {
* @return a map of Iceberg properties derived from the credential
*/
public static Map<String, String> toIcebergProperties(Credential credential)
{
- if (credential instanceof S3TokenCredential
- || credential instanceof S3SecretKeyCredential
- || credential instanceof OSSTokenCredential
- || credential instanceof OSSSecretKeyCredential
- || credential instanceof AzureAccountKeyCredential
- || credential instanceof AwsIrsaCredential) {
- return transformProperties(credential.credentialInfo(),
icebergCredentialPropertyMap);
- }
-
- if (credential instanceof GCSTokenCredential) {
- Map<String, String> icebergGCSCredentialProperties =
- transformProperties(credential.credentialInfo(),
icebergCredentialPropertyMap);
- icebergGCSCredentialProperties.put(
- GCS_OAUTH_2_TOKEN_EXPIRES_AT,
String.valueOf(credential.expireTimeInMs()));
- return icebergGCSCredentialProperties;
- }
-
if (credential instanceof ADLSTokenCredential) {
ADLSTokenCredential adlsCredential = (ADLSTokenCredential) credential;
- String sasTokenKey =
- String.format(
- "%s.%s.%s",
- ICEBERG_ADLS_TOKEN, adlsCredential.accountName(),
ADLSTokenCredential.ADLS_DOMAIN);
+ String adlsHost = adlsCredential.accountName() + "." +
ADLSTokenCredential.ADLS_DOMAIN;
Map<String, String> icebergADLSCredentialProperties = new HashMap<>();
- icebergADLSCredentialProperties.put(sasTokenKey,
adlsCredential.sasToken());
+ icebergADLSCredentialProperties.put(
+ ICEBERG_ADLS_TOKEN + "." + adlsHost, adlsCredential.sasToken());
+ icebergADLSCredentialProperties.put(
+ ICEBERG_ADLS_SAS_TOKEN_EXPIRES_AT_MS_PREFIX + adlsHost,
+ String.valueOf(adlsCredential.expireTimeInMs()));
return icebergADLSCredentialProperties;
}
- return credential.toProperties();
+ Map<String, String> icebergProperties =
+ transformProperties(credential.credentialInfo(),
icebergCredentialPropertyMap);
+ if (credential instanceof S3TokenCredential || credential instanceof
AwsIrsaCredential) {
+ icebergProperties.put(
+ ICEBERG_S3_TOKEN_EXPIRES_AT_MS,
String.valueOf(credential.expireTimeInMs()));
+ return icebergProperties;
+ } else if (credential instanceof OSSTokenCredential) {
+ icebergProperties.put(
+ ICEBERG_OSS_SECURITY_TOKEN_EXPIRES_AT_MS,
String.valueOf(credential.expireTimeInMs()));
+ return icebergProperties;
+ } else if (credential instanceof GCSTokenCredential) {
+ icebergProperties.put(
+ ICEBERG_GCS_TOKEN_EXPIRES_AT,
String.valueOf(credential.expireTimeInMs()));
+ return icebergProperties;
+ } else if (credential instanceof S3SecretKeyCredential
+ || credential instanceof OSSSecretKeyCredential
+ || credential instanceof AzureAccountKeyCredential) {
+ return icebergProperties;
+ } else {
+ return credential.toProperties();
+ }
}
/**
@@ -126,20 +155,69 @@ public class CredentialPropertyUtils {
*/
public static Map<String, String> filterCredentialProperties(Map<String,
String> properties) {
Set<String> credentialPropertyKeys =
Sets.newHashSet(icebergCredentialPropertyMap.values());
- credentialPropertyKeys.add(GCS_OAUTH_2_TOKEN_EXPIRES_AT);
+ credentialPropertyKeys.add(ICEBERG_S3_TOKEN_EXPIRES_AT_MS);
+ credentialPropertyKeys.add(ICEBERG_OSS_SECURITY_TOKEN_EXPIRES_AT_MS);
+ credentialPropertyKeys.add(ICEBERG_GCS_TOKEN_EXPIRES_AT);
Map<String, String> filteredProperties = Maps.newHashMap(properties);
filteredProperties
.entrySet()
.removeIf(
entry ->
!credentialPropertyKeys.contains(entry.getKey())
- && !entry.getKey().startsWith(ICEBERG_ADLS_TOKEN));
+ && !entry.getKey().startsWith(ICEBERG_ADLS_TOKEN)
+ &&
!entry.getKey().startsWith(ICEBERG_ADLS_SAS_TOKEN_EXPIRES_AT_MS_PREFIX));
return filteredProperties;
}
+ /**
+ * Builds refresh credential endpoint properties for Iceberg credential
properties.
+ *
+ * @param encodedCatalogName Iceberg REST encoded catalog name
+ * @param encodedNamespace Iceberg REST encoded namespace
+ * @param encodedTableName Iceberg REST encoded table name
+ * @param credentialProperties Iceberg credential properties used to
determine refresh keys
+ * @return refresh endpoint properties keyed by Iceberg client config names
+ */
+ public static Map<String, String> buildRefreshProps(
+ String encodedCatalogName,
+ String encodedNamespace,
+ String encodedTableName,
+ Map<String, String> credentialProperties) {
+ if (credentialProperties == null || credentialProperties.isEmpty()) {
+ return Collections.emptyMap();
+ }
+
+ String credentialsRefreshPath =
+ String.format(
+ "v1/%s/namespaces/%s/tables/%s/credentials",
+ encodedCatalogName, encodedNamespace, encodedTableName);
+ Map<String, String> refreshEndpointProperties = Maps.newHashMap();
+ if (credentialProperties.containsKey(ICEBERG_S3_TOKEN)) {
+ refreshEndpointProperties.put(
+ ICEBERG_CLIENT_REFRESH_CREDENTIALS_ENDPOINT, credentialsRefreshPath);
+ }
+ if (credentialProperties.containsKey(ICEBERG_GCS_TOKEN)) {
+ refreshEndpointProperties.put(
+ ICEBERG_GCS_OAUTH2_REFRESH_CREDENTIALS_ENDPOINT,
credentialsRefreshPath);
+ }
+ if (credentialProperties.keySet().stream()
+ .anyMatch(key -> key.startsWith(ICEBERG_ADLS_TOKEN + "."))) {
+ refreshEndpointProperties.put(
+ ICEBERG_ADLS_REFRESH_CREDENTIALS_ENDPOINT, credentialsRefreshPath);
+ }
+ return ImmutableMap.copyOf(refreshEndpointProperties);
+ }
+
+ /**
+ * Transforms Gravitino credential keys to Iceberg property keys.
+ *
+ * @param originProperties the source credential info map
+ * @param transformMap mapping from Gravitino keys to Iceberg keys
+ * @return a new mutable map of transformed Iceberg properties
+ */
private static Map<String, String> transformProperties(
Map<String, String> originProperties, Map<String, String> transformMap) {
- HashMap<String, String> properties = new HashMap();
+ Map<String, String> properties = new HashMap<>();
originProperties.forEach(
(k, v) -> {
if (transformMap.containsKey(k)) {
diff --git
a/common/src/test/java/org/apache/gravitino/credential/TestCredentialFactory.java
b/common/src/test/java/org/apache/gravitino/credential/TestCredentialFactory.java
index 443399daf6..259949ae2b 100644
---
a/common/src/test/java/org/apache/gravitino/credential/TestCredentialFactory.java
+++
b/common/src/test/java/org/apache/gravitino/credential/TestCredentialFactory.java
@@ -26,6 +26,30 @@ import org.junit.jupiter.api.Test;
public class TestCredentialFactory {
+ @Test
+ void testAwsIrsaCredential() {
+ Map<String, String> awsIrsaCredentialInfo =
+ ImmutableMap.of(
+ AwsIrsaCredential.ACCESS_KEY_ID,
+ "accessKeyId",
+ AwsIrsaCredential.SECRET_ACCESS_KEY,
+ "secretAccessKey",
+ AwsIrsaCredential.SESSION_TOKEN,
+ "token");
+ long expireTime = 1000;
+ Credential awsIrsaCredential =
+ CredentialFactory.create(
+ AwsIrsaCredential.AWS_IRSA_CREDENTIAL_TYPE, awsIrsaCredentialInfo,
expireTime);
+ Assertions.assertEquals(
+ AwsIrsaCredential.AWS_IRSA_CREDENTIAL_TYPE,
awsIrsaCredential.credentialType());
+ Assertions.assertInstanceOf(AwsIrsaCredential.class, awsIrsaCredential);
+ AwsIrsaCredential awsIrsaCredential1 = (AwsIrsaCredential)
awsIrsaCredential;
+ Assertions.assertEquals("accessKeyId", awsIrsaCredential1.accessKeyId());
+ Assertions.assertEquals("secretAccessKey",
awsIrsaCredential1.secretAccessKey());
+ Assertions.assertEquals("token", awsIrsaCredential1.sessionToken());
+ Assertions.assertEquals(expireTime, awsIrsaCredential1.expireTimeInMs());
+ }
+
@Test
void testS3TokenCredential() {
Map<String, String> s3TokenCredentialInfo =
diff --git
a/common/src/test/java/org/apache/gravitino/credential/TestCredentialPropertiesUtils.java
b/common/src/test/java/org/apache/gravitino/credential/TestCredentialPropertiesUtils.java
index cb5eaabe76..0ee13adac5 100644
---
a/common/src/test/java/org/apache/gravitino/credential/TestCredentialPropertiesUtils.java
+++
b/common/src/test/java/org/apache/gravitino/credential/TestCredentialPropertiesUtils.java
@@ -38,7 +38,9 @@ public class TestCredentialPropertiesUtils {
CredentialPropertyUtils.ICEBERG_S3_SECRET_ACCESS_KEY,
"secret",
CredentialPropertyUtils.ICEBERG_S3_TOKEN,
- "token");
+ "token",
+ CredentialPropertyUtils.ICEBERG_S3_TOKEN_EXPIRES_AT_MS,
+ "100");
Assertions.assertEquals(expectedProperties, icebergProperties);
S3SecretKeyCredential secretKeyCredential = new
S3SecretKeyCredential("key", "secret");
@@ -65,7 +67,9 @@ public class TestCredentialPropertiesUtils {
CredentialPropertyUtils.ICEBERG_OSS_ACCESS_KEY_SECRET,
"secret",
CredentialPropertyUtils.ICEBERG_OSS_SECURITY_TOKEN,
- "security-token");
+ "security-token",
+ CredentialPropertyUtils.ICEBERG_OSS_SECURITY_TOKEN_EXPIRES_AT_MS,
+ "100");
Assertions.assertEquals(expectedProperties, icebergProperties);
}
@@ -80,14 +84,67 @@ public class TestCredentialPropertiesUtils {
Map<String, String> icebergProperties =
CredentialPropertyUtils.toIcebergProperties(adlsTokenCredential);
- String sasTokenKey =
- String.format(
- "%s.%s.%s",
- CredentialPropertyUtils.ICEBERG_ADLS_TOKEN,
- storageAccountName,
- ADLSTokenCredential.ADLS_DOMAIN);
+ String adlsHost = storageAccountName + "." +
ADLSTokenCredential.ADLS_DOMAIN;
+ String sasTokenKey = CredentialPropertyUtils.ICEBERG_ADLS_TOKEN + "." +
adlsHost;
- Map<String, String> expectedProperties = ImmutableMap.of(sasTokenKey,
sasToken);
+ Map<String, String> expectedProperties =
+ ImmutableMap.of(
+ sasTokenKey,
+ sasToken,
+
CredentialPropertyUtils.ICEBERG_ADLS_SAS_TOKEN_EXPIRES_AT_MS_PREFIX + adlsHost,
+ String.valueOf(expireTimeInMS));
Assertions.assertEquals(expectedProperties, icebergProperties);
}
+
+ @Test
+ void testBuildRefreshEndpointsForS3() {
+ Map<String, String> credentialProperties =
+ ImmutableMap.of(
+ CredentialPropertyUtils.ICEBERG_S3_TOKEN, "token",
+ CredentialPropertyUtils.ICEBERG_S3_TOKEN_EXPIRES_AT_MS, "123");
+
+ Map<String, String> refreshEndpointProperties =
+ CredentialPropertyUtils.buildRefreshProps("irc1", "db", "tbl",
credentialProperties);
+
+ Assertions.assertEquals(
+ "v1/irc1/namespaces/db/tables/tbl/credentials",
+ refreshEndpointProperties.get(
+
CredentialPropertyUtils.ICEBERG_CLIENT_REFRESH_CREDENTIALS_ENDPOINT));
+ }
+
+ @Test
+ void testBuildRefreshEndpointsForGCS() {
+ Map<String, String> credentialProperties =
+ ImmutableMap.of(
+ CredentialPropertyUtils.ICEBERG_GCS_TOKEN, "token",
+ CredentialPropertyUtils.ICEBERG_GCS_TOKEN_EXPIRES_AT, "123");
+
+ Map<String, String> refreshEndpointProperties =
+ CredentialPropertyUtils.buildRefreshProps("irc1", "db", "tbl",
credentialProperties);
+
+ Assertions.assertEquals(
+ "v1/irc1/namespaces/db/tables/tbl/credentials",
+ refreshEndpointProperties.get(
+
CredentialPropertyUtils.ICEBERG_GCS_OAUTH2_REFRESH_CREDENTIALS_ENDPOINT));
+ }
+
+ @Test
+ void testBuildRefreshEndpointsForADLS() {
+ String adlsHost = "storage.dfs.core.windows.net";
+ String sasTokenKey = CredentialPropertyUtils.ICEBERG_ADLS_TOKEN + "." +
adlsHost;
+ Map<String, String> credentialProperties =
+ ImmutableMap.of(
+ sasTokenKey,
+ "sas-token",
+
CredentialPropertyUtils.ICEBERG_ADLS_SAS_TOKEN_EXPIRES_AT_MS_PREFIX + adlsHost,
+ "123");
+
+ Map<String, String> refreshEndpointProperties =
+ CredentialPropertyUtils.buildRefreshProps("irc1", "db", "tbl",
credentialProperties);
+
+ Assertions.assertEquals(
+ "v1/irc1/namespaces/db/tables/tbl/credentials",
+ refreshEndpointProperties.get(
+
CredentialPropertyUtils.ICEBERG_ADLS_REFRESH_CREDENTIALS_ENDPOINT));
+ }
}
diff --git
a/core/src/main/java/org/apache/gravitino/credential/CatalogCredentialManager.java
b/core/src/main/java/org/apache/gravitino/credential/CatalogCredentialManager.java
index 88892391c4..9fb5a9eecf 100644
---
a/core/src/main/java/org/apache/gravitino/credential/CatalogCredentialManager.java
+++
b/core/src/main/java/org/apache/gravitino/credential/CatalogCredentialManager.java
@@ -49,6 +49,15 @@ public class CatalogCredentialManager implements Closeable {
credentialCache.initialize(catalogProperties);
}
+ /**
+ * Returns the catalog name this manager was created for.
+ *
+ * @return catalog name
+ */
+ public String catalogName() {
+ return catalogName;
+ }
+
public Optional<Credential> getCredential(String credentialType,
CredentialContext context) {
CredentialCacheKey credentialCacheKey = new
CredentialCacheKey(credentialType, context);
return credentialCache.getCredential(credentialCacheKey, cacheKey ->
doGetCredential(cacheKey));
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java
index 39fc375d7b..7e1aac0c0d 100644
---
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java
@@ -57,6 +57,7 @@ import org.apache.gravitino.utils.PrincipalUtils;
import org.apache.iceberg.BaseMetadataTable;
import org.apache.iceberg.BaseTable;
import org.apache.iceberg.BaseTransaction;
+import org.apache.iceberg.CatalogProperties;
import org.apache.iceberg.FileScanTask;
import org.apache.iceberg.IncrementalAppendScan;
import org.apache.iceberg.MetadataUpdate;
@@ -81,8 +82,16 @@ import org.apache.iceberg.inmemory.InMemoryFileIO;
import org.apache.iceberg.io.CloseableIterable;
import org.apache.iceberg.io.FileIO;
import org.apache.iceberg.rest.CatalogHandlers;
+import org.apache.iceberg.rest.ErrorHandlers;
+import org.apache.iceberg.rest.HTTPClient;
import org.apache.iceberg.rest.PlanStatus;
import org.apache.iceberg.rest.RESTCatalog;
+import org.apache.iceberg.rest.RESTClient;
+import org.apache.iceberg.rest.RESTUtil;
+import org.apache.iceberg.rest.ResourcePaths;
+import org.apache.iceberg.rest.auth.AuthManager;
+import org.apache.iceberg.rest.auth.AuthManagers;
+import org.apache.iceberg.rest.auth.AuthSession;
import org.apache.iceberg.rest.requests.CreateTableRequest;
import org.apache.iceberg.rest.requests.PlanTableScanRequest;
import org.apache.iceberg.rest.requests.RegisterTableRequest;
@@ -192,38 +201,92 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
/**
* Get table credentials.
*
- * @param identifier The table identifier for which to load credentials
- * @return A {@link
org.apache.iceberg.rest.responses.LoadCredentialsResponse} object containing
- * the credentials.
+ * @param identifier table identifier
+ * @param privilege used for local credential vending; ignored for REST
catalog backends
+ * @return table credentials response
*/
public LoadCredentialsResponse getTableCredentials(
TableIdentifier identifier, CredentialPrivilege privilege) {
+ if (isRESTCatalog()) {
+ return getRESTTableCredentials((RESTCatalog) getCatalog(), identifier);
+ } else {
+ return getLocalTableCredentials(identifier, privilege);
+ }
+ }
+
+ private LoadCredentialsResponse getLocalTableCredentials(
+ TableIdentifier identifier, CredentialPrivilege privilege) {
try {
LoadTableResponse loadTableResponse = super.loadTable(identifier);
- Credential credential = getCredential(loadTableResponse, privilege);
- org.apache.iceberg.rest.credentials.Credential icebergCredential =
- new org.apache.iceberg.rest.credentials.Credential() {
- @Override
- public String prefix() {
- return "";
- }
-
- @Override
- public Map<String, String> config() {
- // Convert Gravitino credentials to the Iceberg REST credential
payload format.
- return CredentialPropertyUtils.toIcebergProperties(credential);
- }
-
- @Override
- public void validate() {}
- };
- return
ImmutableLoadCredentialsResponse.builder().addCredentials(icebergCredential).build();
+ Credential credential = getCredential(loadTableResponse.tableMetadata(),
privilege);
+ return ImmutableLoadCredentialsResponse.builder()
+ .addCredentials(
+ IcebergRESTUtils.toRESTCredential(
+ catalogCredentialManager.catalogName(),
+ identifier,
+ credential,
+ loadTableResponse.tableMetadata()))
+ .build();
} catch (ServiceUnavailableException e) {
LOG.warn("Service unavailable when loading table credentials for table:
{}", identifier, e);
return ImmutableLoadCredentialsResponse.builder().build();
}
}
+ private static LoadCredentialsResponse getRESTTableCredentials(
+ RESTCatalog restCatalog, TableIdentifier identifier) {
+ Map<String, String> properties = Maps.newHashMap(restCatalog.properties());
+ String credentialsPath =
+ ResourcePaths.forCatalogProperties(properties).table(identifier) +
"/credentials";
+
+ AuthManager authManager = null;
+ RESTClient client = null;
+ AuthSession authSession = null;
+ try {
+ authManager = AuthManagers.loadAuthManager(restCatalog.name(),
properties);
+ client =
+ HTTPClient.builder(properties)
+ .uri(properties.get(CatalogProperties.URI))
+ .withHeaders(RESTUtil.configHeaders(properties))
+ .build();
+ authSession = authManager.catalogSession(client, properties);
+ return client
+ .withAuthSession(authSession)
+ .get(
+ credentialsPath,
+ LoadCredentialsResponse.class,
+ Collections.emptyMap(),
+ ErrorHandlers.tableErrorHandler());
+ } finally {
+ if (authSession != null) {
+ try {
+ authSession.close();
+ } catch (Exception e) {
+ LOG.warn(
+ "Failed to close auth session when loading credentials for
table: {}", identifier, e);
+ }
+ }
+
+ if (client != null) {
+ try {
+ client.close();
+ } catch (Exception e) {
+ LOG.warn(
+ "Failed to close REST client when loading credentials for table:
{}", identifier, e);
+ }
+ }
+
+ if (authManager != null) {
+ try {
+ authManager.close();
+ } catch (Exception e) {
+ LOG.warn(
+ "Failed to close auth manager when loading credentials for
table: {}", identifier, e);
+ }
+ }
+ }
+ }
+
@Override
public void close() throws Exception {
try {
@@ -318,7 +381,7 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
TableIdentifier tableIdentifier,
LoadTableResponse loadTableResponse,
CredentialPrivilege privilege) {
- final Credential credential = getCredential(loadTableResponse, privilege);
+ final Credential credential =
getCredential(loadTableResponse.tableMetadata(), privilege);
LOG.info(
"Generate credential: {} for Iceberg table: {}",
@@ -327,7 +390,13 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
// Merge temporary credential fields as Iceberg client config entries in
the load-table
// response.
- Map<String, String> credentialConfig =
CredentialPropertyUtils.toIcebergProperties(credential);
+ Map<String, String> credentialConfig =
+ IcebergRESTUtils.toRESTCredential(
+ catalogCredentialManager.catalogName(),
+ tableIdentifier,
+ credential,
+ loadTableResponse.tableMetadata())
+ .config();
return LoadTableResponse.builder()
.withTableMetadata(loadTableResponse.tableMetadata())
.addAllConfig(loadTableResponse.config())
@@ -336,9 +405,7 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
.build();
}
- private Credential getCredential(
- LoadTableResponse loadTableResponse, CredentialPrivilege privilege) {
- TableMetadata tableMetadata = loadTableResponse.tableMetadata();
+ private Credential getCredential(TableMetadata tableMetadata,
CredentialPrivilege privilege) {
String[] path =
Stream.of(
tableMetadata.location(),
@@ -685,13 +752,22 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
if (table instanceof BaseTable) {
Map<String, String> properties = retrieveFileIOProperties(table.io());
+ Map<String, String> filteredCredentialProperties =
+ CredentialPropertyUtils.filterCredentialProperties(properties);
return LoadTableResponse.builder()
.withTableMetadata(((BaseTable) table).operations().current())
.addAllConfig(
MapUtils.getFilteredMap(
properties, key ->
catalogPropertiesToClientKeys.contains(key)))
- // Keep only credential fields from FileIO properties before
returning them to the client.
-
.addAllConfig(CredentialPropertyUtils.filterCredentialProperties(properties))
+ // Keep only credential fields from FileIO properties before
returning them to the
+ // client.
+ .addAllConfig(filteredCredentialProperties)
+ .addAllConfig(
+ IcebergRESTUtils.buildRefreshProps(
+ catalogCredentialManager.catalogName(), ident,
filteredCredentialProperties))
+ .addAllCredentials(
+ IcebergRESTUtils.buildStorageCreds(
+ catalogCredentialManager.catalogName(), ident, table.io()))
.build();
}
@@ -726,10 +802,19 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
}
Map<String, String> tableProperties = retrieveFileIOProperties(table.io());
+ Map<String, String> filteredCredentialProperties =
+ CredentialPropertyUtils.filterCredentialProperties(tableProperties);
config.putAll(
MapUtils.getFilteredMap(
tableProperties, key ->
catalogPropertiesToClientKeys.contains(key)));
-
config.putAll(CredentialPropertyUtils.filterCredentialProperties(tableProperties));
+ config.putAll(filteredCredentialProperties);
+ config.putAll(
+ IcebergRESTUtils.buildRefreshProps(
+ catalogCredentialManager.catalogName(), ident,
filteredCredentialProperties));
+
+ List<org.apache.iceberg.rest.credentials.Credential> credentials =
+ IcebergRESTUtils.buildStorageCreds(
+ catalogCredentialManager.catalogName(), ident, table.io());
TableMetadata metadata =
TableMetadata.newTableMetadata(
@@ -739,7 +824,11 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
table.location(),
properties);
- return
LoadTableResponse.builder().withTableMetadata(metadata).addAllConfig(config).build();
+ return LoadTableResponse.builder()
+ .withTableMetadata(metadata)
+ .addAllConfig(config)
+ .addAllCredentials(credentials)
+ .build();
}
private LoadTableResponse tableUpdateInternal(TableIdentifier ident,
UpdateTableRequest request) {
@@ -793,13 +882,21 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
if (table instanceof BaseTable) {
Map<String, String> properties = retrieveFileIOProperties(table.io());
+ Map<String, String> filteredCredentialProperties =
+ CredentialPropertyUtils.filterCredentialProperties(properties);
return LoadTableResponse.builder()
.withTableMetadata(((BaseTable) table).operations().current())
.addAllConfig(
MapUtils.getFilteredMap(
properties, key ->
catalogPropertiesToClientKeys.contains(key)))
// Keep only credential fields from FileIO properties before
returning them to the client.
-
.addAllConfig(CredentialPropertyUtils.filterCredentialProperties(properties))
+ .addAllConfig(filteredCredentialProperties)
+ .addAllConfig(
+ IcebergRESTUtils.buildRefreshProps(
+ catalogCredentialManager.catalogName(), ident,
filteredCredentialProperties))
+ .addAllCredentials(
+ IcebergRESTUtils.buildStorageCreds(
+ catalogCredentialManager.catalogName(), ident, table.io()))
.build();
} else if (table instanceof BaseMetadataTable) {
// metadata tables are loaded on the client side, return
NoSuchTableException for now
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergRESTUtils.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergRESTUtils.java
index c095ac231b..465ebfc138 100644
---
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergRESTUtils.java
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergRESTUtils.java
@@ -20,6 +20,7 @@ package org.apache.gravitino.iceberg.service;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
@@ -27,9 +28,12 @@ import java.security.NoSuchAlgorithmException;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
+import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collections;
import java.util.Enumeration;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.stream.Stream;
@@ -40,9 +44,15 @@ import javax.ws.rs.core.Response;
import javax.ws.rs.core.Response.Status;
import org.apache.commons.lang3.StringUtils;
import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.credential.Credential;
+import org.apache.gravitino.credential.CredentialPropertyUtils;
import
org.apache.gravitino.iceberg.service.authorization.IcebergRESTServerContext;
+import org.apache.iceberg.TableMetadata;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.StorageCredential;
+import org.apache.iceberg.io.SupportsStorageCredentials;
import org.apache.iceberg.rest.RESTUtil;
import org.apache.iceberg.rest.responses.ErrorResponse;
import org.apache.iceberg.rest.responses.LoadTableResponse;
@@ -83,6 +93,101 @@ public class IcebergRESTUtils {
private IcebergRESTUtils() {}
+ /**
+ * Builds an Iceberg REST {@link
org.apache.iceberg.rest.credentials.Credential} for load-table,
+ * scan-plan, or credentials API responses.
+ *
+ * <p>Refresh credential endpoints are added only for credential types whose
Iceberg client
+ * modules support vended credential refresh: S3 ({@link
+ * org.apache.gravitino.credential.S3TokenCredential}, {@link
+ * org.apache.gravitino.credential.AwsIrsaCredential}), GCS ({@link
+ * org.apache.gravitino.credential.GCSTokenCredential}), and ADLS ({@link
+ * org.apache.gravitino.credential.ADLSTokenCredential}). OSS temporary
credentials ({@link
+ * org.apache.gravitino.credential.OSSTokenCredential}) are omitted because
Iceberg's Aliyun OSS
+ * FileIO does not consume {@code client.refresh-credentials-endpoint}; only
the initial STS
+ * properties are returned.
+ *
+ * @param catalogName IRC catalog name used in the refresh path
+ * @param tableIdentifier table receiving the credential
+ * @param credential Gravitino credential to vend
+ * @param tableMetadata table metadata used to derive the storage prefix
+ * @return Iceberg REST credential with prefix and config
+ */
+ public static org.apache.iceberg.rest.credentials.Credential
toRESTCredential(
+ String catalogName,
+ TableIdentifier tableIdentifier,
+ Credential credential,
+ TableMetadata tableMetadata) {
+ Map<String, String> config =
+ new HashMap<>(CredentialPropertyUtils.toIcebergProperties(credential));
+ config.putAll(buildRefreshProps(catalogName, tableIdentifier, config));
+
+ return toRESTCredential(tableMetadata.location(), config);
+ }
+
+ /**
+ * Builds an Iceberg REST {@link
org.apache.iceberg.rest.credentials.Credential} from a storage
+ * prefix and credential config map.
+ *
+ * @param prefix storage location prefix for the credential
+ * @param config Iceberg credential config properties
+ * @return Iceberg REST credential
+ */
+ public static org.apache.iceberg.rest.credentials.Credential
toRESTCredential(
+ String prefix, Map<String, String> config) {
+ Map<String, String> credentialConfig = ImmutableMap.copyOf(config);
+ return new org.apache.iceberg.rest.credentials.Credential() {
+ @Override
+ public String prefix() {
+ return prefix;
+ }
+
+ @Override
+ public Map<String, String> config() {
+ return credentialConfig;
+ }
+
+ @Override
+ public void validate() {}
+ };
+ }
+
+ /**
+ * Builds Iceberg REST 1.11 {@code storage-credentials} from an upstream
REST catalog proxy.
+ *
+ * <p>Upstream credentials are read from {@link
SupportsStorageCredentials#credentials()}. When
+ * present, they are filtered and rewritten with IRC-local refresh endpoints
via {@link
+ * CredentialPropertyUtils}. Credential properties in {@link
FileIO#properties()} must be handled
+ * separately by the caller.
+ *
+ * @param catalogName IRC catalog name used to build refresh paths
+ * @param tableIdentifier table receiving the credentials
+ * @param fileIO table FileIO returned by the upstream catalog
+ * @return rewritten storage credentials for the downstream client, or an
empty list
+ */
+ public static List<org.apache.iceberg.rest.credentials.Credential>
buildStorageCreds(
+ String catalogName, TableIdentifier tableIdentifier, FileIO fileIO) {
+ if (!(fileIO instanceof SupportsStorageCredentials)) {
+ return Collections.emptyList();
+ }
+
+ List<StorageCredential> credentials = ((SupportsStorageCredentials)
fileIO).credentials();
+ if (credentials == null || credentials.isEmpty()) {
+ return Collections.emptyList();
+ }
+
+ List<org.apache.iceberg.rest.credentials.Credential> restCredentials = new
ArrayList<>();
+ for (StorageCredential credential : credentials) {
+ Map<String, String> filteredConfig =
+
CredentialPropertyUtils.filterCredentialProperties(credential.config());
+ filteredConfig.putAll(buildRefreshProps(catalogName, tableIdentifier,
filteredConfig));
+ restCredentials.add(
+ toRESTCredential(credential.prefix(),
ImmutableMap.copyOf(filteredConfig)));
+ }
+
+ return List.copyOf(restCredentials);
+ }
+
public static <T> Response ok(T t) {
return
Response.status(Response.Status.OK).entity(t).type(MediaType.APPLICATION_JSON).build();
}
@@ -262,6 +367,25 @@ public class IcebergRESTUtils {
return headers;
}
+ /**
+ * Builds refresh credential endpoint properties for a table, with Iceberg
REST path encoding.
+ *
+ * @param catalogName IRC catalog name used in the refresh path
+ * @param tableIdentifier table receiving the credentials
+ * @param credentialProperties Iceberg credential properties used to
determine refresh keys
+ * @return refresh endpoint properties keyed by Iceberg client config names
+ */
+ public static Map<String, String> buildRefreshProps(
+ String catalogName,
+ TableIdentifier tableIdentifier,
+ Map<String, String> credentialProperties) {
+ return CredentialPropertyUtils.buildRefreshProps(
+ RESTUtil.encodeString(catalogName),
+ RESTUtil.encodeNamespace(tableIdentifier.namespace(),
NAMESPACE_SEPARATOR_URLENCODED_UTF_8),
+ RESTUtil.encodeString(tableIdentifier.name()),
+ credentialProperties);
+ }
+
// remove the last '/' from the prefix, for example transform
'iceberg_catalog/' to
// 'iceberg_catalog'
private static String normalizePrefix(String rawPrefix) {
diff --git
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestCatalogWrapperForREST.java
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestCatalogWrapperForREST.java
index b7432cdc31..55d30018a0 100644
---
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestCatalogWrapperForREST.java
+++
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestCatalogWrapperForREST.java
@@ -27,8 +27,13 @@ import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.withSettings;
import com.google.common.collect.ImmutableMap;
+import com.sun.net.httpserver.HttpServer;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
@@ -38,8 +43,13 @@ import java.util.UUID;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants;
+import org.apache.gravitino.credential.CredentialConstants;
+import org.apache.gravitino.credential.CredentialPrivilege;
import org.apache.gravitino.iceberg.common.IcebergConfig;
+import org.apache.gravitino.iceberg.service.extension.DummyCredentialProvider;
+import org.apache.iceberg.BaseTable;
import org.apache.iceberg.BaseTransaction;
+import org.apache.iceberg.CatalogProperties;
import org.apache.iceberg.MetadataUpdate;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
@@ -51,12 +61,22 @@ import org.apache.iceberg.Transaction;
import org.apache.iceberg.UpdateRequirement;
import org.apache.iceberg.catalog.Catalog;
import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.SupportsNamespaces;
import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.ForbiddenException;
+import org.apache.iceberg.exceptions.NoSuchTableException;
+import org.apache.iceberg.exceptions.NotAuthorizedException;
+import org.apache.iceberg.exceptions.ServiceFailureException;
import org.apache.iceberg.io.FileIO;
import org.apache.iceberg.io.ResolvingFileIO;
+import org.apache.iceberg.io.StorageCredential;
+import org.apache.iceberg.io.SupportsStorageCredentials;
import org.apache.iceberg.rest.RESTCatalog;
+import org.apache.iceberg.rest.auth.AuthProperties;
+import org.apache.iceberg.rest.credentials.Credential;
import org.apache.iceberg.rest.requests.CreateTableRequest;
import org.apache.iceberg.rest.requests.UpdateTableRequest;
+import org.apache.iceberg.rest.responses.LoadCredentialsResponse;
import org.apache.iceberg.rest.responses.LoadTableResponse;
import org.apache.iceberg.types.Types;
import org.junit.jupiter.api.Assertions;
@@ -101,6 +121,292 @@ public class TestCatalogWrapperForREST {
Assertions.assertFalse(CatalogWrapperForREST.isLocalOrHdfsLocation(" "));
}
+ @Test
+ void testRESTTableCredentials() throws Exception {
+ TableIdentifier table = TableIdentifier.of(Namespace.of("db"), "tbl");
+ String expectedPath = "/v1/upstream/namespaces/db/tables/tbl/credentials";
+ String upstreamJson =
+
"{\"storage-credentials\":[{\"prefix\":\"s3://upstream/db/tbl/\",\"config\":{"
+ + "\"s3.access-key-id\":\"upstream-key\","
+ + "\"s3.secret-access-key\":\"upstream-secret\","
+ + "\"s3.session-token\":\"upstream-token\"}}]}";
+
+ AtomicReference<String> requestPath = new AtomicReference<>();
+ HttpServer server = HttpServer.create(new InetSocketAddress(0), 0);
+ server.createContext(
+ "/",
+ exchange -> {
+ requestPath.set(exchange.getRequestURI().getPath());
+ byte[] body = upstreamJson.getBytes(StandardCharsets.UTF_8);
+ exchange.getResponseHeaders().add("Content-Type",
"application/json");
+ exchange.sendResponseHeaders(200, body.length);
+ try (OutputStream os = exchange.getResponseBody()) {
+ os.write(body);
+ }
+ });
+ server.start();
+ try {
+ String uri = "http://127.0.0.1:"; + server.getAddress().getPort();
+ RESTCatalog restCatalog = mock(RESTCatalog.class);
+ when(restCatalog.name()).thenReturn("upstream");
+ when(restCatalog.properties())
+ .thenReturn(
+ ImmutableMap.of(
+ CatalogProperties.URI,
+ uri,
+ AuthProperties.AUTH_TYPE,
+ AuthProperties.AUTH_TYPE_NONE,
+ "prefix",
+ "upstream"));
+
+ IcebergConfig config =
+ new IcebergConfig(
+ ImmutableMap.of(
+ IcebergConstants.CATALOG_BACKEND,
+ "memory",
+ IcebergConstants.WAREHOUSE,
+ "/tmp/warehouse"));
+ CatalogWrapperForREST wrapper = new StaticCatalogWrapperForREST("local",
config, restCatalog);
+
+ LoadCredentialsResponse response =
+ wrapper.getTableCredentials(table, CredentialPrivilege.READ);
+
+ Assertions.assertEquals(expectedPath, requestPath.get());
+ Assertions.assertEquals(1, response.credentials().size());
+ Credential credential = response.credentials().get(0);
+ Assertions.assertEquals("s3://upstream/db/tbl/", credential.prefix());
+ Assertions.assertEquals("upstream-key",
credential.config().get("s3.access-key-id"));
+ Assertions.assertEquals("upstream-secret",
credential.config().get("s3.secret-access-key"));
+ Assertions.assertEquals("upstream-token",
credential.config().get("s3.session-token"));
+ } finally {
+ server.stop(0);
+ }
+ }
+
+ @Test
+ void testRESTTableCredentialsOnFailure() throws Exception {
+ HttpServer server = HttpServer.create(new InetSocketAddress(0), 0);
+ server.createContext(
+ "/",
+ exchange -> {
+ exchange.sendResponseHeaders(500, -1);
+ exchange.close();
+ });
+ server.start();
+ try {
+ String uri = "http://127.0.0.1:"; + server.getAddress().getPort();
+ RESTCatalog restCatalog = mock(RESTCatalog.class);
+ when(restCatalog.name()).thenReturn("upstream");
+ when(restCatalog.properties())
+ .thenReturn(
+ ImmutableMap.of(
+ CatalogProperties.URI,
+ uri,
+ AuthProperties.AUTH_TYPE,
+ AuthProperties.AUTH_TYPE_NONE,
+ "prefix",
+ "upstream"));
+
+ IcebergConfig config =
+ new IcebergConfig(
+ ImmutableMap.of(
+ IcebergConstants.CATALOG_BACKEND,
+ "memory",
+ IcebergConstants.WAREHOUSE,
+ "/tmp/warehouse"));
+ CatalogWrapperForREST wrapper = new StaticCatalogWrapperForREST("local",
config, restCatalog);
+
+ Assertions.assertThrows(
+ ServiceFailureException.class,
+ () ->
+ wrapper.getTableCredentials(
+ TableIdentifier.of(Namespace.of("db"), "tbl"),
CredentialPrivilege.READ));
+ } finally {
+ server.stop(0);
+ }
+ }
+
+ @Test
+ void testRESTTableCredentialsOnForbidden() throws Exception {
+ String errorJson =
+
"{\"error\":{\"message\":\"Forbidden\",\"type\":\"ForbiddenException\","
+ + "\"code\":403,\"stack\":[]}}";
+ HttpServer server = HttpServer.create(new InetSocketAddress(0), 0);
+ server.createContext(
+ "/",
+ exchange -> {
+ byte[] body = errorJson.getBytes(StandardCharsets.UTF_8);
+ exchange.getResponseHeaders().add("Content-Type",
"application/json");
+ exchange.sendResponseHeaders(403, body.length);
+ try (OutputStream os = exchange.getResponseBody()) {
+ os.write(body);
+ }
+ });
+ server.start();
+ try {
+ String uri = "http://127.0.0.1:"; + server.getAddress().getPort();
+ RESTCatalog restCatalog = mock(RESTCatalog.class);
+ when(restCatalog.name()).thenReturn("upstream");
+ when(restCatalog.properties())
+ .thenReturn(
+ ImmutableMap.of(
+ CatalogProperties.URI,
+ uri,
+ AuthProperties.AUTH_TYPE,
+ AuthProperties.AUTH_TYPE_NONE,
+ "prefix",
+ "upstream"));
+
+ IcebergConfig config =
+ new IcebergConfig(
+ ImmutableMap.of(
+ IcebergConstants.CATALOG_BACKEND,
+ "memory",
+ IcebergConstants.WAREHOUSE,
+ "/tmp/warehouse"));
+ CatalogWrapperForREST wrapper = new StaticCatalogWrapperForREST("local",
config, restCatalog);
+
+ Assertions.assertThrows(
+ ForbiddenException.class,
+ () ->
+ wrapper.getTableCredentials(
+ TableIdentifier.of(Namespace.of("db"), "tbl"),
CredentialPrivilege.READ));
+ } finally {
+ server.stop(0);
+ }
+ }
+
+ @Test
+ void testRESTTableCredentialsOnUnauthorized() throws Exception {
+ String errorJson =
+ "{\"error\":{\"message\":\"Not
authorized\",\"type\":\"NotAuthorizedException\","
+ + "\"code\":401,\"stack\":[]}}";
+ HttpServer server = HttpServer.create(new InetSocketAddress(0), 0);
+ server.createContext(
+ "/",
+ exchange -> {
+ byte[] body = errorJson.getBytes(StandardCharsets.UTF_8);
+ exchange.getResponseHeaders().add("Content-Type",
"application/json");
+ exchange.sendResponseHeaders(401, body.length);
+ try (OutputStream os = exchange.getResponseBody()) {
+ os.write(body);
+ }
+ });
+ server.start();
+ try {
+ String uri = "http://127.0.0.1:"; + server.getAddress().getPort();
+ RESTCatalog restCatalog = mock(RESTCatalog.class);
+ when(restCatalog.name()).thenReturn("upstream");
+ when(restCatalog.properties())
+ .thenReturn(
+ ImmutableMap.of(
+ CatalogProperties.URI,
+ uri,
+ AuthProperties.AUTH_TYPE,
+ AuthProperties.AUTH_TYPE_NONE,
+ "prefix",
+ "upstream"));
+
+ IcebergConfig config =
+ new IcebergConfig(
+ ImmutableMap.of(
+ IcebergConstants.CATALOG_BACKEND,
+ "memory",
+ IcebergConstants.WAREHOUSE,
+ "/tmp/warehouse"));
+ CatalogWrapperForREST wrapper = new StaticCatalogWrapperForREST("local",
config, restCatalog);
+
+ Assertions.assertThrows(
+ NotAuthorizedException.class,
+ () ->
+ wrapper.getTableCredentials(
+ TableIdentifier.of(Namespace.of("db"), "tbl"),
CredentialPrivilege.READ));
+ } finally {
+ server.stop(0);
+ }
+ }
+
+ @Test
+ void testRESTTableCredentialsNoSuchTable() throws Exception {
+ String errorJson =
+ "{\"error\":{\"message\":\"Table not
found\",\"type\":\"NoSuchTableException\","
+ + "\"code\":404,\"stack\":[]}}";
+ HttpServer server = HttpServer.create(new InetSocketAddress(0), 0);
+ server.createContext(
+ "/",
+ exchange -> {
+ byte[] body = errorJson.getBytes(StandardCharsets.UTF_8);
+ exchange.getResponseHeaders().add("Content-Type",
"application/json");
+ exchange.sendResponseHeaders(404, body.length);
+ try (OutputStream os = exchange.getResponseBody()) {
+ os.write(body);
+ }
+ });
+ server.start();
+ try {
+ String uri = "http://127.0.0.1:"; + server.getAddress().getPort();
+ RESTCatalog restCatalog = mock(RESTCatalog.class);
+ when(restCatalog.name()).thenReturn("upstream");
+ when(restCatalog.properties())
+ .thenReturn(
+ ImmutableMap.of(
+ CatalogProperties.URI,
+ uri,
+ AuthProperties.AUTH_TYPE,
+ AuthProperties.AUTH_TYPE_NONE,
+ "prefix",
+ "upstream"));
+
+ IcebergConfig config =
+ new IcebergConfig(
+ ImmutableMap.of(
+ IcebergConstants.CATALOG_BACKEND,
+ "memory",
+ IcebergConstants.WAREHOUSE,
+ "/tmp/warehouse"));
+ CatalogWrapperForREST wrapper = new StaticCatalogWrapperForREST("local",
config, restCatalog);
+
+ Assertions.assertThrows(
+ NoSuchTableException.class,
+ () ->
+ wrapper.getTableCredentials(
+ TableIdentifier.of(Namespace.of("db"), "missing"),
CredentialPrivilege.READ));
+ } finally {
+ server.stop(0);
+ }
+ }
+
+ @Test
+ void testLocalTableCredentials() {
+ IcebergConfig config =
+ new IcebergConfig(
+ ImmutableMap.of(
+ IcebergConstants.CATALOG_BACKEND,
+ "memory",
+ IcebergConstants.WAREHOUSE,
+ "/tmp/warehouse",
+ CredentialConstants.CREDENTIAL_PROVIDERS,
+ DummyCredentialProvider.DUMMY_CREDENTIAL_TYPE));
+
+ CatalogWrapperForREST wrapper = new CatalogWrapperForREST("local-catalog",
config);
+ Namespace namespace = Namespace.of("db");
+ Catalog catalog = wrapper.getCatalog();
+ ((SupportsNamespaces) catalog).createNamespace(namespace);
+ TableIdentifier table = TableIdentifier.of(namespace, "tbl");
+ Schema schema = new Schema(Types.NestedField.required(1, "id",
Types.IntegerType.get()));
+ catalog.createTable(
+ table,
+ schema,
+ PartitionSpec.unpartitioned(),
+ "s3://bucket/wh/db/tbl",
+ Collections.emptyMap());
+
+ LoadCredentialsResponse response = wrapper.getTableCredentials(table,
CredentialPrivilege.READ);
+
+ Assertions.assertEquals(1, response.credentials().size());
+ Assertions.assertEquals("s3://bucket/wh/db/tbl",
response.credentials().get(0).prefix());
+ }
+
@Test
void testValidateCredentialLocation() {
Assertions.assertDoesNotThrow(
@@ -116,7 +422,113 @@ public class TestCatalogWrapperForREST {
}
@Test
- void testRestCatalogClientConfigMergesRemote() {
+ void testLoadTableRefreshEndpoint() {
+ TableIdentifier ident = TableIdentifier.of(Namespace.of("db"), "tbl");
+ RESTCatalog catalog = mock(RESTCatalog.class);
+ BaseTable baseTable = mock(BaseTable.class);
+ TableOperations ops = mock(TableOperations.class);
+ FileIO fileIO = mock(FileIO.class);
+ TableMetadata metadata =
+ TableMetadata.newTableMetadata(
+ new Schema(Types.NestedField.required(1, "id",
Types.IntegerType.get())),
+ PartitionSpec.unpartitioned(),
+ SortOrder.unsorted(),
+ "s3://bucket/db/tbl",
+ Collections.emptyMap());
+
+ when(catalog.loadTable(ident)).thenReturn(baseTable);
+ when(baseTable.operations()).thenReturn(ops);
+ when(ops.current()).thenReturn(metadata);
+ when(baseTable.io()).thenReturn(fileIO);
+ when(fileIO.properties())
+ .thenReturn(
+ ImmutableMap.of(
+ "s3.session-token",
+ "token",
+ "s3.session-token-expires-at-ms",
+ "123",
+ "client.refresh-credentials-endpoint",
+ "v1/upstream/namespaces/db/tables/tbl/credentials"));
+
+ IcebergConfig config =
+ new IcebergConfig(
+ ImmutableMap.of(
+ IcebergConstants.CATALOG_BACKEND,
+ "memory",
+ IcebergConstants.WAREHOUSE,
+ "/tmp/warehouse"));
+ CatalogWrapperForREST wrapper = new StaticCatalogWrapperForREST("irc1",
config, catalog);
+
+ LoadTableResponse response = wrapper.loadTable(ident, false,
CredentialPrivilege.READ);
+
+ Assertions.assertEquals(
+ "v1/irc1/namespaces/db/tables/tbl/credentials",
+ response.config().get("client.refresh-credentials-endpoint"));
+ Assertions.assertEquals("token",
response.config().get("s3.session-token"));
+ }
+
+ @Test
+ void testLoadTableStorageCreds() {
+ TableIdentifier ident = TableIdentifier.of(Namespace.of("db"), "tbl");
+ RESTCatalog catalog = mock(RESTCatalog.class);
+ BaseTable baseTable = mock(BaseTable.class);
+ TableOperations ops = mock(TableOperations.class);
+ FileIO fileIO =
+ mock(FileIO.class,
withSettings().extraInterfaces(SupportsStorageCredentials.class));
+ SupportsStorageCredentials storageCredentialsFileIO =
(SupportsStorageCredentials) fileIO;
+ TableMetadata metadata =
+ TableMetadata.newTableMetadata(
+ new Schema(Types.NestedField.required(1, "id",
Types.IntegerType.get())),
+ PartitionSpec.unpartitioned(),
+ SortOrder.unsorted(),
+ "s3://bucket/db/tbl",
+ Collections.emptyMap());
+
+ StorageCredential upstreamCredential =
+ StorageCredential.create(
+ "s3://bucket/db/tbl/",
+ ImmutableMap.of(
+ "s3.access-key-id",
+ "upstream-key",
+ "s3.secret-access-key",
+ "upstream-secret",
+ "s3.session-token",
+ "upstream-token",
+ "s3.session-token-expires-at-ms",
+ "123",
+ "client.refresh-credentials-endpoint",
+ "v1/upstream/namespaces/db/tables/tbl/credentials"));
+
+ when(catalog.loadTable(ident)).thenReturn(baseTable);
+ when(baseTable.operations()).thenReturn(ops);
+ when(ops.current()).thenReturn(metadata);
+ when(baseTable.io()).thenReturn(fileIO);
+ when(fileIO.properties()).thenReturn(Collections.emptyMap());
+
when(storageCredentialsFileIO.credentials()).thenReturn(List.of(upstreamCredential));
+
+ IcebergConfig config =
+ new IcebergConfig(
+ ImmutableMap.of(
+ IcebergConstants.CATALOG_BACKEND,
+ "memory",
+ IcebergConstants.WAREHOUSE,
+ "/tmp/warehouse"));
+ CatalogWrapperForREST wrapper = new StaticCatalogWrapperForREST("irc1",
config, catalog);
+
+ LoadTableResponse response = wrapper.loadTable(ident, false,
CredentialPrivilege.READ);
+
+ Assertions.assertEquals(1, response.credentials().size());
+ Credential credential = response.credentials().get(0);
+ Assertions.assertEquals("s3://bucket/db/tbl/", credential.prefix());
+ Assertions.assertEquals("upstream-token",
credential.config().get("s3.session-token"));
+ Assertions.assertEquals(
+ "v1/irc1/namespaces/db/tables/tbl/credentials",
+ credential.config().get("client.refresh-credentials-endpoint"));
+
Assertions.assertFalse(response.config().containsKey("client.refresh-credentials-endpoint"));
+ }
+
+ @Test
+ void testRESTCatalogClientConfigMergesRemote() {
IcebergConfig config =
new IcebergConfig(
ImmutableMap.of(
@@ -172,7 +584,7 @@ public class TestCatalogWrapperForREST {
}
@Test
- void testNonRestCatalogClientConfig() {
+ void testNonRESTCatalogClientConfig() {
Catalog catalog = mock(Catalog.class);
IcebergConfig config =
new IcebergConfig(
diff --git
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergRESTUtils.java
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergRESTUtils.java
index 7c2e05c80b..f7dd7cb2f0 100644
---
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergRESTUtils.java
+++
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergRESTUtils.java
@@ -19,12 +19,31 @@
package org.apache.gravitino.iceberg.service;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.withSettings;
+
+import com.google.common.collect.ImmutableMap;
+import java.util.List;
+import java.util.Map;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants;
+import org.apache.gravitino.credential.ADLSTokenCredential;
+import org.apache.gravitino.credential.AwsIrsaCredential;
+import org.apache.gravitino.credential.GCSTokenCredential;
+import org.apache.gravitino.credential.OSSTokenCredential;
+import org.apache.gravitino.credential.S3SecretKeyCredential;
+import org.apache.gravitino.credential.S3TokenCredential;
import
org.apache.gravitino.iceberg.service.authorization.IcebergRESTServerContext;
import org.apache.gravitino.iceberg.service.provider.IcebergConfigProvider;
import org.apache.iceberg.Schema;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.StorageCredential;
+import org.apache.iceberg.io.SupportsStorageCredentials;
+import org.apache.iceberg.rest.credentials.Credential;
import org.apache.iceberg.rest.requests.CreateTableRequest;
import org.apache.iceberg.types.Types.IntegerType;
import org.apache.iceberg.types.Types.NestedField;
@@ -87,4 +106,216 @@ public class TestIcebergRESTUtils {
Assertions.assertEquals(field, clonedField);
}
}
+
+ @Test
+ void testTableCredentialsPath() {
+ TableIdentifier table = TableIdentifier.of(Namespace.of("ns"), "tbl");
+ TableMetadata tableMetadata = mock(TableMetadata.class);
+ when(tableMetadata.location()).thenReturn("s3://bucket/t/");
+ org.apache.iceberg.rest.credentials.Credential credential =
+ IcebergRESTUtils.toRESTCredential(
+ "my_catalog", table, new S3TokenCredential("k", "s", "t", 99L),
tableMetadata);
+ Assertions.assertEquals(
+ "v1/my_catalog/namespaces/ns/tables/tbl/credentials",
+ credential.config().get("client.refresh-credentials-endpoint"));
+ }
+
+ @Test
+ void testToRESTCredential() {
+ TableIdentifier table = TableIdentifier.of(Namespace.of("ns"), "tbl");
+ String refreshPath = "v1/cat/namespaces/ns/tables/tbl/credentials";
+ TableMetadata metadataWithSlash = mock(TableMetadata.class);
+ when(metadataWithSlash.location()).thenReturn("s3://bucket/t/");
+ org.apache.iceberg.rest.credentials.Credential credentialWithSlash =
+ IcebergRESTUtils.toRESTCredential(
+ "cat", table, new S3TokenCredential("k", "s", "t", 99L),
metadataWithSlash);
+ Assertions.assertEquals("s3://bucket/t/", credentialWithSlash.prefix());
+ Assertions.assertEquals(
+ "99",
credentialWithSlash.config().get("s3.session-token-expires-at-ms"));
+ Assertions.assertEquals(
+ refreshPath,
credentialWithSlash.config().get("client.refresh-credentials-endpoint"));
+
+ TableMetadata metadataWithoutSlash = mock(TableMetadata.class);
+
when(metadataWithoutSlash.location()).thenReturn("s3://bucket/path/to/table");
+ org.apache.iceberg.rest.credentials.Credential credentialWithoutSlash =
+ IcebergRESTUtils.toRESTCredential(
+ "cat", table, new S3TokenCredential("k", "s", "t", 99L),
metadataWithoutSlash);
+ Assertions.assertEquals("s3://bucket/path/to/table",
credentialWithoutSlash.prefix());
+ }
+
+ @Test
+ void testToRESTCredentialForS3Token() {
+ TableIdentifier table = TableIdentifier.of(Namespace.of("ns"), "tbl");
+ TableMetadata tableMetadata = mock(TableMetadata.class);
+ when(tableMetadata.location()).thenReturn("s3://bucket/t/");
+ Map<String, String> config =
+ IcebergRESTUtils.toRESTCredential(
+ "aws", table, new S3TokenCredential("key", "secret", "token",
1234L), tableMetadata)
+ .config();
+
+ Assertions.assertEquals(
+ "v1/aws/namespaces/ns/tables/tbl/credentials",
+ config.get("client.refresh-credentials-endpoint"));
+ Assertions.assertEquals("1234",
config.get("s3.session-token-expires-at-ms"));
+ }
+
+ @Test
+ void testToRESTCredentialForAwsIrsa() {
+ TableIdentifier table = TableIdentifier.of(Namespace.of("ns"), "tbl");
+ TableMetadata tableMetadata = mock(TableMetadata.class);
+ when(tableMetadata.location()).thenReturn("s3://bucket/t/");
+ Map<String, String> config =
+ IcebergRESTUtils.toRESTCredential(
+ "aws", table, new AwsIrsaCredential("key", "secret", "token",
4321L), tableMetadata)
+ .config();
+
+ Assertions.assertEquals(
+ "v1/aws/namespaces/ns/tables/tbl/credentials",
+ config.get("client.refresh-credentials-endpoint"));
+ Assertions.assertEquals("4321",
config.get("s3.session-token-expires-at-ms"));
+ }
+
+ @Test
+ void testToRESTCredentialConfigIsUnmodifiable() {
+ TableIdentifier table = TableIdentifier.of(Namespace.of("ns"), "tbl");
+ TableMetadata tableMetadata = mock(TableMetadata.class);
+ when(tableMetadata.location()).thenReturn("s3://bucket/t/");
+ Map<String, String> config =
+ IcebergRESTUtils.toRESTCredential(
+ "aws", table, new S3TokenCredential("k", "s", "t", 99L),
tableMetadata)
+ .config();
+
+ Assertions.assertThrows(UnsupportedOperationException.class, () ->
config.put("k", "v"));
+ }
+
+ @Test
+ void testToRESTCredentialForGcsToken() {
+ TableIdentifier table = TableIdentifier.of(Namespace.of("ns"), "tbl");
+ TableMetadata tableMetadata = mock(TableMetadata.class);
+ when(tableMetadata.location()).thenReturn("gs://bucket/t/");
+ Map<String, String> config =
+ IcebergRESTUtils.toRESTCredential(
+ "gcs", table, new GCSTokenCredential("gcs-token", 5678L),
tableMetadata)
+ .config();
+
+ Assertions.assertEquals(
+ "v1/gcs/namespaces/ns/tables/tbl/credentials",
+ config.get("gcs.oauth2.refresh-credentials-endpoint"));
+ Assertions.assertEquals("5678", config.get("gcs.oauth2.token-expires-at"));
+ }
+
+ @Test
+ void testToRESTCredentialForOssToken() {
+ TableIdentifier table = TableIdentifier.of(Namespace.of("ns"), "tbl");
+ TableMetadata tableMetadata = mock(TableMetadata.class);
+ when(tableMetadata.location()).thenReturn("oss://bucket/t/");
+ Map<String, String> config =
+ IcebergRESTUtils.toRESTCredential(
+ "oss",
+ table,
+ new OSSTokenCredential("key", "secret", "oss-token", 9012L),
+ tableMetadata)
+ .config();
+
+
Assertions.assertFalse(config.containsKey("client.refresh-credentials-endpoint"));
+ Assertions.assertEquals("9012",
config.get("client.security-token-expires-at-ms"));
+ Assertions.assertEquals("key", config.get("client.access-key-id"));
+ Assertions.assertEquals("secret", config.get("client.access-key-secret"));
+ Assertions.assertEquals("oss-token", config.get("client.security-token"));
+ }
+
+ @Test
+ void testToRESTCredentialForAdlsToken() {
+ TableIdentifier table = TableIdentifier.of(Namespace.of("ns"), "tbl");
+ TableMetadata tableMetadata = mock(TableMetadata.class);
+
when(tableMetadata.location()).thenReturn("abfss://[email protected]/t/");
+ Map<String, String> config =
+ IcebergRESTUtils.toRESTCredential(
+ "adls",
+ table,
+ new ADLSTokenCredential("storageacct", "sas-token", 3456L),
+ tableMetadata)
+ .config();
+
+ Assertions.assertEquals(
+ "v1/adls/namespaces/ns/tables/tbl/credentials",
+ config.get("adls.refresh-credentials-endpoint"));
+ Assertions.assertEquals(
+ "3456",
config.get("adls.sas-token-expires-at-ms.storageacct.dfs.core.windows.net"));
+ }
+
+ @Test
+ void testBuildStorageCreds() {
+ TableIdentifier table = TableIdentifier.of(Namespace.of("db"), "tbl");
+ FileIO fileIO =
+ mock(FileIO.class,
withSettings().extraInterfaces(SupportsStorageCredentials.class));
+ SupportsStorageCredentials storageCredentialsFileIO =
(SupportsStorageCredentials) fileIO;
+ StorageCredential upstreamCredential =
+ StorageCredential.create(
+ "s3://bucket/db/tbl/",
+ ImmutableMap.of(
+ "s3.access-key-id",
+ "upstream-key",
+ "s3.secret-access-key",
+ "upstream-secret",
+ "s3.session-token",
+ "upstream-token",
+ "s3.session-token-expires-at-ms",
+ "123",
+ "client.refresh-credentials-endpoint",
+ "v1/upstream/namespaces/db/tables/tbl/credentials"));
+
when(storageCredentialsFileIO.credentials()).thenReturn(List.of(upstreamCredential));
+
+ List<Credential> credentials = IcebergRESTUtils.buildStorageCreds("irc1",
table, fileIO);
+
+ Assertions.assertEquals(1, credentials.size());
+ Credential credential = credentials.get(0);
+ Assertions.assertEquals("s3://bucket/db/tbl/", credential.prefix());
+ Assertions.assertEquals("upstream-token",
credential.config().get("s3.session-token"));
+ Assertions.assertEquals(
+ "v1/irc1/namespaces/db/tables/tbl/credentials",
+ credential.config().get("client.refresh-credentials-endpoint"));
+ Assertions.assertFalse(
+
credential.config().containsValue("v1/upstream/namespaces/db/tables/tbl/credentials"));
+ }
+
+ @Test
+ void testBuildStorageCredsPreservesSchemePrefix() {
+ TableIdentifier table = TableIdentifier.of(Namespace.of("db"), "tbl");
+ FileIO fileIO =
+ mock(FileIO.class,
withSettings().extraInterfaces(SupportsStorageCredentials.class));
+ SupportsStorageCredentials storageCredentialsFileIO =
(SupportsStorageCredentials) fileIO;
+ StorageCredential upstreamCredential =
+ StorageCredential.create(
+ "s3",
+ ImmutableMap.of(
+ "s3.session-token", "upstream-token",
+ "s3.session-token-expires-at-ms", "123"));
+
when(storageCredentialsFileIO.credentials()).thenReturn(List.of(upstreamCredential));
+
+ Credential credential = IcebergRESTUtils.buildStorageCreds("irc1", table,
fileIO).get(0);
+
+ Assertions.assertEquals("s3", credential.prefix());
+ }
+
+ @Test
+ void testBuildStorageCredsEmpty() {
+ TableIdentifier table = TableIdentifier.of(Namespace.of("db"), "tbl");
+ FileIO fileIO = mock(FileIO.class);
+
+ Assertions.assertTrue(IcebergRESTUtils.buildStorageCreds("irc1", table,
fileIO).isEmpty());
+ }
+
+ @Test
+ void testToRESTCredentialOmitsStaticSecretKey() {
+ TableIdentifier table = TableIdentifier.of(Namespace.of("ns"), "tbl");
+ TableMetadata tableMetadata = mock(TableMetadata.class);
+ when(tableMetadata.location()).thenReturn("s3://bucket/t/");
+ Map<String, String> config =
+ IcebergRESTUtils.toRESTCredential(
+ "aws", table, new S3SecretKeyCredential("key", "secret"),
tableMetadata)
+ .config();
+
+
Assertions.assertFalse(config.containsKey("client.refresh-credentials-endpoint"));
+ }
}
