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 9419878b5d [#11138][#11137] refactor(iceberg-rest): federation
handling via FederatedCatalogWrapper subclass (#11367)
9419878b5d is described below
commit 9419878b5d2bccc7bb9f1f3a3220b11807d3be8a
Author: roryqi <[email protected]>
AuthorDate: Fri Jun 5 20:58:54 2026 +0800
[#11138][#11137] refactor(iceberg-rest): federation handling via
FederatedCatalogWrapper subclass (#11367)
### What changes were proposed in this pull request?
Federation logic in `CatalogWrapperForREST` was implemented as scattered
`instanceof RESTCatalog` / `isRESTCatalog()` short-circuits across the
class.
This PR refactors them into polymorphic dispatch and fixes the missing
`registerTable` federation path:
- Add `FederatedCatalogWrapper extends CatalogWrapperForREST` that
overrides the
operations needing federation-aware behavior: `createTable`,
`loadTable`,
`updateTable`, `registerTable`, and `buildCatalogConfigToClients`.
- Add a `CatalogWrapperForREST.create(...)` factory that returns the
federated
subclass when the backend is `rest`, and wire
`IcebergCatalogWrapperManager`
to it.
- Add `registerTableInternal`, mirroring `loadTableInternal`, so
federated
`registerTable` extracts FileIO-derived client config and credential
fields
from the remote catalog's `table.io()` (previously dropped). (#11137)
- Remove the now-dead `instanceof`/`isRESTCatalog()` branches from the
base
class (`createTable`, `loadTable`, `updateTable`,
`shouldGenerateCredential`),
and make `buildCatalogConfigToClients` an overridable instance method
backed by
a shared `filterCatalogConfigForClients` helper.
### Why are the changes needed?
Adding a new federated operation required remembering to add a
corresponding
`instanceof` check; nothing in the type system enforced it. The missing
`registerTable` federation handling was a direct bug from this pattern:
a
federated `registerTable` returned a `LoadTableResponse` without the
FileIO-derived properties and credential fields that
`createTable`/`loadTable`
include. Polymorphic dispatch removes the foot-gun and fixes the bug.
Fix: #11138
Fix: #11137
### Does this PR introduce _any_ user-facing change?
No. Federated `registerTable` responses now include the
FileIO/credential client
config they should have included before; no API or property-key changes.
### How was this patch tested?
`./gradlew :iceberg:iceberg-rest-server:test -PskipITs` (full module
unit suite
green), plus new unit tests:
- `testCreateReturnsFederatedWrapperForRestBackend`
- `testCreateReturnsBaseWrapperForNonRestBackend`
- `testFederatedRegisterTableIncludesFileIo`
Existing `buildCatalogConfigToClients` tests were updated to exercise
the
polymorphic source selection and the shared filter helper.
---
.../iceberg/service/CatalogWrapperForREST.java | 474 ++-----------------
.../iceberg/service/FederatedCatalogWrapper.java | 506 +++++++++++++++++++++
.../service/IcebergCatalogWrapperManager.java | 14 +-
.../iceberg/service/IcebergRESTUtils.java | 52 ++-
.../iceberg/service/TestCatalogWrapperForREST.java | 115 +++--
.../TestIcebergCatalogWrapperManagerForREST.java | 42 ++
.../iceberg/service/TestIcebergRESTUtils.java | 32 ++
7 files changed, 766 insertions(+), 469 deletions(-)
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 7e1aac0c0d..e02c6d4f55 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
@@ -20,14 +20,10 @@
package org.apache.gravitino.iceberg.service;
import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
-import com.google.common.collect.Maps;
import java.io.IOException;
import java.net.URI;
-import java.time.OffsetDateTime;
-import java.time.ZoneOffset;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
@@ -36,7 +32,6 @@ import java.util.Locale;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
-import java.util.stream.Collectors;
import java.util.stream.Stream;
import org.apache.commons.lang3.StringUtils;
import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants;
@@ -44,7 +39,6 @@ import
org.apache.gravitino.credential.CatalogCredentialManager;
import org.apache.gravitino.credential.Credential;
import org.apache.gravitino.credential.CredentialConstants;
import org.apache.gravitino.credential.CredentialPrivilege;
-import org.apache.gravitino.credential.CredentialPropertyUtils;
import org.apache.gravitino.credential.PathBasedCredentialContext;
import org.apache.gravitino.iceberg.common.IcebergConfig;
import org.apache.gravitino.iceberg.common.ops.IcebergCatalogWrapper;
@@ -54,48 +48,22 @@ import org.apache.gravitino.storage.GCSProperties;
import org.apache.gravitino.utils.ClassUtils;
import org.apache.gravitino.utils.MapUtils;
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;
-import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Scan;
-import org.apache.iceberg.Schema;
-import org.apache.iceberg.SortOrder;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableMetadata;
-import org.apache.iceberg.TableOperations;
import org.apache.iceberg.TableProperties;
import org.apache.iceberg.TableScan;
-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.TableIdentifier;
-import org.apache.iceberg.exceptions.AlreadyExistsException;
-import org.apache.iceberg.exceptions.NoSuchTableException;
import org.apache.iceberg.exceptions.ServiceUnavailableException;
-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;
-import org.apache.iceberg.rest.requests.UpdateTableRequest;
import org.apache.iceberg.rest.responses.ImmutableLoadCredentialsResponse;
import org.apache.iceberg.rest.responses.LoadCredentialsResponse;
import org.apache.iceberg.rest.responses.LoadTableResponse;
@@ -104,8 +72,8 @@ import
org.apache.iceberg.rest.responses.PlanTableScanResponse;
/** Process Iceberg REST specific operations, like credential vending. */
public class CatalogWrapperForREST extends IcebergCatalogWrapper {
- private static final String FORMAT_VERSION = "format-version";
- private final CatalogCredentialManager catalogCredentialManager;
+ /** Vends Gravitino-managed credentials and exposes the catalog name for
credential refresh. */
+ protected final CatalogCredentialManager catalogCredentialManager;
private volatile Map<String, String> catalogConfigToClients;
private final Object catalogConfigToClientsLock = new Object();
@@ -114,9 +82,12 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
private static final String DATA_ACCESS_VENDED_CREDENTIALS =
"vended-credentials";
private static final String DATA_ACCESS_REMOTE_SIGNING = "remote-signing";
- private static final Schema EMPTY_SCHEMA = new Schema();
- private static final Set<String> catalogPropertiesToClientKeys =
+ /**
+ * Client-facing catalog property keys retained when building the IRC {@code
/v1/config} defaults
+ * and when extracting FileIO-derived config in {@link
FederatedCatalogWrapper}.
+ */
+ protected static final Set<String> catalogPropertiesToClientKeys =
ImmutableSet.of(
IcebergConstants.IO_IMPL,
IcebergConstants.AWS_S3_REGION,
@@ -144,12 +115,7 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
public LoadTableResponse createTable(
Namespace namespace, CreateTableRequest request, boolean
requestCredential) {
- LoadTableResponse loadTableResponse;
- if (isRESTCatalog()) {
- loadTableResponse = createTableInternal(namespace, request);
- } else {
- loadTableResponse = super.createTable(namespace, request);
- }
+ LoadTableResponse loadTableResponse = super.createTable(namespace,
request);
if (shouldGenerateCredential(loadTableResponse, requestCredential)) {
return injectCredentialConfig(
TableIdentifier.of(namespace, request.name()),
@@ -161,12 +127,7 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
public LoadTableResponse loadTable(
TableIdentifier identifier, boolean requestCredential,
CredentialPrivilege privilege) {
- LoadTableResponse loadTableResponse;
- if (isRESTCatalog()) {
- loadTableResponse = loadTableInternal(identifier);
- } else {
- loadTableResponse = super.loadTable(identifier);
- }
+ LoadTableResponse loadTableResponse = super.loadTable(identifier);
if (shouldGenerateCredential(loadTableResponse, requestCredential)) {
return injectCredentialConfig(identifier, loadTableResponse, privilege);
}
@@ -188,34 +149,19 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
return loadTableResponse;
}
- @Override
- public LoadTableResponse updateTable(
- TableIdentifier tableIdentifier, UpdateTableRequest updateTableRequest) {
- if (isRESTCatalog()) {
- return tableUpdateInternal(tableIdentifier, updateTableRequest);
- } else {
- return super.updateTable(tableIdentifier, updateTableRequest);
- }
- }
-
/**
- * Get table credentials.
+ * Get table credentials from the local (Gravitino-managed) catalog backend.
+ *
+ * <p>{@link FederatedCatalogWrapper} overrides this to fetch credentials
from the remote REST
+ * catalog instead, so the base implementation never needs an {@code
instanceof RESTCatalog}
+ * check.
*
* @param identifier table identifier
- * @param privilege used for local credential vending; ignored for REST
catalog backends
+ * @param privilege used for local credential vending
* @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.tableMetadata(),
privilege);
@@ -233,60 +179,6 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
}
}
- 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 {
@@ -313,7 +205,7 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
synchronized (catalogConfigToClientsLock) {
if (catalogConfigToClients == null) {
- catalogConfigToClients =
buildCatalogConfigToClients(getIcebergConfig(), getCatalog());
+ catalogConfigToClients = buildCatalogConfigToClients();
}
return catalogConfigToClients;
}
@@ -322,29 +214,35 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
/**
* Builds properties exposed to Iceberg clients via the IRC {@code
/v1/config} defaults.
*
- * <p>For {@link RESTCatalog}, uses {@link RESTCatalog#properties()} so
defaults reflect the
- * remote catalog's config response merged with client properties (after
REST handshake), not only
- * static Gravitino catalog configuration.
+ * <p>The base implementation uses the static Gravitino catalog
configuration as the property
+ * source. {@link FederatedCatalogWrapper} overrides this to use {@code
RESTCatalog.properties()}
+ * so defaults reflect the remote catalog's config response merged with
client properties (after
+ * REST handshake).
*
* <p>{@link IcebergConstants#IO_IMPL} is passed through when present (e.g.
Iceberg {@link
* org.apache.iceberg.io.ResolvingFileIO}), so clients multiplex by URI
scheme without server-side
* rewriting per table.
+ *
+ * @return the immutable, filtered properties exposed to Iceberg clients.
*/
@VisibleForTesting
- static Map<String, String> buildCatalogConfigToClients(IcebergConfig config,
Catalog catalog) {
- Map<String, String> sourceProps;
- if (catalog instanceof RESTCatalog) {
- Map<String, String> merged = ((RESTCatalog) catalog).properties();
- sourceProps = merged != null ? new HashMap<>(merged) : new HashMap<>();
- } else {
- sourceProps = new HashMap<>(config.getIcebergCatalogProperties());
- }
+ Map<String, String> buildCatalogConfigToClients() {
+ return
filterCatalogConfigForClients(getIcebergConfig().getIcebergCatalogProperties());
+ }
+ /**
+ * Keeps only the client-facing catalog properties and validates the
data-access property.
+ *
+ * @param sourceProps the candidate properties to filter.
+ * @return the immutable, filtered properties exposed to Iceberg clients.
+ */
+ protected static Map<String, String> filterCatalogConfigForClients(
+ Map<String, String> sourceProps) {
Map<String, String> filtered =
- MapUtils.getFilteredMap(sourceProps, key ->
catalogPropertiesToClientKeys.contains(key));
- filtered = new HashMap<>(filtered);
+ new HashMap<>(
+ MapUtils.getFilteredMap(
+ sourceProps, key ->
catalogPropertiesToClientKeys.contains(key)));
validateAndNormalizeDataAccessProperty(filtered);
-
return Collections.unmodifiableMap(filtered);
}
@@ -388,20 +286,21 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
credential.credentialType(),
tableIdentifier);
- // Merge temporary credential fields as Iceberg client config entries in
the load-table
- // response.
- Map<String, String> credentialConfig =
+ // Vend the temporary credential both as a first-class REST credential
(storage-prefix scoped,
+ // for Iceberg 1.7+ clients) and, for backward compatibility, flattened
into the load-table
+ // response config. This mirrors FederatedCatalogWrapper, which also
populates both fields.
+ org.apache.iceberg.rest.credentials.Credential restCredential =
IcebergRESTUtils.toRESTCredential(
- catalogCredentialManager.catalogName(),
- tableIdentifier,
- credential,
- loadTableResponse.tableMetadata())
- .config();
+ catalogCredentialManager.catalogName(),
+ tableIdentifier,
+ credential,
+ loadTableResponse.tableMetadata());
return LoadTableResponse.builder()
.withTableMetadata(loadTableResponse.tableMetadata())
.addAllConfig(loadTableResponse.config())
.addAllConfig(getCatalogConfigToClient())
- .addAllConfig(credentialConfig)
+ .addAllConfig(restCredential.config())
+ .addCredential(restCredential)
.build();
}
@@ -437,11 +336,6 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
return false;
}
- // RESTCatalog will fetch credential from the remote catalog instead of
generating credential
- if (getCatalog() instanceof RESTCatalog) {
- return false;
- }
-
validateCredentialLocation(loadTableResponse.tableMetadata().location());
return !isLocalOrHdfsTable(loadTableResponse.tableMetadata());
}
@@ -730,280 +624,4 @@ public class CatalogWrapperForREST extends
IcebergCatalogWrapper {
properties.put(newProperty, deprecatedValue);
}
}
-
- private LoadTableResponse createTableInternal(Namespace namespace,
CreateTableRequest request) {
- Catalog loadedCatalog = getCatalog();
-
- request.validate();
-
- if (request.stageCreate()) {
- return stageTableCreateInternal(namespace, request);
- }
-
- TableIdentifier ident = TableIdentifier.of(namespace, request.name());
- Table table =
- loadedCatalog
- .buildTable(ident, request.schema())
- .withLocation(request.location())
- .withPartitionSpec(request.spec())
- .withSortOrder(request.writeOrder())
- .withProperties(request.properties())
- .create();
-
- 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(filteredCredentialProperties)
- .addAllConfig(
- IcebergRESTUtils.buildRefreshProps(
- catalogCredentialManager.catalogName(), ident,
filteredCredentialProperties))
- .addAllCredentials(
- IcebergRESTUtils.buildStorageCreds(
- catalogCredentialManager.catalogName(), ident, table.io()))
- .build();
- }
-
- throw new IllegalStateException("Cannot wrap catalog that does not produce
BaseTable");
- }
-
- private LoadTableResponse stageTableCreateInternal(
- Namespace namespace, CreateTableRequest request) {
- Catalog loadedCatalog = getCatalog();
- TableIdentifier ident = TableIdentifier.of(namespace, request.name());
- if (loadedCatalog.tableExists(ident)) {
- throw new AlreadyExistsException("Table already exists: %s", ident);
- }
-
- Map<String, String> properties = Maps.newHashMap();
- properties.put("created-at",
OffsetDateTime.now(ZoneOffset.UTC).toString());
- properties.putAll(request.properties());
-
- Map<String, String> config = Maps.newHashMap();
- Catalog.TableBuilder tableBuilder =
- loadedCatalog
- .buildTable(ident, request.schema())
- .withPartitionSpec(request.spec())
- .withSortOrder(request.writeOrder())
- .withProperties(properties);
-
- Table table;
- if (request.location() != null) {
- table =
tableBuilder.withLocation(request.location()).createTransaction().table();
- } else {
- table = tableBuilder.createTransaction().table();
- }
-
- 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(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(
- request.schema(),
- request.spec() != null ? request.spec() :
PartitionSpec.unpartitioned(),
- request.writeOrder() != null ? request.writeOrder() :
SortOrder.unsorted(),
- table.location(),
- properties);
-
- return LoadTableResponse.builder()
- .withTableMetadata(metadata)
- .addAllConfig(config)
- .addAllCredentials(credentials)
- .build();
- }
-
- private LoadTableResponse tableUpdateInternal(TableIdentifier ident,
UpdateTableRequest request) {
- if (isCreate(request)) {
- // this is a hacky way to get TableOperations for an uncommitted table
- Optional<Integer> formatVersion =
- request.updates().stream()
- .filter(update -> update instanceof
MetadataUpdate.UpgradeFormatVersion)
- .map(update -> ((MetadataUpdate.UpgradeFormatVersion)
update).formatVersion())
- .findFirst();
-
- Schema schema =
- request.updates().stream()
- .filter(update -> update instanceof MetadataUpdate.AddSchema)
- .map(update -> ((MetadataUpdate.AddSchema) update).schema())
- .findFirst()
- .orElse(EMPTY_SCHEMA);
-
- Catalog.TableBuilder tableBuilder = getCatalog().buildTable(ident,
schema);
-
- TableMetadata.Builder changedMetadata =
-
formatVersion.map(TableMetadata::buildFromEmpty).orElse(TableMetadata.buildFromEmpty());
- request.updates().forEach(update -> update.applyTo(changedMetadata));
-
- TableMetadata changedTableMeta = changedMetadata.build();
- tableBuilder.withPartitionSpec(changedTableMeta.spec());
- tableBuilder.withSortOrder(changedTableMeta.sortOrder());
- tableBuilder.withLocation(changedTableMeta.location());
- tableBuilder.withProperty(FORMAT_VERSION,
String.valueOf(changedTableMeta.formatVersion()));
- tableBuilder.withProperties(changedTableMeta.properties());
-
- Transaction transaction = tableBuilder.createOrReplaceTransaction();
- if (transaction instanceof BaseTransaction) {
- BaseTransaction baseTransaction = (BaseTransaction) transaction;
-
- return LoadTableResponse.builder()
- .withTableMetadata(create(baseTransaction, request))
- .build();
- } else {
- throw new IllegalStateException(
- "Cannot wrap catalog that does not produce BaseTransaction");
- }
-
- } else {
- return CatalogHandlers.updateTable(getCatalog(), ident, request);
- }
- }
-
- private LoadTableResponse loadTableInternal(TableIdentifier ident) {
- Table table = getCatalog().loadTable(ident);
-
- 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(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
- throw new NoSuchTableException("Table does not exist: %s",
ident.toString());
- }
-
- throw new IllegalStateException("Cannot wrap catalog that does not produce
BaseTable");
- }
-
- private static boolean isCreate(UpdateTableRequest request) {
- boolean isCreate =
- request.requirements().stream()
-
.anyMatch(UpdateRequirement.AssertTableDoesNotExist.class::isInstance);
-
- if (isCreate) {
- List<UpdateRequirement> invalidRequirements =
- request.requirements().stream()
- .filter(req -> !(req instanceof
UpdateRequirement.AssertTableDoesNotExist))
- .collect(Collectors.toList());
- Preconditions.checkArgument(
- invalidRequirements.isEmpty(), "Invalid create requirements: %s",
invalidRequirements);
- }
-
- return isCreate;
- }
-
- private static TableMetadata create(BaseTransaction baseTransaction,
UpdateTableRequest request) {
- // the only valid requirement is that the table will be created
- TableOperations ops = baseTransaction.underlyingOps();
- request.requirements().forEach(requirement ->
requirement.validate(ops.current()));
-
- TableMetadata.Builder builder =
TableMetadata.buildFrom(baseTransaction.currentMetadata());
- request
- .updates()
- .forEach(
- update -> {
- if (shouldApplyMetadataUpdateAfterBuilder(update)) {
- update.applyTo(builder);
- }
- });
-
- // create transactions do not retry. if the table exists, retrying is not
a solution
- ops.commit(null, builder.build());
-
- return ops.current();
- }
-
- /**
- * Returns {@code false} for updates already reflected through {@link
Catalog.TableBuilder} during
- * staged create; those must not be applied again on {@link
TableMetadata.Builder}.
- */
- @VisibleForTesting
- static boolean shouldApplyMetadataUpdateAfterBuilder(MetadataUpdate update) {
- if (update instanceof MetadataUpdate.UpgradeFormatVersion) {
- return false;
- }
-
- if (update instanceof MetadataUpdate.AddSchema) {
- return false;
- }
-
- if (update instanceof MetadataUpdate.SetCurrentSchema) {
- return false;
- }
-
- if (update instanceof MetadataUpdate.RemoveSchemas) {
- return false;
- }
-
- if (update instanceof MetadataUpdate.SetLocation) {
- return false;
- }
-
- if (update instanceof MetadataUpdate.SetProperties) {
- return false;
- }
-
- if (update instanceof MetadataUpdate.RemoveProperties) {
- return false;
- }
-
- if (update instanceof MetadataUpdate.AddSortOrder) {
- return false;
- }
-
- if (update instanceof MetadataUpdate.SetDefaultSortOrder) {
- return false;
- }
-
- if (update instanceof MetadataUpdate.AddPartitionSpec) {
- return false;
- }
-
- if (update instanceof MetadataUpdate.SetDefaultPartitionSpec) {
- return false;
- }
-
- if (update instanceof MetadataUpdate.RemovePartitionSpecs) {
- return false;
- }
-
- return true;
- }
-
- private static Map<String, String> retrieveFileIOProperties(FileIO fileIO) {
- return fileIO instanceof InMemoryFileIO
- ? Maps.newHashMap()
- : new HashMap<>(fileIO.properties());
- }
}
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/FederatedCatalogWrapper.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/FederatedCatalogWrapper.java
new file mode 100644
index 0000000000..e726ca5617
--- /dev/null
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/FederatedCatalogWrapper.java
@@ -0,0 +1,506 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.gravitino.iceberg.service;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Maps;
+import java.time.OffsetDateTime;
+import java.time.ZoneOffset;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import org.apache.gravitino.credential.CredentialPrivilege;
+import org.apache.gravitino.credential.CredentialPropertyUtils;
+import org.apache.gravitino.iceberg.common.IcebergConfig;
+import org.apache.gravitino.utils.MapUtils;
+import org.apache.iceberg.BaseMetadataTable;
+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;
+import org.apache.iceberg.SortOrder;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.TableOperations;
+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.TableIdentifier;
+import org.apache.iceberg.exceptions.AlreadyExistsException;
+import org.apache.iceberg.exceptions.NoSuchTableException;
+import org.apache.iceberg.inmemory.InMemoryFileIO;
+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.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.credentials.Credential;
+import org.apache.iceberg.rest.requests.CreateTableRequest;
+import org.apache.iceberg.rest.requests.RegisterTableRequest;
+import org.apache.iceberg.rest.requests.UpdateTableRequest;
+import org.apache.iceberg.rest.responses.LoadCredentialsResponse;
+import org.apache.iceberg.rest.responses.LoadTableResponse;
+
+/**
+ * A {@link CatalogWrapperForREST} for a federated Iceberg REST catalog (the
underlying catalog is a
+ * {@link RESTCatalog}).
+ *
+ * <p>Federation-specific behavior is expressed through polymorphic overrides
instead of {@code
+ * instanceof RESTCatalog} checks scattered across the base class. Table
operations are routed to
+ * federation-aware {@code *Internal} methods so client-facing FileIO and
credential properties are
+ * extracted from the remote catalog's {@code table.io()}. Credentials are
vended by the remote
+ * catalog, so this wrapper never injects Gravitino-generated credentials.
+ */
+public class FederatedCatalogWrapper extends CatalogWrapperForREST {
+
+ private static final String FORMAT_VERSION = "format-version";
+ private static final Schema EMPTY_SCHEMA = new Schema();
+
+ /**
+ * Creates a federated wrapper.
+ *
+ * @param catalogName the catalog name.
+ * @param config the Iceberg catalog configuration (backend {@code rest}).
+ */
+ public FederatedCatalogWrapper(String catalogName, IcebergConfig config) {
+ super(catalogName, config);
+ }
+
+ @Override
+ public LoadTableResponse createTable(
+ Namespace namespace, CreateTableRequest request, boolean
requestCredential) {
+ // The remote REST catalog vends its own credentials, so the
requestCredential flag is not used
+ // here; FileIO-derived client config is extracted by createTableInternal.
+ return createTableInternal(namespace, request);
+ }
+
+ @Override
+ public LoadTableResponse loadTable(
+ TableIdentifier identifier, boolean requestCredential,
CredentialPrivilege privilege) {
+ return loadTableInternal(identifier);
+ }
+
+ @Override
+ public LoadTableResponse registerTable(
+ Namespace namespace, RegisterTableRequest request, boolean
requestCredential) {
+ return registerTableInternal(namespace, request);
+ }
+
+ @Override
+ public LoadTableResponse updateTable(
+ TableIdentifier tableIdentifier, UpdateTableRequest updateTableRequest) {
+ return tableUpdateInternal(tableIdentifier, updateTableRequest);
+ }
+
+ @Override
+ Map<String, String> buildCatalogConfigToClients() {
+ Map<String, String> merged = ((RESTCatalog) getCatalog()).properties();
+ return filterCatalogConfigForClients(merged != null ? merged :
Collections.emptyMap());
+ }
+
+ /**
+ * Fetches table credentials from the remote REST catalog instead of vending
Gravitino-managed
+ * credentials. The {@code privilege} is ignored because the remote catalog
decides what to vend.
+ *
+ * <p>The upstream credentials are rewritten via {@link
IcebergRESTUtils#rewriteTableCredentials}
+ * so their {@code refresh-credentials-endpoint} points at this IRC catalog
rather than the remote
+ * catalog, consistent with the {@link #loadTable}/{@code createTable}
federation paths.
+ *
+ * @param identifier the table identifier.
+ * @param privilege ignored; the remote REST catalog vends its own
credentials.
+ * @return the remote credentials with IRC-local refresh endpoints.
+ */
+ @Override
+ public LoadCredentialsResponse getTableCredentials(
+ TableIdentifier identifier, CredentialPrivilege privilege) {
+ LoadCredentialsResponse upstream =
+ getRESTTableCredentials((RESTCatalog) getCatalog(), identifier);
+ return IcebergRESTUtils.rewriteTableCredentials(
+ catalogCredentialManager.catalogName(), identifier, upstream);
+ }
+
+ 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);
+ }
+ }
+ }
+ }
+
+ /**
+ * Federation-aware {@code createTable}: creates the table on the underlying
(remote) catalog and
+ * extracts client-facing FileIO/credential properties from {@code
table.io()}.
+ */
+ private LoadTableResponse createTableInternal(Namespace namespace,
CreateTableRequest request) {
+ Catalog loadedCatalog = getCatalog();
+
+ request.validate();
+
+ if (request.stageCreate()) {
+ return stageTableCreateInternal(namespace, request);
+ }
+
+ TableIdentifier ident = TableIdentifier.of(namespace, request.name());
+ Table table =
+ loadedCatalog
+ .buildTable(ident, request.schema())
+ .withLocation(request.location())
+ .withPartitionSpec(request.spec())
+ .withSortOrder(request.writeOrder())
+ .withProperties(request.properties())
+ .create();
+
+ if (table instanceof BaseTable) {
+ return buildLoadTableResponseFromFileIo(ident, (BaseTable) table);
+ }
+
+ throw new IllegalStateException("Cannot wrap catalog that does not produce
BaseTable");
+ }
+
+ private LoadTableResponse stageTableCreateInternal(
+ Namespace namespace, CreateTableRequest request) {
+ Catalog loadedCatalog = getCatalog();
+ TableIdentifier ident = TableIdentifier.of(namespace, request.name());
+ if (loadedCatalog.tableExists(ident)) {
+ throw new AlreadyExistsException("Table already exists: %s", ident);
+ }
+
+ Map<String, String> properties = Maps.newHashMap();
+ properties.put("created-at",
OffsetDateTime.now(ZoneOffset.UTC).toString());
+ properties.putAll(request.properties());
+
+ Map<String, String> config = Maps.newHashMap();
+ Catalog.TableBuilder tableBuilder =
+ loadedCatalog
+ .buildTable(ident, request.schema())
+ .withPartitionSpec(request.spec())
+ .withSortOrder(request.writeOrder())
+ .withProperties(properties);
+
+ Table table;
+ if (request.location() != null) {
+ table =
tableBuilder.withLocation(request.location()).createTransaction().table();
+ } else {
+ table = tableBuilder.createTransaction().table();
+ }
+
+ 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(filteredCredentialProperties);
+ config.putAll(
+ IcebergRESTUtils.buildRefreshProps(
+ catalogCredentialManager.catalogName(), ident,
filteredCredentialProperties));
+
+ List<Credential> credentials =
+ IcebergRESTUtils.buildStorageCreds(
+ catalogCredentialManager.catalogName(), ident, table.io());
+
+ TableMetadata metadata =
+ TableMetadata.newTableMetadata(
+ request.schema(),
+ request.spec() != null ? request.spec() :
PartitionSpec.unpartitioned(),
+ request.writeOrder() != null ? request.writeOrder() :
SortOrder.unsorted(),
+ table.location(),
+ properties);
+
+ return LoadTableResponse.builder()
+ .withTableMetadata(metadata)
+ .addAllConfig(config)
+ .addAllCredentials(credentials)
+ .build();
+ }
+
+ /**
+ * Federation-aware {@code registerTable}: registers the existing table
metadata on the underlying
+ * (remote) catalog and extracts client-facing FileIO/credential properties
from {@code
+ * table.io()}, mirroring {@link #loadTableInternal(TableIdentifier)}.
+ */
+ private LoadTableResponse registerTableInternal(
+ Namespace namespace, RegisterTableRequest request) {
+ TableIdentifier ident = TableIdentifier.of(namespace, request.name());
+ Table table = getCatalog().registerTable(ident,
request.metadataLocation());
+
+ if (table instanceof BaseTable) {
+ return buildLoadTableResponseFromFileIo(ident, (BaseTable) table);
+ }
+
+ throw new IllegalStateException("Cannot wrap catalog that does not produce
BaseTable");
+ }
+
+ /**
+ * Federation-aware {@code updateTable}: applies the update against the
underlying (remote)
+ * catalog, including the staged-create path used by federated table
creation.
+ */
+ private LoadTableResponse tableUpdateInternal(TableIdentifier ident,
UpdateTableRequest request) {
+ if (isCreate(request)) {
+ // this is a hacky way to get TableOperations for an uncommitted table
+ Optional<Integer> formatVersion =
+ request.updates().stream()
+ .filter(update -> update instanceof
MetadataUpdate.UpgradeFormatVersion)
+ .map(update -> ((MetadataUpdate.UpgradeFormatVersion)
update).formatVersion())
+ .findFirst();
+
+ Schema schema =
+ request.updates().stream()
+ .filter(update -> update instanceof MetadataUpdate.AddSchema)
+ .map(update -> ((MetadataUpdate.AddSchema) update).schema())
+ .findFirst()
+ .orElse(EMPTY_SCHEMA);
+
+ Catalog.TableBuilder tableBuilder = getCatalog().buildTable(ident,
schema);
+
+ TableMetadata.Builder changedMetadata =
+
formatVersion.map(TableMetadata::buildFromEmpty).orElse(TableMetadata.buildFromEmpty());
+ request.updates().forEach(update -> update.applyTo(changedMetadata));
+
+ TableMetadata changedTableMeta = changedMetadata.build();
+ tableBuilder.withPartitionSpec(changedTableMeta.spec());
+ tableBuilder.withSortOrder(changedTableMeta.sortOrder());
+ tableBuilder.withLocation(changedTableMeta.location());
+ tableBuilder.withProperty(FORMAT_VERSION,
String.valueOf(changedTableMeta.formatVersion()));
+ tableBuilder.withProperties(changedTableMeta.properties());
+
+ Transaction transaction = tableBuilder.createOrReplaceTransaction();
+ if (transaction instanceof BaseTransaction) {
+ BaseTransaction baseTransaction = (BaseTransaction) transaction;
+
+ return LoadTableResponse.builder()
+ .withTableMetadata(create(baseTransaction, request))
+ .build();
+ } else {
+ throw new IllegalStateException(
+ "Cannot wrap catalog that does not produce BaseTransaction");
+ }
+
+ } else {
+ return CatalogHandlers.updateTable(getCatalog(), ident, request);
+ }
+ }
+
+ /**
+ * Federation-aware {@code loadTable}: loads the table from the underlying
(remote) catalog and
+ * extracts client-facing FileIO/credential properties from {@code
table.io()}.
+ */
+ private LoadTableResponse loadTableInternal(TableIdentifier ident) {
+ Table table = getCatalog().loadTable(ident);
+
+ if (table instanceof BaseTable) {
+ return buildLoadTableResponseFromFileIo(ident, (BaseTable) table);
+ } else if (table instanceof BaseMetadataTable) {
+ // metadata tables are loaded on the client side, return
NoSuchTableException for now
+ throw new NoSuchTableException("Table does not exist: %s",
ident.toString());
+ }
+
+ throw new IllegalStateException("Cannot wrap catalog that does not produce
BaseTable");
+ }
+
+ /**
+ * Builds a {@link LoadTableResponse} from a remote {@link BaseTable},
exposing the client-facing
+ * FileIO and credential properties extracted from {@code table.io()},
including the refreshable
+ * vended credentials and refresh properties for the remote storage.
+ *
+ * @param ident the table identifier, used to build the credential-refresh
endpoint.
+ * @param table the remote base table whose {@code io()} carries the storage
credentials.
+ * @return the load-table response including FileIO-derived client config
and vended credentials.
+ */
+ private LoadTableResponse buildLoadTableResponseFromFileIo(
+ TableIdentifier ident, BaseTable table) {
+ Map<String, String> properties = retrieveFileIOProperties(table.io());
+ Map<String, String> filteredCredentialProperties =
+ CredentialPropertyUtils.filterCredentialProperties(properties);
+ return LoadTableResponse.builder()
+ .withTableMetadata(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(filteredCredentialProperties)
+ .addAllConfig(
+ IcebergRESTUtils.buildRefreshProps(
+ catalogCredentialManager.catalogName(), ident,
filteredCredentialProperties))
+ .addAllCredentials(
+ IcebergRESTUtils.buildStorageCreds(
+ catalogCredentialManager.catalogName(), ident, table.io()))
+ .build();
+ }
+
+ private static boolean isCreate(UpdateTableRequest request) {
+ boolean isCreate =
+ request.requirements().stream()
+
.anyMatch(UpdateRequirement.AssertTableDoesNotExist.class::isInstance);
+
+ if (isCreate) {
+ List<UpdateRequirement> invalidRequirements =
+ request.requirements().stream()
+ .filter(req -> !(req instanceof
UpdateRequirement.AssertTableDoesNotExist))
+ .collect(Collectors.toList());
+ Preconditions.checkArgument(
+ invalidRequirements.isEmpty(), "Invalid create requirements: %s",
invalidRequirements);
+ }
+
+ return isCreate;
+ }
+
+ private static TableMetadata create(BaseTransaction baseTransaction,
UpdateTableRequest request) {
+ // the only valid requirement is that the table will be created
+ TableOperations ops = baseTransaction.underlyingOps();
+ request.requirements().forEach(requirement ->
requirement.validate(ops.current()));
+
+ TableMetadata.Builder builder =
TableMetadata.buildFrom(baseTransaction.currentMetadata());
+ request
+ .updates()
+ .forEach(
+ update -> {
+ if (shouldApplyMetadataUpdateAfterBuilder(update)) {
+ update.applyTo(builder);
+ }
+ });
+
+ // create transactions do not retry. if the table exists, retrying is not
a solution
+ ops.commit(null, builder.build());
+
+ return ops.current();
+ }
+
+ /**
+ * Returns {@code false} for updates already reflected through {@link
Catalog.TableBuilder} during
+ * staged create; those must not be applied again on {@link
TableMetadata.Builder}.
+ */
+ @VisibleForTesting
+ static boolean shouldApplyMetadataUpdateAfterBuilder(MetadataUpdate update) {
+ if (update instanceof MetadataUpdate.UpgradeFormatVersion) {
+ return false;
+ }
+
+ if (update instanceof MetadataUpdate.AddSchema) {
+ return false;
+ }
+
+ if (update instanceof MetadataUpdate.SetCurrentSchema) {
+ return false;
+ }
+
+ if (update instanceof MetadataUpdate.RemoveSchemas) {
+ return false;
+ }
+
+ if (update instanceof MetadataUpdate.SetLocation) {
+ return false;
+ }
+
+ if (update instanceof MetadataUpdate.SetProperties) {
+ return false;
+ }
+
+ if (update instanceof MetadataUpdate.RemoveProperties) {
+ return false;
+ }
+
+ if (update instanceof MetadataUpdate.AddSortOrder) {
+ return false;
+ }
+
+ if (update instanceof MetadataUpdate.SetDefaultSortOrder) {
+ return false;
+ }
+
+ if (update instanceof MetadataUpdate.AddPartitionSpec) {
+ return false;
+ }
+
+ if (update instanceof MetadataUpdate.SetDefaultPartitionSpec) {
+ return false;
+ }
+
+ if (update instanceof MetadataUpdate.RemovePartitionSpecs) {
+ return false;
+ }
+
+ return true;
+ }
+
+ private static Map<String, String> retrieveFileIOProperties(FileIO fileIO) {
+ return fileIO instanceof InMemoryFileIO
+ ? Maps.newHashMap()
+ : new HashMap<>(fileIO.properties());
+ }
+}
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergCatalogWrapperManager.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergCatalogWrapperManager.java
index 8d8edcf185..f9561e3f8f 100644
---
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergCatalogWrapperManager.java
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergCatalogWrapperManager.java
@@ -23,11 +23,13 @@ import com.github.benmanes.caffeine.cache.Caffeine;
import com.github.benmanes.caffeine.cache.Scheduler;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import java.util.Locale;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.apache.gravitino.GravitinoEnv;
+import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergCatalogBackend;
import org.apache.gravitino.exceptions.NoSuchCatalogException;
import org.apache.gravitino.iceberg.common.IcebergConfig;
import org.apache.gravitino.iceberg.common.authentication.AuthenticationConfig;
@@ -127,7 +129,17 @@ public class IcebergCatalogWrapperManager implements
AutoCloseable {
@VisibleForTesting
protected CatalogWrapperForREST createCatalogWrapper(
String catalogName, IcebergConfig icebergConfig) {
- CatalogWrapperForREST rest = new CatalogWrapperForREST(catalogName,
icebergConfig);
+ // When the backend is a federated Iceberg REST catalog, use
FederatedCatalogWrapper so
+ // federation-aware behavior (FileIO property extraction, remote
credential vending, remote
+ // /v1/config defaults) is applied through polymorphic dispatch rather
than scattered
+ // instanceof checks. All other backends use the base
CatalogWrapperForREST.
+ IcebergCatalogBackend backend =
+ IcebergCatalogBackend.valueOf(
+
icebergConfig.get(IcebergConfig.CATALOG_BACKEND).toUpperCase(Locale.ROOT));
+ CatalogWrapperForREST rest =
+ backend == IcebergCatalogBackend.REST
+ ? new FederatedCatalogWrapper(catalogName, icebergConfig)
+ : new CatalogWrapperForREST(catalogName, icebergConfig);
AuthenticationConfig authenticationConfig =
new AuthenticationConfig(icebergConfig.getAllConfig());
if (authenticationConfig.isKerberosAuth() && rest.getCatalog() instanceof
SupportsKerberos) {
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 465ebfc138..9a4d74091d 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
@@ -55,6 +55,8 @@ 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.ImmutableLoadCredentialsResponse;
+import org.apache.iceberg.rest.responses.LoadCredentialsResponse;
import org.apache.iceberg.rest.responses.LoadTableResponse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -178,16 +180,58 @@ public class IcebergRESTUtils {
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)));
+ rewriteCredential(
+ catalogName, tableIdentifier, credential.prefix(),
credential.config()));
}
return List.copyOf(restCredentials);
}
+ /**
+ * Rewrites credentials returned by an upstream REST catalog so their {@code
+ * *.refresh-credentials-endpoint} entries point at this IRC instance
instead of the upstream
+ * catalog. Without this, a federated {@code getTableCredentials} would
return refresh endpoints
+ * scoped to the remote catalog's prefix, inconsistent with the
loadTable/createTable paths that
+ * already rewrite via {@link #buildStorageCreds}.
+ *
+ * @param catalogName IRC catalog name used to build refresh paths
+ * @param tableIdentifier table receiving the credentials
+ * @param upstream the credentials response returned by the upstream REST
catalog
+ * @return a credentials response with IRC-local refresh endpoints
+ */
+ public static LoadCredentialsResponse rewriteTableCredentials(
+ String catalogName, TableIdentifier tableIdentifier,
LoadCredentialsResponse upstream) {
+ ImmutableLoadCredentialsResponse.Builder builder =
ImmutableLoadCredentialsResponse.builder();
+ for (org.apache.iceberg.rest.credentials.Credential credential :
upstream.credentials()) {
+ builder.addCredentials(
+ rewriteCredential(
+ catalogName, tableIdentifier, credential.prefix(),
credential.config()));
+ }
+ return builder.build();
+ }
+
+ /**
+ * Filters an upstream credential's config to recognized credential
properties and re-applies
+ * IRC-local refresh-endpoint properties, dropping any refresh endpoint
scoped to the upstream
+ * catalog.
+ *
+ * @param catalogName IRC catalog name used to build refresh paths
+ * @param tableIdentifier table receiving the credential
+ * @param prefix storage location prefix for the credential
+ * @param config upstream credential config
+ * @return an Iceberg REST credential with IRC-local refresh endpoints
+ */
+ private static org.apache.iceberg.rest.credentials.Credential
rewriteCredential(
+ String catalogName,
+ TableIdentifier tableIdentifier,
+ String prefix,
+ Map<String, String> config) {
+ Map<String, String> filteredConfig =
CredentialPropertyUtils.filterCredentialProperties(config);
+ filteredConfig.putAll(buildRefreshProps(catalogName, tableIdentifier,
filteredConfig));
+ return toRESTCredential(prefix, ImmutableMap.copyOf(filteredConfig));
+ }
+
public static <T> Response ok(T t) {
return
Response.status(Response.Status.OK).entity(t).type(MediaType.APPLICATION_JSON).build();
}
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 55d30018a0..ce845ce020 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
@@ -75,6 +75,8 @@ 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.ImmutableRegisterTableRequest;
+import org.apache.iceberg.rest.requests.RegisterTableRequest;
import org.apache.iceberg.rest.requests.UpdateTableRequest;
import org.apache.iceberg.rest.responses.LoadCredentialsResponse;
import org.apache.iceberg.rest.responses.LoadTableResponse;
@@ -129,7 +131,9 @@ public class TestCatalogWrapperForREST {
"{\"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\"}}]}";
+ + "\"s3.session-token\":\"upstream-token\","
+ + "\"client.refresh-credentials-endpoint\":"
+ + "\"v1/upstream/namespaces/db/tables/tbl/credentials\"}}]}";
AtomicReference<String> requestPath = new AtomicReference<>();
HttpServer server = HttpServer.create(new InetSocketAddress(0), 0);
@@ -178,6 +182,11 @@ public class TestCatalogWrapperForREST {
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"));
+ // The upstream refresh endpoint is rewritten to this IRC catalog's
prefix ("local"), not the
+ // remote catalog's ("upstream"), matching the loadTable/createTable
federation paths.
+ Assertions.assertEquals(
+ "v1/local/namespaces/db/tables/tbl/credentials",
+ credential.config().get("client.refresh-credentials-endpoint"));
} finally {
server.stop(0);
}
@@ -552,8 +561,9 @@ public class TestCatalogWrapperForREST {
IcebergConstants.WAREHOUSE,
"s3://remote/warehouse"));
- Map<String, String> configToClients =
- CatalogWrapperForREST.buildCatalogConfigToClients(config, restCatalog);
+ // FederatedCatalogWrapper sources the client config from the remote
RESTCatalog's properties().
+ CatalogWrapperForREST wrapper = new StaticCatalogWrapperForREST("test",
config, restCatalog);
+ Map<String, String> configToClients =
wrapper.buildCatalogConfigToClients();
Assertions.assertEquals(
"org.apache.iceberg.aws.s3.S3FileIO",
configToClients.get(IcebergConstants.IO_IMPL));
@@ -570,33 +580,30 @@ public class TestCatalogWrapperForREST {
ImmutableMap.of(
IcebergConstants.CATALOG_BACKEND,
"hive",
+ IcebergConstants.URI,
+ "thrift://hive-metastore:9083",
IcebergConstants.IO_IMPL,
ResolvingFileIO.class.getName(),
IcebergConstants.WAREHOUSE,
"s3://bucket/warehouse"));
- Catalog catalog = mock(Catalog.class);
- Map<String, String> configToClients =
- CatalogWrapperForREST.buildCatalogConfigToClients(config, catalog);
+ // Base CatalogWrapperForREST sources the client config from static
catalog configuration.
+ CatalogWrapperForREST wrapper = new CatalogWrapperForREST("test", config);
+ Map<String, String> configToClients =
wrapper.buildCatalogConfigToClients();
Assertions.assertEquals(
ResolvingFileIO.class.getName(),
configToClients.get(IcebergConstants.IO_IMPL));
}
@Test
- void testNonRESTCatalogClientConfig() {
- Catalog catalog = mock(Catalog.class);
- IcebergConfig config =
- new IcebergConfig(
+ void testNonRestCatalogClientConfig() {
+ Map<String, String> configToClients =
+ CatalogWrapperForREST.filterCatalogConfigForClients(
ImmutableMap.of(
- IcebergConstants.CATALOG_BACKEND,
- "hive",
IcebergConstants.URI,
"thrift://hive-metastore:9083",
IcebergConstants.IO_IMPL,
"org.apache.iceberg.aws.s3.S3FileIO"));
- Map<String, String> configToClients =
- CatalogWrapperForREST.buildCatalogConfigToClients(config, catalog);
Assertions.assertFalse(configToClients.containsKey(IcebergConstants.URI));
Assertions.assertEquals(
"org.apache.iceberg.aws.s3.S3FileIO",
configToClients.get(IcebergConstants.IO_IMPL));
@@ -605,18 +612,52 @@ public class TestCatalogWrapperForREST {
@Test
void testCatalogClientConfigRejectsBadDataAccess() {
- Catalog catalog = mock(Catalog.class);
+ Map<String, String> source =
+ ImmutableMap.of(IcebergConstants.ICEBERG_ACCESS_DELEGATION,
"invalid-mode");
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> CatalogWrapperForREST.filterCatalogConfigForClients(source));
+ }
+
+ @Test
+ void testFederatedRegisterTableIncludesFileIo() {
+ RESTCatalog catalog = mock(RESTCatalog.class);
+ BaseTable table = mock(BaseTable.class);
+ TableOperations ops = mock(TableOperations.class);
+ FileIO fileIO = mock(FileIO.class);
+ when(catalog.registerTable(any(TableIdentifier.class),
anyString())).thenReturn(table);
+ when(table.operations()).thenReturn(ops);
+ when(ops.current()).thenReturn(minimalTableMetadataForStagedCreateTest());
+ when(table.io()).thenReturn(fileIO);
+ when(fileIO.properties())
+ .thenReturn(
+ ImmutableMap.of(
+ IcebergConstants.IO_IMPL,
+ "org.apache.iceberg.aws.s3.S3FileIO",
+ IcebergConstants.ICEBERG_S3_ENDPOINT,
+ "http://localhost:9000";));
+
IcebergConfig config =
new IcebergConfig(
ImmutableMap.of(
IcebergConstants.CATALOG_BACKEND,
- "hive",
- IcebergConstants.ICEBERG_ACCESS_DELEGATION,
- "invalid-mode"));
+ "memory",
+ IcebergConstants.WAREHOUSE,
+ "/tmp/warehouse"));
+ CatalogWrapperForREST wrapper = new StaticCatalogWrapperForREST("test",
config, catalog);
- Assertions.assertThrows(
- IllegalArgumentException.class,
- () -> CatalogWrapperForREST.buildCatalogConfigToClients(config,
catalog));
+ RegisterTableRequest request =
+ ImmutableRegisterTableRequest.builder()
+ .name("tbl")
+
.metadataLocation("s3://bucket/warehouse/tbl/metadata/v1.metadata.json")
+ .build();
+
+ LoadTableResponse response = wrapper.registerTable(Namespace.of("db"),
request, false);
+
+ Assertions.assertEquals(
+ "org.apache.iceberg.aws.s3.S3FileIO",
response.config().get(IcebergConstants.IO_IMPL));
+ Assertions.assertEquals(
+ "http://localhost:9000";,
response.config().get(IcebergConstants.ICEBERG_S3_ENDPOINT));
}
@Test
@@ -760,41 +801,41 @@ public class TestCatalogWrapperForREST {
void testPostBuilderMetadataSkipsHandledKinds() {
Schema schema = new Schema(Types.NestedField.required(1, "c",
Types.LongType.get()));
Assertions.assertFalse(
- CatalogWrapperForREST.shouldApplyMetadataUpdateAfterBuilder(
+ FederatedCatalogWrapper.shouldApplyMetadataUpdateAfterBuilder(
new MetadataUpdate.AddSchema(schema)));
Assertions.assertFalse(
- CatalogWrapperForREST.shouldApplyMetadataUpdateAfterBuilder(
+ FederatedCatalogWrapper.shouldApplyMetadataUpdateAfterBuilder(
new MetadataUpdate.UpgradeFormatVersion(2)));
Assertions.assertFalse(
- CatalogWrapperForREST.shouldApplyMetadataUpdateAfterBuilder(
+ FederatedCatalogWrapper.shouldApplyMetadataUpdateAfterBuilder(
new MetadataUpdate.SetCurrentSchema(-1)));
Assertions.assertFalse(
- CatalogWrapperForREST.shouldApplyMetadataUpdateAfterBuilder(
+ FederatedCatalogWrapper.shouldApplyMetadataUpdateAfterBuilder(
new MetadataUpdate.SetLocation("file:///tmp/loc")));
Assertions.assertFalse(
- CatalogWrapperForREST.shouldApplyMetadataUpdateAfterBuilder(
+ FederatedCatalogWrapper.shouldApplyMetadataUpdateAfterBuilder(
new MetadataUpdate.SetProperties(ImmutableMap.of("k", "v"))));
Assertions.assertFalse(
- CatalogWrapperForREST.shouldApplyMetadataUpdateAfterBuilder(
+ FederatedCatalogWrapper.shouldApplyMetadataUpdateAfterBuilder(
new MetadataUpdate.RemoveProperties(Collections.singleton("k"))));
Assertions.assertFalse(
- CatalogWrapperForREST.shouldApplyMetadataUpdateAfterBuilder(
+ FederatedCatalogWrapper.shouldApplyMetadataUpdateAfterBuilder(
new
MetadataUpdate.AddPartitionSpec(PartitionSpec.unpartitioned())));
Assertions.assertFalse(
- CatalogWrapperForREST.shouldApplyMetadataUpdateAfterBuilder(
+ FederatedCatalogWrapper.shouldApplyMetadataUpdateAfterBuilder(
new
MetadataUpdate.SetDefaultPartitionSpec(PartitionSpec.unpartitioned().specId())));
Assertions.assertFalse(
- CatalogWrapperForREST.shouldApplyMetadataUpdateAfterBuilder(
+ FederatedCatalogWrapper.shouldApplyMetadataUpdateAfterBuilder(
new MetadataUpdate.AddSortOrder(SortOrder.unsorted())));
Assertions.assertFalse(
- CatalogWrapperForREST.shouldApplyMetadataUpdateAfterBuilder(
+ FederatedCatalogWrapper.shouldApplyMetadataUpdateAfterBuilder(
new
MetadataUpdate.SetDefaultSortOrder(SortOrder.unsorted().orderId())));
}
@Test
void testPostBuilderMetadataAllowsAssignUuid() {
Assertions.assertTrue(
- CatalogWrapperForREST.shouldApplyMetadataUpdateAfterBuilder(
+ FederatedCatalogWrapper.shouldApplyMetadataUpdateAfterBuilder(
new MetadataUpdate.AssignUUID(UUID.randomUUID().toString())));
}
@@ -906,8 +947,8 @@ public class TestCatalogWrapperForREST {
}
/**
- * Same derivation as {@link CatalogWrapperForREST#tableUpdateInternal} for
staged create: replay
- * metadata updates and read {@link TableMetadata#formatVersion()}.
+ * Same derivation as {@link FederatedCatalogWrapper#tableUpdateInternal}
for staged create:
+ * replay metadata updates and read {@link TableMetadata#formatVersion()}.
*/
private static String expectedFormatVersionStringAfterStagedUpdates(
Schema schema, Optional<Integer> formatVersionForUpgrade) {
@@ -925,7 +966,7 @@ public class TestCatalogWrapperForREST {
/**
* Minimal valid Iceberg staged-create update sequence so {@link
TableMetadata.Builder#build()}
- * succeeds in {@link CatalogWrapperForREST#tableUpdateInternal}.
+ * succeeds in {@link FederatedCatalogWrapper#tableUpdateInternal}.
*/
private static List<MetadataUpdate> stagedCreateMetadataUpdates(
Schema schema, Optional<Integer> formatVersion) {
@@ -969,7 +1010,9 @@ public class TestCatalogWrapperForREST {
}
}
- private static class StaticCatalogWrapperForREST extends
CatalogWrapperForREST {
+ // Extends FederatedCatalogWrapper so table operations route through the
federation-aware
+ // *Internal paths (FileIO extraction) against the injected catalog.
+ private static class StaticCatalogWrapperForREST extends
FederatedCatalogWrapper {
private final Catalog catalog;
StaticCatalogWrapperForREST(String catalogName, IcebergConfig config,
Catalog catalog) {
diff --git
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergCatalogWrapperManagerForREST.java
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergCatalogWrapperManagerForREST.java
index f81ea32633..2d4386b730 100644
---
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergCatalogWrapperManagerForREST.java
+++
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergCatalogWrapperManagerForREST.java
@@ -18,12 +18,15 @@
*/
package org.apache.gravitino.iceberg.service;
+import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Maps;
import java.util.Map;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.reflect.FieldUtils;
import org.apache.gravitino.GravitinoEnv;
import org.apache.gravitino.catalog.CatalogManager;
+import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants;
+import org.apache.gravitino.iceberg.common.IcebergConfig;
import org.apache.gravitino.iceberg.common.ops.IcebergCatalogWrapper;
import
org.apache.gravitino.iceberg.service.authorization.IcebergRESTServerContext;
import org.apache.gravitino.iceberg.service.provider.IcebergConfigProvider;
@@ -110,4 +113,43 @@ public class TestIcebergCatalogWrapperManagerForREST {
.getMessage()
.contains("gravitino.iceberg-rest.catalog-config-provider=dynamic-config-provider"));
}
+
+ @Test
+ public void testCreateFederatedWrapperForRestBackend() {
+ IcebergConfig icebergConfig =
+ new IcebergConfig(
+ ImmutableMap.of(
+ IcebergConstants.CATALOG_BACKEND,
+ "rest",
+ IcebergConstants.URI,
+ "http://localhost:8181";));
+
+ CatalogWrapperForREST wrapper = newManager().createCatalogWrapper("test",
icebergConfig);
+
+ Assertions.assertInstanceOf(FederatedCatalogWrapper.class, wrapper);
+ }
+
+ @Test
+ public void testCreateBaseWrapperForNonRestBackend() {
+ IcebergConfig icebergConfig =
+ new IcebergConfig(
+ ImmutableMap.of(
+ IcebergConstants.CATALOG_BACKEND,
+ "memory",
+ IcebergConstants.WAREHOUSE,
+ "/tmp/warehouse"));
+
+ CatalogWrapperForREST wrapper = newManager().createCatalogWrapper("test",
icebergConfig);
+
+ Assertions.assertFalse(wrapper instanceof FederatedCatalogWrapper);
+ Assertions.assertEquals(CatalogWrapperForREST.class, wrapper.getClass());
+ }
+
+ private static IcebergCatalogWrapperManager newManager() {
+ Map<String, String> config = Maps.newHashMap();
+ IcebergConfigProvider configProvider =
IcebergConfigProviderFactory.create(config);
+ configProvider.initialize(config);
+ return new IcebergCatalogWrapperManager(
+ config, configProvider, false, configProvider.getMetalakeName());
+ }
}
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 f7dd7cb2f0..2ad1f85c3d 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
@@ -45,6 +45,8 @@ 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.rest.responses.ImmutableLoadCredentialsResponse;
+import org.apache.iceberg.rest.responses.LoadCredentialsResponse;
import org.apache.iceberg.types.Types.IntegerType;
import org.apache.iceberg.types.Types.NestedField;
import org.apache.iceberg.types.Types.StringType;
@@ -318,4 +320,34 @@ public class TestIcebergRESTUtils {
Assertions.assertFalse(config.containsKey("client.refresh-credentials-endpoint"));
}
+
+ @Test
+ void testRewriteTableCredentials() {
+ TableIdentifier table = TableIdentifier.of(Namespace.of("db"), "tbl");
+ LoadCredentialsResponse upstream =
+ ImmutableLoadCredentialsResponse.builder()
+ .addCredentials(
+ IcebergRESTUtils.toRESTCredential(
+ "s3://bucket/db/tbl/",
+ ImmutableMap.of(
+ "s3.session-token",
+ "upstream-token",
+ "s3.session-token-expires-at-ms",
+ "123",
+ "client.refresh-credentials-endpoint",
+ "v1/upstream/namespaces/db/tables/tbl/credentials")))
+ .build();
+
+ LoadCredentialsResponse rewritten =
+ IcebergRESTUtils.rewriteTableCredentials("irc1", table, upstream);
+
+ Assertions.assertEquals(1, rewritten.credentials().size());
+ Credential credential = rewritten.credentials().get(0);
+ Assertions.assertEquals("s3://bucket/db/tbl/", credential.prefix());
+ Assertions.assertEquals("upstream-token",
credential.config().get("s3.session-token"));
+ // The refresh endpoint is re-scoped to the IRC catalog, dropping the
upstream-scoped one.
+ Assertions.assertEquals(
+ "v1/irc1/namespaces/db/tables/tbl/credentials",
+ credential.config().get("client.refresh-credentials-endpoint"));
+ }
}
