From a1caa75997fd705be18a54354a4259595b655338 Mon Sep 17 00:00:00 2001 From: Xinyao Zhang <43081360+zhangxinyao88@users.noreply.github.com> Date: Mon, 31 Aug 2026 22:23:39 -0400 Subject: [PATCH 1/2] AWS: Resolve relative storage credential refresh endpoint Generated-by: Codex --- .../org/apache/iceberg/aws/s3/S3FileIO.java | 8 ++++++- .../aws/s3/VendedCredentialsProvider.java | 3 ++- .../aws/s3/TestS3FileIOCredentialRefresh.java | 5 ++-- .../aws/s3/TestVendedCredentialsProvider.java | 23 ++++++++++++++++++ .../gcs/OAuth2RefreshCredentialsHandler.java | 4 +++- .../TestOAuth2RefreshCredentialsHandler.java | 24 +++++++++++++++++++ 6 files changed, 62 insertions(+), 5 deletions(-) diff --git a/aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java b/aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java index 3301074613ef..d067ef6c3568 100644 --- a/aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java +++ b/aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java @@ -36,6 +36,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.stream.Collectors; +import org.apache.iceberg.aws.AwsClientProperties; import org.apache.iceberg.aws.S3FileIOAwsClientFactories; import org.apache.iceberg.common.DynConstructors; import org.apache.iceberg.io.BulkDeletionFailureException; @@ -461,7 +462,12 @@ private void refreshStorageCredentials() { return; } - try (VendedCredentialsProvider provider = VendedCredentialsProvider.create(properties)) { + Map refreshProperties = Maps.newHashMap(properties); + refreshProperties.putIfAbsent( + VendedCredentialsProvider.URI, + properties.get(AwsClientProperties.REFRESH_CREDENTIALS_ENDPOINT)); + + try (VendedCredentialsProvider provider = VendedCredentialsProvider.create(refreshProperties)) { List refreshed = provider.fetchCredentials().credentials().stream() .filter(c -> c.prefix().startsWith(ROOT_PREFIX)) diff --git a/aws/src/main/java/org/apache/iceberg/aws/s3/VendedCredentialsProvider.java b/aws/src/main/java/org/apache/iceberg/aws/s3/VendedCredentialsProvider.java index 0e530d02b30b..017541d5144b 100644 --- a/aws/src/main/java/org/apache/iceberg/aws/s3/VendedCredentialsProvider.java +++ b/aws/src/main/java/org/apache/iceberg/aws/s3/VendedCredentialsProvider.java @@ -65,7 +65,8 @@ private VendedCredentialsProvider(Map properties) { .cachedValueName(VendedCredentialsProvider.class.getName()) .build(); this.catalogEndpoint = properties.get(CatalogProperties.URI); - this.credentialsEndpoint = properties.get(URI); + this.credentialsEndpoint = + RESTUtil.resolveEndpoint(properties.get(CatalogProperties.URI), properties.get(URI)); } @Override diff --git a/aws/src/test/java/org/apache/iceberg/aws/s3/TestS3FileIOCredentialRefresh.java b/aws/src/test/java/org/apache/iceberg/aws/s3/TestS3FileIOCredentialRefresh.java index 170857ca84b4..4742bc8796ec 100644 --- a/aws/src/test/java/org/apache/iceberg/aws/s3/TestS3FileIOCredentialRefresh.java +++ b/aws/src/test/java/org/apache/iceberg/aws/s3/TestS3FileIOCredentialRefresh.java @@ -30,6 +30,7 @@ import java.util.concurrent.TimeUnit; import org.apache.iceberg.CatalogProperties; import org.apache.iceberg.TestHelpers; +import org.apache.iceberg.aws.AwsClientProperties; import org.apache.iceberg.aws.AwsProperties; import org.apache.iceberg.io.StorageCredential; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; @@ -145,8 +146,8 @@ public void credentialRefreshSchedulesNextRefresh() { ImmutableMap.of( AwsProperties.CLIENT_FACTORY, StaticClientFactory.class.getName(), - VendedCredentialsProvider.URI, - CREDENTIALS_URI, + AwsClientProperties.REFRESH_CREDENTIALS_ENDPOINT, + "/credentials", CatalogProperties.URI, CATALOG_URI, "init-creation-stacktrace", diff --git a/aws/src/test/java/org/apache/iceberg/aws/s3/TestVendedCredentialsProvider.java b/aws/src/test/java/org/apache/iceberg/aws/s3/TestVendedCredentialsProvider.java index 4749d0c8f89a..a3a248863f15 100644 --- a/aws/src/test/java/org/apache/iceberg/aws/s3/TestVendedCredentialsProvider.java +++ b/aws/src/test/java/org/apache/iceberg/aws/s3/TestVendedCredentialsProvider.java @@ -120,6 +120,29 @@ public void noS3Credentials() { } } + @Test + public void resolvesRelativeCredentialsUri() { + HttpRequest mockRequest = request("/v1/credentials").withMethod(HttpMethod.GET.name()); + HttpResponse mockResponse = + response( + LoadCredentialsResponseParser.toJson( + ImmutableLoadCredentialsResponse.builder().build())) + .withStatusCode(200); + mockServer.when(mockRequest).respond(mockResponse); + + try (VendedCredentialsProvider provider = + VendedCredentialsProvider.create( + ImmutableMap.of( + VendedCredentialsProvider.URI, + "/credentials", + CatalogProperties.URI, + CATALOG_URI))) { + assertThat(provider.fetchCredentials().credentials()).isEmpty(); + } + + mockServer.verify(mockRequest, VerificationTimes.once()); + } + @Test public void accessKeyIdAndSecretAccessKeyWithoutToken() { HttpRequest mockRequest = request("/v1/credentials").withMethod(HttpMethod.GET.name()); diff --git a/gcp/src/main/java/org/apache/iceberg/gcp/gcs/OAuth2RefreshCredentialsHandler.java b/gcp/src/main/java/org/apache/iceberg/gcp/gcs/OAuth2RefreshCredentialsHandler.java index 4b233d73305c..61d4f56fd070 100644 --- a/gcp/src/main/java/org/apache/iceberg/gcp/gcs/OAuth2RefreshCredentialsHandler.java +++ b/gcp/src/main/java/org/apache/iceberg/gcp/gcs/OAuth2RefreshCredentialsHandler.java @@ -57,7 +57,9 @@ private OAuth2RefreshCredentialsHandler(Map properties) { Preconditions.checkArgument( null != properties.get(CatalogProperties.URI), "Invalid catalog endpoint: null"); this.credentialsEndpoint = - properties.get(GCPProperties.GCS_OAUTH2_REFRESH_CREDENTIALS_ENDPOINT); + RESTUtil.resolveEndpoint( + properties.get(CatalogProperties.URI), + properties.get(GCPProperties.GCS_OAUTH2_REFRESH_CREDENTIALS_ENDPOINT)); this.catalogEndpoint = properties.get(CatalogProperties.URI); this.properties = properties; } diff --git a/gcp/src/test/java/org/apache/iceberg/gcp/gcs/TestOAuth2RefreshCredentialsHandler.java b/gcp/src/test/java/org/apache/iceberg/gcp/gcs/TestOAuth2RefreshCredentialsHandler.java index 0f29b9781895..872ea39d78bc 100644 --- a/gcp/src/test/java/org/apache/iceberg/gcp/gcs/TestOAuth2RefreshCredentialsHandler.java +++ b/gcp/src/test/java/org/apache/iceberg/gcp/gcs/TestOAuth2RefreshCredentialsHandler.java @@ -120,6 +120,30 @@ public void badRequest() { .hasMessageStartingWith("Malformed request"); } + @Test + public void resolvesRelativeCredentialsUri() { + HttpRequest mockRequest = + HttpRequest.request("/v1/credentials").withMethod(HttpMethod.GET.name()); + HttpResponse mockResponse = + HttpResponse.response( + LoadCredentialsResponseParser.toJson( + ImmutableLoadCredentialsResponse.builder().build())) + .withStatusCode(200); + mockServer.when(mockRequest).respond(mockResponse); + + try (OAuth2RefreshCredentialsHandler handler = + OAuth2RefreshCredentialsHandler.create( + ImmutableMap.of( + GCPProperties.GCS_OAUTH2_REFRESH_CREDENTIALS_ENDPOINT, + "/credentials", + CatalogProperties.URI, + CATALOG_URI))) { + assertThat(handler.fetchCredentials().credentials()).isEmpty(); + } + + mockServer.verify(mockRequest, VerificationTimes.once()); + } + @Test public void noGcsCredentialInResponse() { HttpRequest mockRequest = From d04edae934e48d2a3881b57ae5b9dc438cb41654 Mon Sep 17 00:00:00 2001 From: Xinyao Zhang <43081360+zhangxinyao88@users.noreply.github.com> Date: Tue, 6 Oct 2026 23:25:36 -0400 Subject: [PATCH 2/2] REST: Configure shared storage credential refresh endpoints Resolve the table credentials endpoint when constructing FileIO with vended credentials. AWS and GCP consume the shared endpoint and retain their existing configuration as a compatibility fallback. Generated-by: Codex --- .../iceberg/aws/AwsClientProperties.java | 8 +- .../org/apache/iceberg/aws/s3/S3FileIO.java | 8 +- .../aws/s3/VendedCredentialsProvider.java | 13 +++- .../iceberg/aws/TestAwsClientProperties.java | 17 +++++ .../aws/s3/TestS3FileIOCredentialRefresh.java | 15 ++-- .../aws/s3/TestVendedCredentialsProvider.java | 25 +++++++ .../iceberg/rest/RESTCatalogProperties.java | 8 ++ .../iceberg/rest/RESTSessionCatalog.java | 12 ++- .../apache/iceberg/rest/TestRESTCatalog.java | 75 +++++++++++++++++++ .../org/apache/iceberg/gcp/GCPProperties.java | 9 ++- .../gcs/OAuth2RefreshCredentialsHandler.java | 15 ++-- .../apache/iceberg/gcp/TestGCPProperties.java | 15 ++++ .../gcs/TestGCSFileIOCredentialRefresh.java | 13 ++-- .../TestOAuth2RefreshCredentialsHandler.java | 26 +++++++ 14 files changed, 224 insertions(+), 35 deletions(-) diff --git a/aws/src/main/java/org/apache/iceberg/aws/AwsClientProperties.java b/aws/src/main/java/org/apache/iceberg/aws/AwsClientProperties.java index c5850d0c79c1..77e5ac98b0d6 100644 --- a/aws/src/main/java/org/apache/iceberg/aws/AwsClientProperties.java +++ b/aws/src/main/java/org/apache/iceberg/aws/AwsClientProperties.java @@ -27,6 +27,7 @@ import org.apache.iceberg.common.DynMethods; import org.apache.iceberg.relocated.com.google.common.base.Preconditions; import org.apache.iceberg.relocated.com.google.common.base.Strings; +import org.apache.iceberg.rest.RESTCatalogProperties; import org.apache.iceberg.rest.RESTUtil; import org.apache.iceberg.util.PropertyUtil; import software.amazon.awssdk.auth.credentials.AwsBasicCredentials; @@ -118,8 +119,11 @@ public AwsClientProperties(Map properties) { Predicate.not(property -> property.startsWith(CLIENT_CREDENTIAL_PROVIDER_PREFIX))), PropertyUtil.propertiesWithPrefix(properties, CLIENT_CREDENTIAL_PROVIDER_PREFIX)); this.refreshCredentialsEndpoint = - RESTUtil.resolveEndpoint( - properties.get(CatalogProperties.URI), properties.get(REFRESH_CREDENTIALS_ENDPOINT)); + properties.containsKey(RESTCatalogProperties.CREDENTIALS_ENDPOINT) + ? properties.get(RESTCatalogProperties.CREDENTIALS_ENDPOINT) + : RESTUtil.resolveEndpoint( + properties.get(CatalogProperties.URI), + properties.get(REFRESH_CREDENTIALS_ENDPOINT)); this.refreshCredentialsEnabled = PropertyUtil.propertyAsBoolean(properties, REFRESH_CREDENTIALS_ENABLED, true); this.legacyMd5pluginEnabled = diff --git a/aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java b/aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java index d067ef6c3568..3301074613ef 100644 --- a/aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java +++ b/aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java @@ -36,7 +36,6 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.stream.Collectors; -import org.apache.iceberg.aws.AwsClientProperties; import org.apache.iceberg.aws.S3FileIOAwsClientFactories; import org.apache.iceberg.common.DynConstructors; import org.apache.iceberg.io.BulkDeletionFailureException; @@ -462,12 +461,7 @@ private void refreshStorageCredentials() { return; } - Map refreshProperties = Maps.newHashMap(properties); - refreshProperties.putIfAbsent( - VendedCredentialsProvider.URI, - properties.get(AwsClientProperties.REFRESH_CREDENTIALS_ENDPOINT)); - - try (VendedCredentialsProvider provider = VendedCredentialsProvider.create(refreshProperties)) { + try (VendedCredentialsProvider provider = VendedCredentialsProvider.create(properties)) { List refreshed = provider.fetchCredentials().credentials().stream() .filter(c -> c.prefix().startsWith(ROOT_PREFIX)) diff --git a/aws/src/main/java/org/apache/iceberg/aws/s3/VendedCredentialsProvider.java b/aws/src/main/java/org/apache/iceberg/aws/s3/VendedCredentialsProvider.java index 017541d5144b..41fc8c56db40 100644 --- a/aws/src/main/java/org/apache/iceberg/aws/s3/VendedCredentialsProvider.java +++ b/aws/src/main/java/org/apache/iceberg/aws/s3/VendedCredentialsProvider.java @@ -25,10 +25,12 @@ import java.util.Optional; import java.util.stream.Collectors; import org.apache.iceberg.CatalogProperties; +import org.apache.iceberg.aws.AwsClientProperties; import org.apache.iceberg.relocated.com.google.common.base.Preconditions; import org.apache.iceberg.relocated.com.google.common.base.Strings; import org.apache.iceberg.rest.ErrorHandlers; import org.apache.iceberg.rest.HTTPClient; +import org.apache.iceberg.rest.RESTCatalogProperties; import org.apache.iceberg.rest.RESTClient; import org.apache.iceberg.rest.RESTUtil; import org.apache.iceberg.rest.auth.AuthManager; @@ -56,7 +58,14 @@ public class VendedCredentialsProvider implements AwsCredentialsProvider, SdkAut private VendedCredentialsProvider(Map properties) { Preconditions.checkArgument(null != properties, "Invalid properties: null"); - Preconditions.checkArgument(null != properties.get(URI), "Invalid credentials endpoint: null"); + this.credentialsEndpoint = + properties.containsKey(RESTCatalogProperties.CREDENTIALS_ENDPOINT) + ? properties.get(RESTCatalogProperties.CREDENTIALS_ENDPOINT) + : RESTUtil.resolveEndpoint( + properties.get(CatalogProperties.URI), + properties.getOrDefault( + URI, properties.get(AwsClientProperties.REFRESH_CREDENTIALS_ENDPOINT))); + Preconditions.checkArgument(null != credentialsEndpoint, "Invalid credentials endpoint: null"); Preconditions.checkArgument( null != properties.get(CatalogProperties.URI), "Invalid catalog endpoint: null"); this.properties = properties; @@ -65,8 +74,6 @@ private VendedCredentialsProvider(Map properties) { .cachedValueName(VendedCredentialsProvider.class.getName()) .build(); this.catalogEndpoint = properties.get(CatalogProperties.URI); - this.credentialsEndpoint = - RESTUtil.resolveEndpoint(properties.get(CatalogProperties.URI), properties.get(URI)); } @Override diff --git a/aws/src/test/java/org/apache/iceberg/aws/TestAwsClientProperties.java b/aws/src/test/java/org/apache/iceberg/aws/TestAwsClientProperties.java index 47bb12c83b40..74a4e7021336 100644 --- a/aws/src/test/java/org/apache/iceberg/aws/TestAwsClientProperties.java +++ b/aws/src/test/java/org/apache/iceberg/aws/TestAwsClientProperties.java @@ -25,6 +25,7 @@ import org.apache.iceberg.aws.s3.VendedCredentialsProvider; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; import org.apache.iceberg.relocated.com.google.common.collect.Maps; +import org.apache.iceberg.rest.RESTCatalogProperties; import org.apache.iceberg.rest.auth.OAuth2Properties; import org.junit.jupiter.api.Test; import org.mockito.ArgumentCaptor; @@ -143,6 +144,22 @@ public void refreshCredentialsEndpoint() { .isInstanceOf(VendedCredentialsProvider.class); } + @Test + void usesCatalogCredentialsEndpoint() { + String endpoint = "https://catalog.example/table/credentials"; + AwsClientProperties properties = + new AwsClientProperties( + ImmutableMap.of( + CatalogProperties.URI, "https://catalog.example", + RESTCatalogProperties.CREDENTIALS_ENDPOINT, endpoint, + AwsClientProperties.REFRESH_CREDENTIALS_ENDPOINT, "/legacy")); + + try (VendedCredentialsProvider provider = + (VendedCredentialsProvider) properties.credentialsProvider("key", "secret", "token")) { + assertThat(provider).extracting("credentialsEndpoint").isEqualTo(endpoint); + } + } + @Test public void refreshCredentialsEndpointSetButRefreshDisabled() { AwsClientProperties awsClientProperties = diff --git a/aws/src/test/java/org/apache/iceberg/aws/s3/TestS3FileIOCredentialRefresh.java b/aws/src/test/java/org/apache/iceberg/aws/s3/TestS3FileIOCredentialRefresh.java index 4742bc8796ec..e60adf88c782 100644 --- a/aws/src/test/java/org/apache/iceberg/aws/s3/TestS3FileIOCredentialRefresh.java +++ b/aws/src/test/java/org/apache/iceberg/aws/s3/TestS3FileIOCredentialRefresh.java @@ -30,11 +30,11 @@ import java.util.concurrent.TimeUnit; import org.apache.iceberg.CatalogProperties; import org.apache.iceberg.TestHelpers; -import org.apache.iceberg.aws.AwsClientProperties; import org.apache.iceberg.aws.AwsProperties; import org.apache.iceberg.io.StorageCredential; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; import org.apache.iceberg.rest.HttpMethod; +import org.apache.iceberg.rest.RESTCatalogProperties; import org.apache.iceberg.rest.credentials.ImmutableCredential; import org.apache.iceberg.rest.responses.ImmutableLoadCredentialsResponse; import org.apache.iceberg.rest.responses.LoadCredentialsResponse; @@ -44,6 +44,8 @@ import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; import org.mockserver.integration.ClientAndServer; import org.mockserver.model.HttpRequest; import org.mockserver.model.HttpResponse; @@ -72,8 +74,9 @@ public void before() { mockServer.reset(); } - @Test - public void credentialRefreshSchedulesNextRefresh() { + @ParameterizedTest + @ValueSource(strings = {"rest.credentials.endpoint", "client.refresh-credentials-endpoint"}) + public void credentialRefreshSchedulesNextRefresh(String endpointProperty) { String nearExpiryMs = Long.toString(Instant.now().plus(3, ChronoUnit.MINUTES).toEpochMilli()); StorageCredential initialCredential = @@ -146,8 +149,10 @@ public void credentialRefreshSchedulesNextRefresh() { ImmutableMap.of( AwsProperties.CLIENT_FACTORY, StaticClientFactory.class.getName(), - AwsClientProperties.REFRESH_CREDENTIALS_ENDPOINT, - "/credentials", + endpointProperty, + endpointProperty.equals(RESTCatalogProperties.CREDENTIALS_ENDPOINT) + ? CREDENTIALS_URI + : "/credentials", CatalogProperties.URI, CATALOG_URI, "init-creation-stacktrace", diff --git a/aws/src/test/java/org/apache/iceberg/aws/s3/TestVendedCredentialsProvider.java b/aws/src/test/java/org/apache/iceberg/aws/s3/TestVendedCredentialsProvider.java index a3a248863f15..77d8f89715a6 100644 --- a/aws/src/test/java/org/apache/iceberg/aws/s3/TestVendedCredentialsProvider.java +++ b/aws/src/test/java/org/apache/iceberg/aws/s3/TestVendedCredentialsProvider.java @@ -143,6 +143,31 @@ public void resolvesRelativeCredentialsUri() { mockServer.verify(mockRequest, VerificationTimes.once()); } + @Test + public void usesCatalogCredentialsEndpoint() { + HttpRequest mockRequest = request("/v1/credentials").withMethod(HttpMethod.GET.name()); + HttpResponse mockResponse = + response( + LoadCredentialsResponseParser.toJson( + ImmutableLoadCredentialsResponse.builder().build())) + .withStatusCode(200); + mockServer.when(mockRequest).respond(mockResponse); + + try (VendedCredentialsProvider provider = + VendedCredentialsProvider.create( + ImmutableMap.of( + RESTCatalogProperties.CREDENTIALS_ENDPOINT, + CREDENTIALS_URI, + VendedCredentialsProvider.URI, + "https://legacy.example/wrong", + CatalogProperties.URI, + CATALOG_URI))) { + assertThat(provider.fetchCredentials().credentials()).isEmpty(); + } + + mockServer.verify(mockRequest, VerificationTimes.once()); + } + @Test public void accessKeyIdAndSecretAccessKeyWithoutToken() { HttpRequest mockRequest = request("/v1/credentials").withMethod(HttpMethod.GET.name()); diff --git a/core/src/main/java/org/apache/iceberg/rest/RESTCatalogProperties.java b/core/src/main/java/org/apache/iceberg/rest/RESTCatalogProperties.java index fbb122e4e1c6..ae809f62a1e8 100644 --- a/core/src/main/java/org/apache/iceberg/rest/RESTCatalogProperties.java +++ b/core/src/main/java/org/apache/iceberg/rest/RESTCatalogProperties.java @@ -136,6 +136,14 @@ public static ScanPlanningMode fromString(String mode) { */ public static final String REMOTE_SIGNING_ENDPOINT = "rest.remote-signing.endpoint"; + /** + * The absolute endpoint for refreshing table storage credentials. + * + *

The REST catalog client sets this property when creating a table FileIO with vended + * credentials. FileIO implementations use it in preference to their legacy refresh properties. + */ + public static final String CREDENTIALS_ENDPOINT = "rest.credentials.endpoint"; + /** * The {@link RemoteSigningConfig}, JSON-encoded. * diff --git a/core/src/main/java/org/apache/iceberg/rest/RESTSessionCatalog.java b/core/src/main/java/org/apache/iceberg/rest/RESTSessionCatalog.java index e7b68dad4aae..e05d4bea75b9 100644 --- a/core/src/main/java/org/apache/iceberg/rest/RESTSessionCatalog.java +++ b/core/src/main/java/org/apache/iceberg/rest/RESTSessionCatalog.java @@ -1320,7 +1320,17 @@ private FileIO tableFileIO( RemoteSigningConfigParser.toJson(remoteSigningConfig)); } - return newFileIO(context, fullConf.buildKeepingLast(), storageCredentials); + Map ioProperties = Maps.newHashMap(fullConf.buildKeepingLast()); + ioProperties.remove(RESTCatalogProperties.CREDENTIALS_ENDPOINT); + if (!storageCredentials.isEmpty()) { + ioProperties.put( + RESTCatalogProperties.CREDENTIALS_ENDPOINT, + RESTUtil.resolveEndpoint( + properties().get(CatalogProperties.URI), + paths.table(tableIdentifier) + "/credentials")); + } + + return newFileIO(context, ioProperties, storageCredentials); } /** diff --git a/core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java b/core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java index 32b292309e81..9e315cf1ec66 100644 --- a/core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java +++ b/core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java @@ -3881,6 +3881,81 @@ public void testSequenceNumberConflictThrowsCommitFailed() { .hasMessageContaining("Validation failed, please retry"); } + @ParameterizedTest + @ValueSource(booleans = {true, false}) + void configuresCredentialsEndpointOnlyWithStorageCredentials(boolean hasCredentials) + throws IOException { + Credential credential = + ImmutableCredential.builder() + .prefix("s3://test-bucket/") + .putConfig("s3.access-key-id", "test-access-key") + .putConfig("s3.secret-access-key", "test-secret-key") + .build(); + + RESTCatalogAdapter adapter = + new RESTCatalogAdapter(backendCatalog) { + @SuppressWarnings("unchecked") + @Override + public T handleRequest( + Route route, + Map vars, + HTTPRequest httpRequest, + Class responseType, + Consumer> responseHeaders) { + T response = + super.handleRequest(route, vars, httpRequest, responseType, responseHeaders); + if (route == Route.LOAD_TABLE && response instanceof LoadTableResponse loadResponse) { + LoadTableResponse.Builder builder = + LoadTableResponse.builder() + .withTableMetadata(loadResponse.tableMetadata()) + .addAllConfig(loadResponse.config()) + .addConfig("client.refresh-credentials-endpoint", "v1/legacy/credentials") + .addConfig(RESTCatalogProperties.CREDENTIALS_ENDPOINT, "unresolved"); + if (hasCredentials) { + builder.addCredential(credential); + } else { + builder.addConfig("s3.remote-signing-enabled", "true"); + } + return (T) builder.build(); + } + return response; + } + }; + + AtomicReference> fileIOProperties = new AtomicReference<>(); + try (RESTCatalog catalog = + new RESTCatalog(SessionCatalog.SessionContext.createEmpty(), config -> adapter) { + @Override + protected RESTSessionCatalog newSessionCatalog( + Function, RESTClient> clientBuilder) { + return new RESTSessionCatalog( + clientBuilder, + (context, config) -> { + fileIOProperties.set(config); + return new TestCatalogUtil.TestFileIOWithStorageCredentials(); + }); + } + }) { + String catalogUri = "https://catalog.example/api/catalog/"; + catalog.initialize("test", ImmutableMap.of(CatalogProperties.URI, catalogUri)); + assertThat(fileIOProperties.get()) + .doesNotContainKey(RESTCatalogProperties.CREDENTIALS_ENDPOINT); + catalog.createNamespace(NS); + catalog.createTable(TABLE, SCHEMA); + catalog.loadTable(TABLE); + + if (hasCredentials) { + assertThat(fileIOProperties.get()) + .containsEntry( + RESTCatalogProperties.CREDENTIALS_ENDPOINT, + RESTUtil.resolveEndpoint(catalogUri, RESOURCE_PATHS.table(TABLE) + "/credentials")); + } else { + assertThat(fileIOProperties.get()) + .doesNotContainKey(RESTCatalogProperties.CREDENTIALS_ENDPOINT); + } + } + } + @Test public void testIoBuilderReceivesStorageCredentials() { Credential credential = diff --git a/gcp/src/main/java/org/apache/iceberg/gcp/GCPProperties.java b/gcp/src/main/java/org/apache/iceberg/gcp/GCPProperties.java index 2702ce565d4e..b6508303f1e6 100644 --- a/gcp/src/main/java/org/apache/iceberg/gcp/GCPProperties.java +++ b/gcp/src/main/java/org/apache/iceberg/gcp/GCPProperties.java @@ -30,6 +30,7 @@ import org.apache.iceberg.relocated.com.google.common.base.Preconditions; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; +import org.apache.iceberg.rest.RESTCatalogProperties; import org.apache.iceberg.rest.RESTUtil; import org.apache.iceberg.util.PropertyUtil; @@ -168,9 +169,11 @@ public GCPProperties(Map properties) { } gcsOauth2RefreshCredentialsEndpoint = - RESTUtil.resolveEndpoint( - properties.get(CatalogProperties.URI), - properties.get(GCS_OAUTH2_REFRESH_CREDENTIALS_ENDPOINT)); + properties.containsKey(RESTCatalogProperties.CREDENTIALS_ENDPOINT) + ? properties.get(RESTCatalogProperties.CREDENTIALS_ENDPOINT) + : RESTUtil.resolveEndpoint( + properties.get(CatalogProperties.URI), + properties.get(GCS_OAUTH2_REFRESH_CREDENTIALS_ENDPOINT)); gcsOauth2RefreshCredentialsEnabled = PropertyUtil.propertyAsBoolean(properties, GCS_OAUTH2_REFRESH_CREDENTIALS_ENABLED, true); gcsNoAuth = Boolean.parseBoolean(properties.getOrDefault(GCS_NO_AUTH, "false")); diff --git a/gcp/src/main/java/org/apache/iceberg/gcp/gcs/OAuth2RefreshCredentialsHandler.java b/gcp/src/main/java/org/apache/iceberg/gcp/gcs/OAuth2RefreshCredentialsHandler.java index 61d4f56fd070..90c1a8e0a3e9 100644 --- a/gcp/src/main/java/org/apache/iceberg/gcp/gcs/OAuth2RefreshCredentialsHandler.java +++ b/gcp/src/main/java/org/apache/iceberg/gcp/gcs/OAuth2RefreshCredentialsHandler.java @@ -32,6 +32,7 @@ import org.apache.iceberg.relocated.com.google.common.base.Preconditions; import org.apache.iceberg.rest.ErrorHandlers; import org.apache.iceberg.rest.HTTPClient; +import org.apache.iceberg.rest.RESTCatalogProperties; import org.apache.iceberg.rest.RESTClient; import org.apache.iceberg.rest.RESTUtil; import org.apache.iceberg.rest.auth.AuthManager; @@ -51,15 +52,15 @@ public class OAuth2RefreshCredentialsHandler private AuthSession authSession; private OAuth2RefreshCredentialsHandler(Map properties) { - Preconditions.checkArgument( - null != properties.get(GCPProperties.GCS_OAUTH2_REFRESH_CREDENTIALS_ENDPOINT), - "Invalid credentials endpoint: null"); + this.credentialsEndpoint = + properties.containsKey(RESTCatalogProperties.CREDENTIALS_ENDPOINT) + ? properties.get(RESTCatalogProperties.CREDENTIALS_ENDPOINT) + : RESTUtil.resolveEndpoint( + properties.get(CatalogProperties.URI), + properties.get(GCPProperties.GCS_OAUTH2_REFRESH_CREDENTIALS_ENDPOINT)); + Preconditions.checkArgument(null != credentialsEndpoint, "Invalid credentials endpoint: null"); Preconditions.checkArgument( null != properties.get(CatalogProperties.URI), "Invalid catalog endpoint: null"); - this.credentialsEndpoint = - RESTUtil.resolveEndpoint( - properties.get(CatalogProperties.URI), - properties.get(GCPProperties.GCS_OAUTH2_REFRESH_CREDENTIALS_ENDPOINT)); this.catalogEndpoint = properties.get(CatalogProperties.URI); this.properties = properties; } diff --git a/gcp/src/test/java/org/apache/iceberg/gcp/TestGCPProperties.java b/gcp/src/test/java/org/apache/iceberg/gcp/TestGCPProperties.java index 0ec2183fc355..ede3b4c2d9f2 100644 --- a/gcp/src/test/java/org/apache/iceberg/gcp/TestGCPProperties.java +++ b/gcp/src/test/java/org/apache/iceberg/gcp/TestGCPProperties.java @@ -26,6 +26,7 @@ import static org.assertj.core.api.Assertions.assertThatIllegalStateException; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; +import org.apache.iceberg.rest.RESTCatalogProperties; import org.junit.jupiter.api.Test; public class TestGCPProperties { @@ -62,6 +63,20 @@ public void refreshCredentialsEndpointSet() { .isEqualTo("/v1/credentials"); } + @Test + void usesCatalogCredentialsEndpoint() { + String endpoint = "https://catalog.example/table/credentials"; + GCPProperties properties = + new GCPProperties( + ImmutableMap.of( + RESTCatalogProperties.CREDENTIALS_ENDPOINT, + endpoint, + GCS_OAUTH2_REFRESH_CREDENTIALS_ENDPOINT, + "/legacy")); + + assertThat(properties.oauth2RefreshCredentialsEndpoint()).contains(endpoint); + } + @Test public void refreshCredentialsEndpointSetButRefreshDisabled() { GCPProperties gcpProperties = diff --git a/gcp/src/test/java/org/apache/iceberg/gcp/gcs/TestGCSFileIOCredentialRefresh.java b/gcp/src/test/java/org/apache/iceberg/gcp/gcs/TestGCSFileIOCredentialRefresh.java index 0b9bb37f5f90..15d9dd9c3bb8 100644 --- a/gcp/src/test/java/org/apache/iceberg/gcp/gcs/TestGCSFileIOCredentialRefresh.java +++ b/gcp/src/test/java/org/apache/iceberg/gcp/gcs/TestGCSFileIOCredentialRefresh.java @@ -43,6 +43,8 @@ import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; import org.mockserver.integration.ClientAndServer; import org.mockserver.model.HttpRequest; import org.mockserver.model.HttpResponse; @@ -72,8 +74,9 @@ void before() { mockServer.reset(); } - @Test - void credentialRefreshSchedulesNextRefresh() { + @ParameterizedTest + @ValueSource(strings = {"rest.credentials.endpoint", "gcs.oauth2.refresh-credentials-endpoint"}) + void credentialRefreshSchedulesNextRefresh(String endpointProperty) { String nearExpiryMs = Long.toString(Instant.now().plus(3, ChronoUnit.MINUTES).toEpochMilli()); StorageCredential initialCredential = @@ -132,11 +135,7 @@ void credentialRefreshSchedulesNextRefresh() { .withStatusCode(200)); Map properties = - ImmutableMap.of( - GCPProperties.GCS_OAUTH2_REFRESH_CREDENTIALS_ENDPOINT, - credentialsUri, - CatalogProperties.URI, - catalogUri); + ImmutableMap.of(endpointProperty, credentialsUri, CatalogProperties.URI, catalogUri); try (GCSFileIO fileIO = new GCSFileIO()) { fileIO.initialize(properties); diff --git a/gcp/src/test/java/org/apache/iceberg/gcp/gcs/TestOAuth2RefreshCredentialsHandler.java b/gcp/src/test/java/org/apache/iceberg/gcp/gcs/TestOAuth2RefreshCredentialsHandler.java index 872ea39d78bc..726cba452514 100644 --- a/gcp/src/test/java/org/apache/iceberg/gcp/gcs/TestOAuth2RefreshCredentialsHandler.java +++ b/gcp/src/test/java/org/apache/iceberg/gcp/gcs/TestOAuth2RefreshCredentialsHandler.java @@ -144,6 +144,32 @@ public void resolvesRelativeCredentialsUri() { mockServer.verify(mockRequest, VerificationTimes.once()); } + @Test + public void usesCatalogCredentialsEndpoint() { + HttpRequest mockRequest = + HttpRequest.request("/v1/credentials").withMethod(HttpMethod.GET.name()); + HttpResponse mockResponse = + HttpResponse.response( + LoadCredentialsResponseParser.toJson( + ImmutableLoadCredentialsResponse.builder().build())) + .withStatusCode(200); + mockServer.when(mockRequest).respond(mockResponse); + + try (OAuth2RefreshCredentialsHandler handler = + OAuth2RefreshCredentialsHandler.create( + ImmutableMap.of( + RESTCatalogProperties.CREDENTIALS_ENDPOINT, + CREDENTIALS_URI, + GCPProperties.GCS_OAUTH2_REFRESH_CREDENTIALS_ENDPOINT, + "https://legacy.example/wrong", + CatalogProperties.URI, + CATALOG_URI))) { + assertThat(handler.fetchCredentials().credentials()).isEmpty(); + } + + mockServer.verify(mockRequest, VerificationTimes.once()); + } + @Test public void noGcsCredentialInResponse() { HttpRequest mockRequest =