diff --git a/catalogs/catalog-common/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergConstants.java b/catalogs/catalog-common/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergConstants.java index e7f97b6bdf6..120cebb6196 100644 --- a/catalogs/catalog-common/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergConstants.java +++ b/catalogs/catalog-common/src/main/java/org/apache/gravitino/catalog/lakehouse/iceberg/IcebergConstants.java @@ -77,6 +77,20 @@ public class IcebergConstants { public static final String AZURE_CLIENT_SECRET_TOKEN_CREDENTIAL_PROVIDER = "org.apache.gravitino.iceberg.common.credential.AzureClientSecretTokenCredentialProvider"; + /** Iceberg GCSFileIO OAuth2 access token property. */ + public static final String ICEBERG_GCS_OAUTH2_TOKEN = "gcs.oauth2.token"; + + /** Iceberg GCSFileIO OAuth2 token expiry property (epoch millis). */ + public static final String ICEBERG_GCS_OAUTH2_TOKEN_EXPIRES_AT = "gcs.oauth2.token-expires-at"; + + /** + * Whether Iceberg GCSFileIO should refresh OAuth2 tokens via a credentials endpoint. Defaults to + * true in Iceberg; Gravitino disables it when minting a token from {@code + * gcs-service-account-file} because that path has no table credentials refresh endpoint. + */ + public static final String ICEBERG_GCS_OAUTH2_REFRESH_CREDENTIALS_ENABLED = + "gcs.oauth2.refresh-credentials-enabled"; + // Iceberg Table properties constants public static final String COMMENT = "comment"; diff --git a/catalogs/catalog-fileset/src/main/java/org/apache/gravitino/catalog/fileset/FilesetCatalogOperations.java b/catalogs/catalog-fileset/src/main/java/org/apache/gravitino/catalog/fileset/FilesetCatalogOperations.java index 7753ff6d42d..ac75ec03c23 100644 --- a/catalogs/catalog-fileset/src/main/java/org/apache/gravitino/catalog/fileset/FilesetCatalogOperations.java +++ b/catalogs/catalog-fileset/src/main/java/org/apache/gravitino/catalog/fileset/FilesetCatalogOperations.java @@ -39,6 +39,7 @@ import com.google.common.util.concurrent.ThreadFactoryBuilder; import java.io.FileNotFoundException; import java.io.IOException; +import java.io.UncheckedIOException; import java.time.Instant; import java.util.ArrayList; import java.util.Arrays; @@ -116,6 +117,7 @@ import org.apache.gravitino.metrics.MetricsSystem; import org.apache.gravitino.metrics.source.FilesetCatalogMetricsSource; import org.apache.gravitino.utils.ClassLoaderResourceCleanerUtils; +import org.apache.gravitino.utils.ExceptionMessages; import org.apache.gravitino.utils.FilesetUtil; import org.apache.gravitino.utils.NameIdentifierUtil; import org.apache.gravitino.utils.NamespaceUtil; @@ -347,7 +349,7 @@ public NameIdentifier[] listFilesets(Namespace namespace) throws NoSuchSchemaExc .map(f -> NameIdentifier.of(namespace, f.name())) .toArray(NameIdentifier[]::new); } catch (IOException e) { - throw new RuntimeException("Failed to list filesets under namespace " + namespace, e); + throw ExceptionMessages.wrap("Failed to list filesets under namespace " + namespace, e); } } @@ -369,7 +371,7 @@ public Fileset loadFileset(NameIdentifier ident) throws NoSuchFilesetException { } catch (NoSuchEntityException exception) { throw new NoSuchFilesetException(exception, FILESET_DOES_NOT_EXIST_MSG, ident); } catch (IOException ioe) { - throw new RuntimeException("Failed to load fileset %s" + ident, ioe); + throw ExceptionMessages.wrap("Failed to load fileset %s" + ident, ioe); } } @@ -416,7 +418,7 @@ public FileInfo[] listFiles(NameIdentifier filesetIdent, String locationName, St .toArray(FileInfo[]::new); } catch (IOException e) { - throw new RuntimeException("Failed to list files in fileset" + filesetIdent, e); + throw ExceptionMessages.wrap("Failed to list files in fileset" + filesetIdent, e); } } @@ -444,7 +446,7 @@ public Fileset createMultipleLocationFileset( throw new FilesetAlreadyExistsException("Fileset %s already exists", ident); } } catch (IOException ioe) { - throw new RuntimeException("Failed to check if fileset " + ident + " exists", ioe); + throw ExceptionMessages.wrap("Failed to check if fileset " + ident + " exists", ioe); } SchemaEntity schemaEntity; @@ -454,7 +456,7 @@ public Fileset createMultipleLocationFileset( } catch (NoSuchEntityException exception) { throw new NoSuchSchemaException(exception, SCHEMA_DOES_NOT_EXIST_MSG, schemaIdent); } catch (IOException ioe) { - throw new RuntimeException("Failed to load schema " + schemaIdent, ioe); + throw ExceptionMessages.wrap("Failed to load schema " + schemaIdent, ioe); } // For external fileset, the storageLocation must be set. @@ -557,7 +559,7 @@ public Fileset createMultipleLocationFileset( } } catch (IOException ioe) { - throw new RuntimeException("Failed to create fileset " + ident, ioe); + throw ExceptionMessages.wrap("Failed to create fileset " + ident, ioe); } } @@ -591,7 +593,7 @@ public Fileset createMultipleLocationFileset( try { store.put(filesetEntity, true /* overwrite */); } catch (IOException ioe) { - throw new RuntimeException("Failed to create fileset " + ident, ioe); + throw ExceptionMessages.wrap("Failed to create fileset " + ident, ioe); } return FilesetImpl.builder() @@ -654,7 +656,7 @@ public Fileset alterFileset(NameIdentifier ident, FilesetChange... changes) throw new NoSuchFilesetException(FILESET_DOES_NOT_EXIST_MSG, ident); } } catch (IOException ioe) { - throw new RuntimeException("Failed to load fileset " + ident, ioe); + throw ExceptionMessages.wrap("Failed to load fileset " + ident, ioe); } try { @@ -674,12 +676,12 @@ public Fileset alterFileset(NameIdentifier ident, FilesetChange... changes) .withAuditInfo(updatedFilesetEntity.auditInfo()) .build(); } catch (IOException ioe) { - throw new RuntimeException("Failed to update fileset " + ident, ioe); + throw ExceptionMessages.wrap("Failed to update fileset " + ident, ioe); } catch (NoSuchEntityException nsee) { throw new NoSuchFilesetException(nsee, FILESET_DOES_NOT_EXIST_MSG, ident); } catch (AlreadyExistsException aee) { // This is happened when renaming a fileset to an existing fileset name. - throw new RuntimeException( + throw ExceptionMessages.wrap( "Fileset with the same name " + ident.name() + " already exists", aee); } } @@ -735,8 +737,10 @@ public boolean dropFileset(NameIdentifier ident) { } catch (NoSuchEntityException ne) { LOG.warn("Fileset {} does not exist", ident); return false; + } catch (UncheckedIOException uioe) { + throw ExceptionMessages.wrap("Failed to delete fileset " + ident, uioe.getCause()); } catch (IOException ioe) { - throw new RuntimeException("Failed to delete fileset " + ident, ioe); + throw ExceptionMessages.wrap("Failed to delete fileset " + ident, ioe); } } @@ -760,7 +764,7 @@ public Schema createSchema(NameIdentifier ident, String comment, Map schemaPaths = getAndCheckSchemaPaths(ident.name(), properties); @@ -804,7 +808,7 @@ public Schema createSchema(NameIdentifier ident, String comment, Map getAndCheckCatalogStorageLocations(Map + locationName); } } catch (IOException e) { - throw new RuntimeException( + throw ExceptionMessages.wrap( "Failed to check if fileset catalog location exists: " + v, e); } } @@ -1450,7 +1454,7 @@ FileSystem getFileSystem(Path path, Map config) throws IOExcepti "Interrupted when getting FileSystem for path: {}, possibly the server is" + " shutting down or catalog is been dropped", path); - throw new RuntimeException("Interrupted when getting FileSystem for path: " + path, e); + throw ExceptionMessages.wrap("Interrupted when getting FileSystem for path: " + path, e); } catch (ExecutionException e) { Throwable cause = e.getCause(); if (cause instanceof IOException) { diff --git a/catalogs/catalog-glue/src/main/java/org/apache/gravitino/catalog/glue/GlueExceptionConverter.java b/catalogs/catalog-glue/src/main/java/org/apache/gravitino/catalog/glue/GlueExceptionConverter.java index 544726bf5bb..4ab3ed9c9b3 100644 --- a/catalogs/catalog-glue/src/main/java/org/apache/gravitino/catalog/glue/GlueExceptionConverter.java +++ b/catalogs/catalog-glue/src/main/java/org/apache/gravitino/catalog/glue/GlueExceptionConverter.java @@ -19,12 +19,15 @@ package org.apache.gravitino.catalog.glue; import org.apache.commons.lang3.StringUtils; +import org.apache.gravitino.exceptions.ForbiddenException; import org.apache.gravitino.exceptions.NoSuchSchemaException; import org.apache.gravitino.exceptions.NoSuchTableException; import org.apache.gravitino.exceptions.SchemaAlreadyExistsException; import org.apache.gravitino.exceptions.TableAlreadyExistsException; +import org.apache.gravitino.utils.ExceptionMessages; import software.amazon.awssdk.awscore.exception.AwsErrorDetails; import software.amazon.awssdk.core.exception.SdkClientException; +import software.amazon.awssdk.services.glue.model.AccessDeniedException; import software.amazon.awssdk.services.glue.model.AlreadyExistsException; import software.amazon.awssdk.services.glue.model.EntityNotFoundException; import software.amazon.awssdk.services.glue.model.GlueException; @@ -84,7 +87,10 @@ static RuntimeException toSchemaException(GlueException e, String context) { return new SchemaAlreadyExistsException(e, "%s already exists", context); } if (e instanceof InvalidInputException) { - return new IllegalArgumentException(context + ": " + e.getMessage(), e); + return ExceptionMessages.illegalArgument(context, e); + } + if (e instanceof AccessDeniedException) { + return new ForbiddenException(e, "Glue error: %s: %s", context, awsErrorDetail(e)); } return new RuntimeException("Glue error: " + context + ": " + awsErrorDetail(e), e); } @@ -104,7 +110,10 @@ static RuntimeException toTableException(GlueException e, String context) { return new TableAlreadyExistsException(e, "%s already exists", context); } if (e instanceof InvalidInputException) { - return new IllegalArgumentException(context + ": " + e.getMessage(), e); + return ExceptionMessages.illegalArgument(context, e); + } + if (e instanceof AccessDeniedException) { + return new ForbiddenException(e, "Glue error: %s: %s", context, awsErrorDetail(e)); } return new RuntimeException("Glue error: " + context + ": " + awsErrorDetail(e), e); } diff --git a/catalogs/catalog-glue/src/main/java/org/apache/gravitino/catalog/glue/GlueTableOperations.java b/catalogs/catalog-glue/src/main/java/org/apache/gravitino/catalog/glue/GlueTableOperations.java index 4891932f294..7b5af82f495 100644 --- a/catalogs/catalog-glue/src/main/java/org/apache/gravitino/catalog/glue/GlueTableOperations.java +++ b/catalogs/catalog-glue/src/main/java/org/apache/gravitino/catalog/glue/GlueTableOperations.java @@ -32,6 +32,7 @@ import org.apache.gravitino.rel.partitions.IdentityPartition; import org.apache.gravitino.rel.partitions.Partition; import org.apache.gravitino.rel.partitions.Partitions; +import org.apache.gravitino.utils.ExceptionMessages; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import software.amazon.awssdk.services.glue.GlueClient; @@ -99,7 +100,7 @@ public String[] listPartitionNames() { nextToken = resp.nextToken(); } while (nextToken != null); } catch (GlueException e) { - throw new RuntimeException("Failed to list partitions for table " + tableName, e); + throw ExceptionMessages.wrap("Failed to list partitions for table " + tableName, e); } return names.toArray(new String[0]); } @@ -121,7 +122,7 @@ public Partition[] listPartitions() { nextToken = resp.nextToken(); } while (nextToken != null); } catch (GlueException e) { - throw new RuntimeException("Failed to list partitions for table " + tableName, e); + throw ExceptionMessages.wrap("Failed to list partitions for table " + tableName, e); } return partitions.toArray(new Partition[0]); } @@ -141,7 +142,7 @@ public Partition getPartition(String partitionName) throws NoSuchPartitionExcept throw new NoSuchPartitionException( e, "Partition %s does not exist in table %s", partitionName, tableName); } catch (GlueException e) { - throw new RuntimeException("Failed to get partition " + partitionName, e); + throw ExceptionMessages.wrap("Failed to get partition " + partitionName, e); } } @@ -181,7 +182,7 @@ public Partition addPartition(Partition partition) throws PartitionAlreadyExists throw new PartitionAlreadyExistsException( e, "Partition %s already exists in table %s", partition.name(), tableName); } catch (GlueException e) { - throw new RuntimeException("Failed to add partition " + partition.name(), e); + throw ExceptionMessages.wrap("Failed to add partition " + partition.name(), e); } LOG.info("Added partition {} to {}.{}", partition.name(), dbName, tableName); @@ -210,7 +211,7 @@ public boolean dropPartition(String partitionName) { } catch (EntityNotFoundException e) { return false; } catch (GlueException e) { - throw new RuntimeException("Failed to drop partition " + partitionName, e); + throw ExceptionMessages.wrap("Failed to drop partition " + partitionName, e); } } diff --git a/catalogs/catalog-glue/src/test/java/org/apache/gravitino/catalog/glue/TestGlueExceptionConverter.java b/catalogs/catalog-glue/src/test/java/org/apache/gravitino/catalog/glue/TestGlueExceptionConverter.java index 33ff92a56e6..23a32114a7b 100644 --- a/catalogs/catalog-glue/src/test/java/org/apache/gravitino/catalog/glue/TestGlueExceptionConverter.java +++ b/catalogs/catalog-glue/src/test/java/org/apache/gravitino/catalog/glue/TestGlueExceptionConverter.java @@ -24,6 +24,7 @@ import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertTrue; +import org.apache.gravitino.exceptions.ForbiddenException; import org.apache.gravitino.exceptions.NoSuchSchemaException; import org.apache.gravitino.exceptions.NoSuchTableException; import org.apache.gravitino.exceptions.SchemaAlreadyExistsException; @@ -97,7 +98,7 @@ public void testSchemaAccessDeniedKeepsAwsMessage() { RuntimeException converted = GlueExceptionConverter.toSchemaException(e, "schema drop_me"); - assertEquals(RuntimeException.class, converted.getClass()); + assertInstanceOf(ForbiddenException.class, converted); assertSame(e, converted.getCause()); String message = converted.getMessage(); assertTrue(message.contains("schema drop_me"), message); @@ -120,7 +121,7 @@ public void testTableAccessDeniedKeepsAwsMessage() { RuntimeException converted = GlueExceptionConverter.toTableException(e, "table ctas_test"); - assertEquals(RuntimeException.class, converted.getClass()); + assertInstanceOf(ForbiddenException.class, converted); assertSame(e, converted.getCause()); String message = converted.getMessage(); assertTrue(message.contains("table ctas_test"), message); diff --git a/catalogs/catalog-hive/src/main/java/org/apache/gravitino/catalog/hive/HiveCatalogOperations.java b/catalogs/catalog-hive/src/main/java/org/apache/gravitino/catalog/hive/HiveCatalogOperations.java index d2e17aa6c7e..afd52501fd8 100644 --- a/catalogs/catalog-hive/src/main/java/org/apache/gravitino/catalog/hive/HiveCatalogOperations.java +++ b/catalogs/catalog-hive/src/main/java/org/apache/gravitino/catalog/hive/HiveCatalogOperations.java @@ -89,6 +89,7 @@ import org.apache.gravitino.rel.indexes.Index; import org.apache.gravitino.rel.types.Type; import org.apache.gravitino.utils.ClassLoaderResourceCleanerUtils; +import org.apache.gravitino.utils.ExceptionMessages; import org.apache.gravitino.utils.PrincipalUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -493,7 +494,7 @@ private HiveTableHandle loadHiveTable(NameIdentifier tableIdent) { return new HiveTableHandle(table, clientPool); } catch (InterruptedException e) { - throw new RuntimeException( + throw ExceptionMessages.wrap( "Failed to load Hive table " + tableIdent.name() + " from Hive metastore", e); } } diff --git a/catalogs/catalog-hive/src/main/java/org/apache/gravitino/catalog/hive/HiveTableOperations.java b/catalogs/catalog-hive/src/main/java/org/apache/gravitino/catalog/hive/HiveTableOperations.java index 5ba130a8f3a..bf78f57f3ec 100644 --- a/catalogs/catalog-hive/src/main/java/org/apache/gravitino/catalog/hive/HiveTableOperations.java +++ b/catalogs/catalog-hive/src/main/java/org/apache/gravitino/catalog/hive/HiveTableOperations.java @@ -36,6 +36,7 @@ import org.apache.gravitino.rel.SupportsPartitions; import org.apache.gravitino.rel.partitions.IdentityPartition; import org.apache.gravitino.rel.partitions.Partition; +import org.apache.gravitino.utils.ExceptionMessages; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -56,7 +57,7 @@ public String[] listPartitionNames() { .clientPool() .run(c -> c.listPartitionNames(tableHandle.table(), (short) -1).toArray(new String[0])); } catch (InterruptedException e) { - throw new RuntimeException( + throw ExceptionMessages.wrap( "Failed to list partition names of table " + tableHandle.name() + "from Hive Metastore", e); } @@ -70,7 +71,7 @@ public Partition[] listPartitions() { .run(c -> c.listPartitions(tableHandle.table(), (short) -1)) .toArray(new Partition[0]); } catch (InterruptedException e) { - throw new RuntimeException( + throw ExceptionMessages.wrap( "Failed to list partitions of table " + tableHandle.name() + "from Hive Metastore", e); } } @@ -81,7 +82,7 @@ public Partition getPartition(String partitionName) throws NoSuchPartitionExcept return tableHandle.clientPool().run(c -> c.getPartition(tableHandle.table(), partitionName)); } catch (InterruptedException e) { - throw new RuntimeException( + throw ExceptionMessages.wrap( "Failed to get partition " + partitionName + " of table " @@ -172,7 +173,7 @@ public boolean dropPartition(String partitionName) { return false; } catch (InterruptedException e) { - throw new RuntimeException( + throw ExceptionMessages.wrap( "Failed to get partition " + partitionName + " of table " diff --git a/catalogs/catalog-hive/src/main/java/org/apache/gravitino/catalog/hive/HiveViewCatalogOperations.java b/catalogs/catalog-hive/src/main/java/org/apache/gravitino/catalog/hive/HiveViewCatalogOperations.java index f3d078ff2e7..f571fc4eeae 100644 --- a/catalogs/catalog-hive/src/main/java/org/apache/gravitino/catalog/hive/HiveViewCatalogOperations.java +++ b/catalogs/catalog-hive/src/main/java/org/apache/gravitino/catalog/hive/HiveViewCatalogOperations.java @@ -50,6 +50,7 @@ import org.apache.gravitino.rel.View; import org.apache.gravitino.rel.ViewCatalog; import org.apache.gravitino.rel.ViewChange; +import org.apache.gravitino.utils.ExceptionMessages; import org.apache.gravitino.utils.PrincipalUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -90,7 +91,7 @@ public NameIdentifier[] listViews(Namespace namespace) throws NoSuchSchemaExcept .map(name -> NameIdentifier.of(namespace, name)) .toArray(NameIdentifier[]::new); } catch (InterruptedException e) { - throw new RuntimeException("Failed to list Hive views in " + namespace, e); + throw ExceptionMessages.wrap("Failed to list Hive views in " + namespace, e); } } @@ -157,9 +158,9 @@ public View createView( } catch (TableAlreadyExistsException e) { throw new ViewAlreadyExistsException(e, "View %s already exists in Hive Metastore", ident); } catch (InterruptedException e) { - throw new RuntimeException("Failed to create Hive view " + ident, e); + throw ExceptionMessages.wrap("Failed to create Hive view " + ident, e); } catch (Exception e) { - throw new RuntimeException("Failed to create Hive view " + ident, e); + throw ExceptionMessages.wrap("Failed to create Hive view " + ident, e); } } @@ -264,9 +265,9 @@ public View alterView(NameIdentifier ident, ViewChange... changes) } catch (UnsupportedOperationException e) { throw e; } catch (InterruptedException e) { - throw new RuntimeException("Failed to alter Hive view " + ident, e); + throw ExceptionMessages.wrap("Failed to alter Hive view " + ident, e); } catch (Exception e) { - throw new RuntimeException("Failed to alter Hive view " + ident, e); + throw ExceptionMessages.wrap("Failed to alter Hive view " + ident, e); } } @@ -311,9 +312,9 @@ public boolean dropView(NameIdentifier ident) { } catch (NoSuchTableException e) { return false; } catch (InterruptedException e) { - throw new RuntimeException("Failed to drop Hive view " + ident, e); + throw ExceptionMessages.wrap("Failed to drop Hive view " + ident, e); } catch (Exception e) { - throw new RuntimeException("Failed to drop Hive view " + ident, e); + throw ExceptionMessages.wrap("Failed to drop Hive view " + ident, e); } } @@ -349,9 +350,9 @@ private HiveView loadHiveView(NameIdentifier ident) throws NoSuchViewException { } catch (NoSuchTableException e) { throw new NoSuchViewException(e, "View %s does not exist in Hive Metastore", ident); } catch (InterruptedException e) { - throw new RuntimeException("Failed to load Hive view " + ident, e); + throw ExceptionMessages.wrap("Failed to load Hive view " + ident, e); } catch (Exception e) { - throw new RuntimeException("Failed to load Hive view " + ident, e); + throw ExceptionMessages.wrap("Failed to load Hive view " + ident, e); } } diff --git a/catalogs/catalog-jdbc-doris/src/main/java/org/apache/gravitino/catalog/doris/operation/DorisTableOperations.java b/catalogs/catalog-jdbc-doris/src/main/java/org/apache/gravitino/catalog/doris/operation/DorisTableOperations.java index b002af801b1..8d0bf8464ac 100644 --- a/catalogs/catalog-jdbc-doris/src/main/java/org/apache/gravitino/catalog/doris/operation/DorisTableOperations.java +++ b/catalogs/catalog-jdbc-doris/src/main/java/org/apache/gravitino/catalog/doris/operation/DorisTableOperations.java @@ -70,6 +70,7 @@ import org.apache.gravitino.rel.indexes.Indexes; import org.apache.gravitino.rel.partitions.ListPartition; import org.apache.gravitino.rel.partitions.RangePartition; +import org.apache.gravitino.utils.ExceptionMessages; /** Table operations for Apache Doris. */ public class DorisTableOperations extends JdbcTableOperations { @@ -215,7 +216,7 @@ Map appendNecessaryProperties(Map properties) { .toString()); } } catch (Exception e) { - throw new RuntimeException("Failed to get the number of backend servers", e); + throw ExceptionMessages.wrap("Failed to get the number of backend servers", e); } } diff --git a/catalogs/catalog-kafka/src/main/java/org/apache/gravitino/catalog/kafka/KafkaCatalogOperations.java b/catalogs/catalog-kafka/src/main/java/org/apache/gravitino/catalog/kafka/KafkaCatalogOperations.java index 980b92b542c..85a5feaa29a 100644 --- a/catalogs/catalog-kafka/src/main/java/org/apache/gravitino/catalog/kafka/KafkaCatalogOperations.java +++ b/catalogs/catalog-kafka/src/main/java/org/apache/gravitino/catalog/kafka/KafkaCatalogOperations.java @@ -66,6 +66,7 @@ import org.apache.gravitino.meta.AuditInfo; import org.apache.gravitino.meta.SchemaEntity; import org.apache.gravitino.storage.IdGenerator; +import org.apache.gravitino.utils.ExceptionMessages; import org.apache.gravitino.utils.NamespaceUtil; import org.apache.gravitino.utils.PrincipalUtils; import org.apache.kafka.clients.admin.AdminClient; @@ -153,10 +154,9 @@ public void initialize( adminClient = AdminClient.create(adminClientConfig); } catch (KafkaException e) { if (e.getCause() instanceof ConfigException) { - throw new IllegalArgumentException( - "Invalid configuration for Kafka AdminClient: " + e.getCause().getMessage(), e); + throw ExceptionMessages.illegalArgument("Invalid configuration for Kafka AdminClient", e); } - throw new RuntimeException("Failed to create Kafka AdminClient", e); + throw ExceptionMessages.wrap("Failed to create Kafka AdminClient", e); } createDefaultSchemaIfNecessary(); } @@ -173,13 +173,9 @@ public NameIdentifier[] listTopics(Namespace namespace) throws NoSuchSchemaExcep .map(name -> NameIdentifier.of(namespace, name)) .toArray(NameIdentifier[]::new); } catch (ExecutionException e) { - throw new RuntimeException( - String.format( - "Failed to list topics under the schema %s: %s", - namespace, e.getCause().getMessage()), - e); + throw ExceptionMessages.wrap("Failed to list topics under the schema " + namespace, e); } catch (InterruptedException e) { - throw new RuntimeException("Failed to list topics under the schema " + namespace, e); + throw ExceptionMessages.wrap("Failed to list topics under the schema " + namespace, e); } } @@ -226,10 +222,10 @@ public Topic loadTopic(NameIdentifier ident) throws NoSuchTopicException { if (e.getCause() instanceof UnknownTopicOrPartitionException) { throw new NoSuchTopicException(e, "Topic %s does not exist", ident); } else { - throw new RuntimeException("Failed to load topic " + ident.name() + " from Kafka", e); + throw ExceptionMessages.wrap("Failed to load topic " + ident.name() + " from Kafka", e); } } catch (InterruptedException e) { - throw new RuntimeException("Failed to load topic " + ident.name() + " from Kafka", e); + throw ExceptionMessages.wrap("Failed to load topic " + ident.name() + " from Kafka", e); } LOG.info("Loaded topic {} from Kafka", ident); @@ -297,18 +293,16 @@ public Topic createTopic( throw new TopicAlreadyExistsException(e, "Topic %s already exists", ident); } else if (e.getCause() instanceof InvalidReplicationFactorException) { - throw new IllegalArgumentException( - "Invalid replication factor for topic " + ident + e.getCause().getMessage(), e); + throw ExceptionMessages.illegalArgument("Invalid replication factor for topic " + ident, e); } else if (e.getCause() instanceof InvalidConfigurationException) { - throw new IllegalArgumentException( - "Invalid properties for topic " + ident + e.getCause().getMessage(), e); + throw ExceptionMessages.illegalArgument("Invalid properties for topic " + ident, e); } else { - throw new RuntimeException("Failed to create topic in Kafka" + ident, e); + throw ExceptionMessages.wrap("Failed to create topic in Kafka " + ident, e); } } catch (InterruptedException e) { - throw new RuntimeException("Failed to create topic in Kafka" + ident, e); + throw ExceptionMessages.wrap("Failed to create topic in Kafka " + ident, e); } } @@ -381,10 +375,10 @@ public boolean dropTopic(NameIdentifier ident) { if (e.getCause() instanceof UnknownTopicOrPartitionException) { return false; } else { - throw new RuntimeException("Failed to drop topic " + ident.name() + " from Kafka", e); + throw ExceptionMessages.wrap("Failed to drop topic " + ident.name() + " from Kafka", e); } } catch (InterruptedException e) { - throw new RuntimeException("Failed to drop topic " + ident.name() + " from Kafka", e); + throw ExceptionMessages.wrap("Failed to drop topic " + ident.name() + " from Kafka", e); } } @@ -397,7 +391,7 @@ public NameIdentifier[] listSchemas(Namespace namespace) throws NoSuchCatalogExc .map(s -> NameIdentifier.of(namespace, s.name())) .toArray(NameIdentifier[]::new); } catch (IOException e) { - throw new RuntimeException("Failed to list schemas under namespace " + namespace, e); + throw ExceptionMessages.wrap("Failed to list schemas under namespace " + namespace, e); } } @@ -426,7 +420,7 @@ public Schema loadSchema(NameIdentifier ident) throws NoSuchSchemaException { } catch (NoSuchEntityException exception) { throw new NoSuchSchemaException(exception, "Schema %s does not exist", ident); } catch (IOException ioe) { - throw new RuntimeException("Failed to load schema " + ident, ioe); + throw ExceptionMessages.wrap("Failed to load schema " + ident, ioe); } } @@ -534,8 +528,16 @@ private void doPartitionCountIncrement(String topicName, int newPartitionCount) Collections.singletonMap(topicName, NewPartitions.increaseTo(newPartitionCount))) .all() .get(); - } catch (Exception e) { - throw new RuntimeException("Failed to increase partition count for topic " + topicName, e); + } catch (ExecutionException e) { + if (e.getCause() instanceof InvalidConfigurationException + || e.getCause() instanceof IllegalArgumentException) { + throw ExceptionMessages.illegalArgument( + "Failed to increase partition count for topic " + topicName, e); + } + throw ExceptionMessages.wrap("Failed to increase partition count for topic " + topicName, e); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw ExceptionMessages.wrap("Failed to increase partition count for topic " + topicName, e); } } @@ -546,10 +548,19 @@ private void doAlterTopicConfig(String topicName, List alterConfi .incrementalAlterConfigs(Collections.singletonMap(topicResource, alterConfigOps)) .all() .get(); - } catch (UnknownTopicOrPartitionException e) { - throw new NoSuchTopicException(e, "Topic %s does not exist", topicName); - } catch (Exception e) { - throw new RuntimeException("Failed to alter topic properties for topic " + topicName, e); + } catch (ExecutionException e) { + if (e.getCause() instanceof UnknownTopicOrPartitionException) { + throw new NoSuchTopicException(e, "Topic %s does not exist", topicName); + } + if (e.getCause() instanceof InvalidConfigurationException + || e.getCause() instanceof IllegalArgumentException) { + throw ExceptionMessages.illegalArgument( + "Failed to alter topic properties for topic " + topicName, e); + } + throw ExceptionMessages.wrap("Failed to alter topic properties for topic " + topicName, e); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw ExceptionMessages.wrap("Failed to alter topic properties for topic " + topicName, e); } } @@ -590,7 +601,8 @@ private void createDefaultSchemaIfNecessary() { return; } } catch (IOException e) { - throw new RuntimeException("Failed to check if schema " + defaultSchemaIdent + " exists", e); + throw ExceptionMessages.wrap( + "Failed to check if schema " + defaultSchemaIdent + " exists", e); } // Create the default schema @@ -617,7 +629,7 @@ private void createDefaultSchemaIfNecessary() { try { store.put(defaultSchema, true /* overwrite */); } catch (IOException ioe) { - throw new RuntimeException("Failed to create default schema for Kafka catalog", ioe); + throw ExceptionMessages.wrap("Failed to create default schema for Kafka catalog", ioe); } } } diff --git a/catalogs/catalog-lakehouse-generic/src/main/java/org/apache/gravitino/catalog/lakehouse/generic/GenericCatalogOperations.java b/catalogs/catalog-lakehouse-generic/src/main/java/org/apache/gravitino/catalog/lakehouse/generic/GenericCatalogOperations.java index 445b87dcc41..11497e4d68b 100644 --- a/catalogs/catalog-lakehouse-generic/src/main/java/org/apache/gravitino/catalog/lakehouse/generic/GenericCatalogOperations.java +++ b/catalogs/catalog-lakehouse-generic/src/main/java/org/apache/gravitino/catalog/lakehouse/generic/GenericCatalogOperations.java @@ -66,6 +66,7 @@ import org.apache.gravitino.rel.expressions.transforms.Transform; import org.apache.gravitino.rel.indexes.Index; import org.apache.gravitino.storage.IdGenerator; +import org.apache.gravitino.utils.ExceptionMessages; /** Operations for interacting with a generic lakehouse catalog in Apache Gravitino. */ public class GenericCatalogOperations implements CatalogOperations, SupportsSchemas, TableCatalog { @@ -352,11 +353,9 @@ private ManagedTableOperations tableOps(NameIdentifier tableIdent) { } else if (t instanceof IllegalArgumentException) { throw (IllegalArgumentException) t; } else if (t instanceof IOException) { - throw new RuntimeException( - String.format("Failed to load table %s: %s", tableIdent, t.getMessage()), t); + throw ExceptionMessages.wrap("Failed to load table " + tableIdent, t); } else { - throw new RuntimeException( - String.format("Unexpected exception when loading table %s", tableIdent), t); + throw ExceptionMessages.wrap("Unexpected exception when loading table " + tableIdent, t); } } } diff --git a/catalogs/catalog-lakehouse-generic/src/main/java/org/apache/gravitino/catalog/lakehouse/lance/LanceTableOperations.java b/catalogs/catalog-lakehouse-generic/src/main/java/org/apache/gravitino/catalog/lakehouse/lance/LanceTableOperations.java index dd7dc2b314f..fb2bfaba1f3 100644 --- a/catalogs/catalog-lakehouse-generic/src/main/java/org/apache/gravitino/catalog/lakehouse/lance/LanceTableOperations.java +++ b/catalogs/catalog-lakehouse-generic/src/main/java/org/apache/gravitino/catalog/lakehouse/lance/LanceTableOperations.java @@ -66,6 +66,7 @@ import org.apache.gravitino.rel.indexes.Index; import org.apache.gravitino.storage.IdGenerator; import org.apache.gravitino.storage.relational.service.TableMetaService; +import org.apache.gravitino.utils.ExceptionMessages; import org.apache.gravitino.utils.PrincipalUtils; import org.lance.Dataset; import org.lance.ReadOptions; @@ -303,7 +304,7 @@ public boolean purgeTable(NameIdentifier ident) { } catch (NoSuchTableException e) { return false; } catch (Exception e) { - throw new RuntimeException("Failed to purge Lance dataset for table " + ident, e); + throw ExceptionMessages.wrap("Failed to purge Lance dataset for table " + ident, e); } } @@ -337,7 +338,7 @@ public boolean dropTable(NameIdentifier ident) { } catch (NoSuchTableException e) { return false; } catch (Exception e) { - throw new RuntimeException("Failed to drop Lance dataset for table " + ident, e); + throw ExceptionMessages.wrap("Failed to drop Lance dataset for table " + ident, e); } } @@ -355,7 +356,7 @@ private void dropLanceDataset(Table table) { && e.getMessage().contains("Not found:")) { LOG.warn("Lance dataset at {} was already deleted, skipping.", location); } else { - throw new RuntimeException("Failed to delete Lance dataset at " + location, e); + throw ExceptionMessages.wrap("Failed to delete Lance dataset at " + location, e); } } } @@ -433,7 +434,7 @@ Table createTableInternal( } throw e; } catch (Exception e) { - throw new RuntimeException("Failed to create Lance dataset at location " + location, e); + throw ExceptionMessages.wrap("Failed to create Lance dataset at location " + location, e); } } @@ -491,7 +492,9 @@ private Table loadTableInternal(NameIdentifier ident, boolean forAlter) { } catch (Exception e) { if (forAlter) { throw new IllegalStateException( - "Failed to load Lance schema before altering table " + ident, e); + ExceptionMessages.withCause( + "Failed to load Lance schema before altering table " + ident, e), + e); } LOG.debug( "Failed to load Lance schema from location {} for table {}. Return stored metadata.", @@ -577,9 +580,9 @@ private Table repairTableMetadata(NameIdentifier ident, Column[] columns, long d } catch (NoSuchEntityException e) { throw new NoSuchTableException(e, "Table %s does not exist", ident); } catch (EntityAlreadyExistsException e) { - throw new IllegalArgumentException("Failed to repair table " + ident, e); + throw ExceptionMessages.illegalArgument("Failed to repair table " + ident, e); } catch (IOException e) { - throw new RuntimeException("Failed to repair table " + ident, e); + throw ExceptionMessages.wrap("Failed to repair table " + ident, e); } } @@ -688,9 +691,10 @@ private Table recordCheckedEmptyVersion(NameIdentifier ident, long datasetVersio } catch (NoSuchEntityException e) { throw new NoSuchTableException(e, "Table %s does not exist", ident); } catch (EntityAlreadyExistsException e) { - throw new IllegalArgumentException("Failed to record empty version for table " + ident, e); + throw ExceptionMessages.illegalArgument( + "Failed to record empty version for table " + ident, e); } catch (IOException e) { - throw new RuntimeException("Failed to record empty version for table " + ident, e); + throw ExceptionMessages.wrap("Failed to record empty version for table " + ident, e); } } @@ -815,7 +819,7 @@ long handleLanceTableChange(Table table, TableChange[] changes) { } catch (RuntimeException e) { throw e; } catch (Exception e) { - throw new RuntimeException( + throw ExceptionMessages.wrap( "Failed to handle alterations to Lance dataset at location " + location, e); } } diff --git a/catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/utils/CatalogUtils.java b/catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/utils/CatalogUtils.java index be3febd7d5f..dfe5ac749e0 100644 --- a/catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/utils/CatalogUtils.java +++ b/catalogs/catalog-lakehouse-paimon/src/main/java/org/apache/gravitino/catalog/lakehouse/paimon/utils/CatalogUtils.java @@ -41,6 +41,7 @@ import org.apache.gravitino.catalog.lakehouse.paimon.authentication.kerberos.KerberosClient; import org.apache.gravitino.catalog.lakehouse.paimon.ops.PaimonBackendCatalogWrapper; import org.apache.gravitino.exceptions.ConnectionFailedException; +import org.apache.gravitino.utils.ExceptionMessages; import org.apache.hadoop.conf.Configuration; import org.apache.paimon.catalog.Catalog; import org.apache.paimon.catalog.CatalogContext; @@ -71,7 +72,7 @@ public static PaimonBackendCatalogWrapper loadCatalogBackend(PaimonConfig paimon Catalog catalog = loadCatalogBackendWithKerberosAuth(paimonConfig, configuration); return new PaimonBackendCatalogWrapper(catalog, kerberosClient); } catch (Exception e) { - throw new RuntimeException("Failed to login with kerberos", e); + throw ExceptionMessages.wrap("Failed to login with kerberos", e); } } else { throw new UnsupportedOperationException( diff --git a/catalogs/catalog-model/src/main/java/org/apache/gravitino/catalog/model/ModelCatalogOperations.java b/catalogs/catalog-model/src/main/java/org/apache/gravitino/catalog/model/ModelCatalogOperations.java index 1f2d14c6e0f..1d0c232366b 100644 --- a/catalogs/catalog-model/src/main/java/org/apache/gravitino/catalog/model/ModelCatalogOperations.java +++ b/catalogs/catalog-model/src/main/java/org/apache/gravitino/catalog/model/ModelCatalogOperations.java @@ -54,6 +54,7 @@ import org.apache.gravitino.model.ModelChange; import org.apache.gravitino.model.ModelVersion; import org.apache.gravitino.model.ModelVersionChange; +import org.apache.gravitino.utils.ExceptionMessages; import org.apache.gravitino.utils.NameIdentifierUtil; import org.apache.gravitino.utils.NamespaceUtil; import org.apache.gravitino.utils.PrincipalUtils; @@ -101,7 +102,7 @@ public NameIdentifier[] listModels(Namespace namespace) throws NoSuchSchemaExcep } catch (NoSuchEntityException e) { throw new NoSuchSchemaException(e, "Schema %s does not exist", namespace); } catch (IOException ioe) { - throw new RuntimeException("Failed to list models under namespace " + namespace, ioe); + throw ExceptionMessages.wrap("Failed to list models under namespace " + namespace, ioe); } } @@ -116,7 +117,7 @@ public Model getModel(NameIdentifier ident) throws NoSuchModelException { } catch (NoSuchEntityException e) { throw new NoSuchModelException(e, "Model %s does not exist", ident); } catch (IOException ioe) { - throw new RuntimeException("Failed to get model " + ident, ioe); + throw ExceptionMessages.wrap("Failed to get model " + ident, ioe); } } @@ -146,7 +147,7 @@ public Model registerModel(NameIdentifier ident, String comment, Map config.set(k.toString(), v.toString())); resolveMetastoreUriHosts(config); } catch (Exception e) { - throw new RuntimeException("Failed to create configuration", e); + throw ExceptionMessages.wrap("Failed to create configuration", e); } } diff --git a/catalogs/hive-metastore-common/src/main/java/org/apache/gravitino/hive/kerberos/KerberosClient.java b/catalogs/hive-metastore-common/src/main/java/org/apache/gravitino/hive/kerberos/KerberosClient.java index 57ed6949e18..8be06ff7873 100644 --- a/catalogs/hive-metastore-common/src/main/java/org/apache/gravitino/hive/kerberos/KerberosClient.java +++ b/catalogs/hive-metastore-common/src/main/java/org/apache/gravitino/hive/kerberos/KerberosClient.java @@ -36,6 +36,7 @@ import java.util.concurrent.TimeUnit; import org.apache.commons.lang3.StringUtils; import org.apache.gravitino.hive.client.HiveClient; +import org.apache.gravitino.utils.ExceptionMessages; import org.apache.gravitino.utils.FileFetcher; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hive.thrift.DelegationTokenIdentifier; @@ -111,7 +112,7 @@ public UserGroupInformation loginProxyUser(String currentUser) { return proxyUser; } catch (Exception e) { - throw new RuntimeException("Failed to create proxy user for Kerberos Hive client", e); + throw ExceptionMessages.wrap("Failed to create proxy user for Kerberos Hive client", e); } } diff --git a/common/src/main/java/org/apache/gravitino/utils/ExceptionMessages.java b/common/src/main/java/org/apache/gravitino/utils/ExceptionMessages.java new file mode 100644 index 00000000000..4818c3c6b1d --- /dev/null +++ b/common/src/main/java/org/apache/gravitino/utils/ExceptionMessages.java @@ -0,0 +1,155 @@ +/* + * 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.utils; + +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.UndeclaredThrowableException; +import java.util.Collections; +import java.util.IdentityHashMap; +import java.util.Set; +import java.util.concurrent.CompletionException; +import java.util.concurrent.ExecutionException; +import javax.annotation.Nullable; +import org.apache.commons.lang3.StringUtils; + +/** + * Helpers for preserving underlying system error messages when wrapping exceptions. + * + *

Catalog and server code often adds operation context when rethrowing. Context is useful, but + * it must not replace the upstream message that operators need to act on. + */ +public final class ExceptionMessages { + + private ExceptionMessages() {} + + /** + * Returns a non-blank diagnostic message from {@code throwable} or its cause chain. + * + *

Transparent wrappers such as {@link ExecutionException} are skipped when selecting candidate + * messages. Among remaining frames, the shallowest non-blank message is preferred so nested + * {@link #wrap(String, Throwable)} / {@link #withCause(String, Throwable)} context is not + * discarded. When a deeper non-blank reason is not already contained in that message, it is + * appended. + * + * @param throwable the throwable to inspect, may be null + * @return a useful message, or null if none is available + */ + @Nullable + public static String usefulMessage(@Nullable Throwable throwable) { + if (throwable == null) { + return null; + } + + Set visited = Collections.newSetFromMap(new IdentityHashMap<>()); + String firstUseful = null; + String lastUseful = null; + Throwable current = throwable; + while (current != null) { + if (!visited.add(current)) { + break; + } + + Throwable cause = current.getCause(); + if (!isTransparentWrapper(current)) { + String message = current.getMessage(); + // new RuntimeException(executionException) copies cause.toString() as the detail + // message; ignore that synthetic text and keep walking into the real cause. + boolean syntheticTransparentMessage = + cause != null + && isTransparentWrapper(cause) + && message != null + && message.equals(cause.toString()); + if (StringUtils.isNotBlank(message) && !syntheticTransparentMessage) { + if (firstUseful == null) { + firstUseful = message; + } + lastUseful = message; + } + } + + if (cause == null || cause == current) { + break; + } + current = cause; + } + + if (firstUseful == null) { + return null; + } + if (lastUseful == null || firstUseful.equals(lastUseful) || firstUseful.contains(lastUseful)) { + return firstUseful; + } + return firstUseful + ": " + lastUseful; + } + + /** + * Combines operation context with the underlying cause message. + * + *

If {@code context} already contains the useful cause message, {@code context} is returned + * unchanged. If there is no useful cause message, {@code context} is returned as-is. + * + * @param context operation/object context such as {@code "Failed to alter topic X"} + * @param throwable the underlying failure + * @return a message that preserves both context and the upstream reason when available + */ + public static String withCause(String context, @Nullable Throwable throwable) { + String useful = usefulMessage(throwable); + if (StringUtils.isBlank(useful)) { + return context; + } + if (StringUtils.isBlank(context)) { + return useful; + } + if (context.contains(useful)) { + return context; + } + return context + ": " + useful; + } + + /** + * Wraps {@code throwable} in a {@link RuntimeException} whose message includes both {@code + * context} and the underlying cause message. + * + * @param context operation/object context + * @param throwable the underlying failure + * @return a runtime exception suitable for rethrowing from catalog operations + */ + public static RuntimeException wrap(String context, Throwable throwable) { + return new RuntimeException(withCause(context, throwable), throwable); + } + + /** + * Creates an {@link IllegalArgumentException} whose message includes both {@code context} and the + * underlying cause message. + * + * @param context operation/object context + * @param throwable the underlying failure + * @return an illegal-argument exception for client-caused failures + */ + public static IllegalArgumentException illegalArgument(String context, Throwable throwable) { + return new IllegalArgumentException(withCause(context, throwable), throwable); + } + + private static boolean isTransparentWrapper(Throwable throwable) { + return throwable instanceof ExecutionException + || throwable instanceof CompletionException + || throwable instanceof InvocationTargetException + || throwable instanceof UndeclaredThrowableException; + } +} diff --git a/common/src/test/java/org/apache/gravitino/utils/TestExceptionMessages.java b/common/src/test/java/org/apache/gravitino/utils/TestExceptionMessages.java new file mode 100644 index 00000000000..79d8b9ec9a4 --- /dev/null +++ b/common/src/test/java/org/apache/gravitino/utils/TestExceptionMessages.java @@ -0,0 +1,143 @@ +/* + * 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.utils; + +import java.io.IOException; +import java.util.concurrent.ExecutionException; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +public class TestExceptionMessages { + + @Test + public void testUsefulMessageUnwrapsTransparentWrappers() { + Throwable root = new IllegalArgumentException("root reason"); + Throwable mid = new ExecutionException(root); + Throwable top = new RuntimeException(mid); + + Assertions.assertEquals("root reason", ExceptionMessages.usefulMessage(top)); + } + + @Test + public void testUsefulMessageKeepsImmediateContextAndAppendsDeeperReason() { + Throwable root = new IllegalArgumentException("root reason"); + Throwable mid = new ExecutionException(root); + Throwable top = new RuntimeException("wrapper", mid); + + Assertions.assertEquals("wrapper: root reason", ExceptionMessages.usefulMessage(top)); + } + + @Test + public void testUsefulMessageIgnoresBlankMessages() { + Throwable blankCause = new IOException(" "); + Throwable cause = new IOException("HMS connection refused", blankCause); + + Assertions.assertEquals( + "Failed to load table: HMS connection refused", + ExceptionMessages.withCause("Failed to load table", cause)); + Assertions.assertEquals("HMS connection refused", ExceptionMessages.usefulMessage(cause)); + } + + @Test + public void testUsefulMessageBlankOnlyChainReturnsNull() { + Throwable blank = new IOException("\n\t "); + Assertions.assertNull(ExceptionMessages.usefulMessage(blank)); + Assertions.assertEquals( + "Failed to load table", ExceptionMessages.withCause("Failed to load table", blank)); + } + + @Test + public void testUsefulMessageStopsOnTwoNodeCycle() { + RuntimeException a = new RuntimeException("a"); + RuntimeException b = new RuntimeException("b"); + a.initCause(b); + b.initCause(a); + + Assertions.assertEquals("a: b", ExceptionMessages.usefulMessage(a)); + } + + @Test + public void testUsefulMessageStopsOnThreeNodeCycle() { + RuntimeException a = new RuntimeException("a"); + RuntimeException b = new RuntimeException("b"); + RuntimeException c = new RuntimeException("c"); + a.initCause(b); + b.initCause(c); + c.initCause(a); + + Assertions.assertEquals("a: c", ExceptionMessages.usefulMessage(a)); + } + + @Test + public void testNestedWrapPreservesIntermediatePropertyContext() { + Throwable root = new IOException("write failed"); + Throwable inner = ExceptionMessages.wrap("Failed to write property: fs.s3a.endpoint", root); + RuntimeException outer = ExceptionMessages.wrap("Failed to create configuration", inner); + + Assertions.assertTrue( + outer.getMessage().contains("fs.s3a.endpoint"), + "nested wrap must keep intermediate property context, got: " + outer.getMessage()); + Assertions.assertEquals( + "Failed to create configuration: Failed to write property: fs.s3a.endpoint: write failed", + outer.getMessage()); + } + + @Test + public void testWithCauseAppendsUpstreamMessage() { + Throwable cause = + new IllegalArgumentException( + "Invalid value nonsense for configuration cleanup.policy: String must be one of:" + + " compact, delete"); + + String combined = + ExceptionMessages.withCause("Failed to alter topic properties for topic prop_probe", cause); + + Assertions.assertEquals( + "Failed to alter topic properties for topic prop_probe: Invalid value nonsense for" + + " configuration cleanup.policy: String must be one of: compact, delete", + combined); + } + + @Test + public void testWithCauseDoesNotDuplicateMessage() { + String context = "Failed to alter topic: bad value"; + Throwable cause = new IllegalArgumentException("bad value"); + + Assertions.assertEquals(context, ExceptionMessages.withCause(context, cause)); + } + + @Test + public void testWrapPreservesCauseAndMessage() { + Throwable cause = new IllegalStateException("glue denied"); + RuntimeException wrapped = ExceptionMessages.wrap("Glue error: schema drop_me", cause); + + Assertions.assertEquals("Glue error: schema drop_me: glue denied", wrapped.getMessage()); + Assertions.assertSame(cause, wrapped.getCause()); + } + + @Test + public void testIllegalArgumentPreservesCauseAndMessage() { + Throwable cause = new IllegalArgumentException("not allowed"); + IllegalArgumentException wrapped = + ExceptionMessages.illegalArgument("Invalid properties for topic t1", cause); + + Assertions.assertEquals("Invalid properties for topic t1: not allowed", wrapped.getMessage()); + Assertions.assertSame(cause, wrapped.getCause()); + } +} diff --git a/docs/iceberg-rest-service.md b/docs/iceberg-rest-service.md index 2d07fc4f789..c424e391273 100644 --- a/docs/iceberg-rest-service.md +++ b/docs/iceberg-rest-service.md @@ -529,16 +529,17 @@ Please set the `gravitino.iceberg-rest.warehouse` parameter to `oss://{bucket_na Supports using static GCS credential file or generating GCS token to access GCS data. -| Configuration item | Description | Default value | Required | -|----------------------------------|------------------------------------------------------------------------------------------------------------------------------|-----------------------------------------|----------| -| `gravitino.iceberg-rest.io-impl` | The IO implementation for `FileIO` in Iceberg. Set it to `org.apache.iceberg.gcp.gcs.GCSFileIO` to explicitly use GCSFileIO. | `org.apache.iceberg.io.ResolvingFileIO` | No | +| Configuration item | Description | Default value | Required | +|---------------------------------------------------|------------------------------------------------------------------------------------------------------------------------------|-----------------------------------------|----------| +| `gravitino.iceberg-rest.io-impl` | The IO implementation for `FileIO` in Iceberg. Set it to `org.apache.iceberg.gcp.gcs.GCSFileIO` to explicitly use GCSFileIO. | `org.apache.iceberg.io.ResolvingFileIO` | No | +| `gravitino.iceberg-rest.gcs-service-account-file` | Path of the GCS service account JSON file. Used for server-side FileIO and for `gcs-token` credential vending. | GCS Application default credential. | No | For other Iceberg GCS properties not managed by Gravitino like `gcs.project-id`, you could config it directly by `gravitino.iceberg-rest.gcs.project-id`. Refer to [GCS credentials](./security/credential-vending.md#gcs-credentials) for credential related configurations. :::note -Ensure that the credential file is accessible by the Gravitino server. For example, the server may be running on a GCE machine, or you may set the environment variable `export GOOGLE_APPLICATION_CREDENTIALS=/xx/application_default_credentials.json` even when `gcs-service-account-file` is already configured. +When `gcs-service-account-file` is set, Gravitino loads it at catalog initialization and injects Iceberg `gcs.oauth2.token` for FileIO. The IRC catalog cache evicts that catalog before the token expires so the next request recreates the catalog and mints a fresh token. If unset, use Application Default Credentials (for example GCE metadata or `GOOGLE_APPLICATION_CREDENTIALS`). ::: :::info diff --git a/docs/lakehouse-iceberg-catalog.md b/docs/lakehouse-iceberg-catalog.md index d4c0179243c..881d77699af 100644 --- a/docs/lakehouse-iceberg-catalog.md +++ b/docs/lakehouse-iceberg-catalog.md @@ -174,13 +174,14 @@ The Gravitino Iceberg aliyun bundle jar already includes the Iceberg aliyun nece Supports using google credential file to access GCS data. -| Configuration item | Description | Default value | Required | -|--------------------|------------------------------------------------------------------------------------------------------------------------------|-----------------------------------------|----------| -| `io-impl` | The IO implementation for `FileIO` in Iceberg. Set it to `org.apache.iceberg.gcp.gcs.GCSFileIO` to explicitly use GCSFileIO. | `org.apache.iceberg.io.ResolvingFileIO` | No | +| Configuration item | Description | Default value | Required | +|----------------------------|------------------------------------------------------------------------------------------------------------------------------|-----------------------------------------|----------| +| `io-impl` | The IO implementation for `FileIO` in Iceberg. Set it to `org.apache.iceberg.gcp.gcs.GCSFileIO` to explicitly use GCSFileIO. | `org.apache.iceberg.io.ResolvingFileIO` | No | +| `gcs-service-account-file` | Path of the GCS service account JSON file. Used for server-side FileIO and for `gcs-token` credential vending. | GCS Application default credential. | No | For other Iceberg GCS properties not managed by Gravitino like `gcs.project-id`, you could config it directly by `gravitino.bypass.gcs.project-id`. -Please make sure the credential file is accessible by Gravitino, like using `export GOOGLE_APPLICATION_CREDENTIALS=/xx/application_default_credentials.json` before Gravitino server is started. +When `gcs-service-account-file` is set, Gravitino loads it at catalog initialization and injects Iceberg `gcs.oauth2.token` for FileIO (Iceberg's `GCSFileIO` has no service-account-file property). If that property is unset, fall back to Application Default Credentials, for example `export GOOGLE_APPLICATION_CREDENTIALS=/xx/application_default_credentials.json`. :::info Please set `warehouse` to `gs://{bucket_name}/${prefix_name}`, and download [Gravitino Iceberg GCP bundle jar](https://mvnrepository.com/artifact/org.apache.gravitino/gravitino-iceberg-gcp-bundle) and place it to `catalogs/lakehouse-iceberg/libs/`. diff --git a/docs/security/credential-vending.md b/docs/security/credential-vending.md index 5c2d077f8a6..20b56966ad3 100644 --- a/docs/security/credential-vending.md +++ b/docs/security/credential-vending.md @@ -159,7 +159,7 @@ An GCS token is a token credential with scoped privileges, by leveraging GCS [Cr | `gcs-service-account-file` | `gravitino.iceberg-rest.gcs-service-account-file` | The location of GCS credential file. | GCS Application default credential. | No | :::note -For the Gravitino Iceberg REST server, ensure that the credential file is accessible by the server. For example, the server may be running on a GCE machine, or you may set the environment variable `export GOOGLE_APPLICATION_CREDENTIALS=/xx/application_default_credentials.json` even when `gcs-service-account-file` is already configured. +`gcs-service-account-file` is used both to vend downscoped tokens and to authenticate Iceberg `GCSFileIO` on the server (Gravitino injects `gcs.oauth2.token` at catalog load because Iceberg has no service-account-file property). Ensure the file is readable by the server process. If the property is unset, FileIO and token vending fall back to Application Default Credentials (for example GCE metadata or `GOOGLE_APPLICATION_CREDENTIALS`). ::: ## Custom Credentials diff --git a/iceberg/iceberg-common/build.gradle.kts b/iceberg/iceberg-common/build.gradle.kts index c6275006d4b..296db0d8e9d 100644 --- a/iceberg/iceberg-common/build.gradle.kts +++ b/iceberg/iceberg-common/build.gradle.kts @@ -59,6 +59,8 @@ dependencies { implementation(libs.iceberg.azure) implementation(libs.iceberg.hive.metastore) implementation(libs.iceberg.gcp) + // Load gcs-service-account-file into Iceberg GCSFileIO properties (gcs.oauth2.token). + implementation(libs.google.auth.http) // Upgrade to Hadoop 3.3+ for Iceberg 1.10 compatibility // Iceberg 1.10 requires Hadoop 3.3+ APIs like FileSystem.openFile() and FsTracer.get() implementation(libs.hadoop3.client.api) diff --git a/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/utils/IcebergCatalogUtil.java b/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/utils/IcebergCatalogUtil.java index f849b77ad56..03a494a4ecb 100644 --- a/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/utils/IcebergCatalogUtil.java +++ b/iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/utils/IcebergCatalogUtil.java @@ -21,15 +21,25 @@ import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.HADOOP_SECURITY_AUTHENTICATION; import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.HADOOP_SECURITY_AUTHORIZATION; +import com.google.auth.oauth2.AccessToken; +import com.google.auth.oauth2.GoogleCredentials; import com.google.common.annotations.VisibleForTesting; import com.google.common.collect.Maps; import java.io.IOException; +import java.io.InputStream; +import java.io.UncheckedIOException; +import java.nio.file.Files; +import java.nio.file.NoSuchFileException; +import java.nio.file.Path; +import java.nio.file.Paths; import java.sql.SQLException; import java.util.Collections; +import java.util.Date; import java.util.HashMap; import java.util.Locale; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; +import org.apache.commons.lang3.StringUtils; import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergCatalogBackend; import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants; import org.apache.gravitino.exceptions.ConnectionFailedException; @@ -37,6 +47,7 @@ import org.apache.gravitino.iceberg.common.IcebergConfig; import org.apache.gravitino.iceberg.common.authentication.AuthenticationConfig; import org.apache.gravitino.iceberg.common.rest.auth.UserPrincipalForwardingAuthManager; +import org.apache.gravitino.storage.GCSProperties; import org.apache.hadoop.hdfs.HdfsConfiguration; import org.apache.iceberg.CatalogProperties; import org.apache.iceberg.CatalogUtil; @@ -58,6 +69,18 @@ public class IcebergCatalogUtil { private static final Logger LOG = LoggerFactory.getLogger(IcebergCatalogUtil.class); + /** + * SQLSTATE {@code 28000}: MySQL error 1045 (Access denied), H2 wrong user/password, and + * PostgreSQL {@code invalid_authorization_specification} (for example unknown role). + */ + private static final String SQLSTATE_INVALID_AUTHORIZATION = "28000"; + + /** SQLSTATE {@code 28P01}: PostgreSQL {@code invalid_password}. */ + private static final String SQLSTATE_INVALID_PASSWORD = "28P01"; + + private static final String GCS_CLOUD_PLATFORM_SCOPE = + "https://www.googleapis.com/auth/cloud-platform"; + private static final ConcurrentHashMap MEMORY_CATALOGS = new ConcurrentHashMap<>(); @@ -170,9 +193,13 @@ private static JdbcCatalog loadJdbcCatalog(IcebergConfig icebergConfig) { try { jdbcCatalog.initialize(icebergCatalogName, properties); } catch (UncheckedSQLException e) { - if (e.getCause() instanceof SQLException - && e.getCause().getMessage().contains("Access denied")) { - throw new ConnectionFailedException(e, e.getMessage()); + Throwable cause = e.getCause(); + if (cause instanceof SQLException) { + String sqlState = ((SQLException) cause).getSQLState(); + if (SQLSTATE_INVALID_AUTHORIZATION.equals(sqlState) + || SQLSTATE_INVALID_PASSWORD.equals(sqlState)) { + throw new ConnectionFailedException(e, e.getMessage()); + } } throw e; } @@ -208,6 +235,84 @@ private static Catalog loadCustomCatalog(IcebergConfig icebergConfig) { @VisibleForTesting public static void applyDefaultResolvingFileIO(Map properties) { properties.putIfAbsent(IcebergConstants.IO_IMPL, ResolvingFileIO.class.getName()); + applyGcsServiceAccountCredentials(properties); + } + + /** + * When {@code gcs-service-account-file} is set, mint an OAuth2 access token and inject Iceberg + * {@code gcs.oauth2.token} / {@code gcs.oauth2.token-expires-at} so the built-in {@code + * GCSFileIO} can authenticate. Iceberg's FileIO does not understand Gravitino's + * service-account-file property; S3/OSS/ADLS instead map static keys directly via {@link + * org.apache.gravitino.catalog.lakehouse.iceberg.IcebergPropertiesUtils}. + * + *

Skips injection when {@code gcs.oauth2.token} is already present. Disables Iceberg's + * credentials-endpoint refresh because that path is for vended table credentials, not catalog + * bootstrap from a service account file. + * + * @param properties Iceberg catalog properties, mutated in place + */ + @VisibleForTesting + static void applyGcsServiceAccountCredentials(Map properties) { + String serviceAccountFile = properties.get(GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE); + if (StringUtils.isBlank(serviceAccountFile)) { + return; + } + if (StringUtils.isNotBlank(properties.get(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN))) { + return; + } + + AccessToken accessToken = loadAccessTokenFromFile(serviceAccountFile); + if (accessToken == null || StringUtils.isBlank(accessToken.getTokenValue())) { + throw new IllegalStateException( + "Failed to obtain GCS access token from service account file: " + serviceAccountFile); + } + + properties.put(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN, accessToken.getTokenValue()); + Date expirationTime = accessToken.getExpirationTime(); + if (expirationTime != null) { + properties.put( + IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN_EXPIRES_AT, + String.valueOf(expirationTime.toInstant().toEpochMilli())); + } + properties.put(IcebergConstants.ICEBERG_GCS_OAUTH2_REFRESH_CREDENTIALS_ENABLED, "false"); + LOG.info( + "Injected {} from {} for Iceberg GCSFileIO", + IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN, + GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE); + } + + /** + * Returns an {@link IcebergConfig} that includes a minted GCS OAuth2 token when {@code + * gcs-service-account-file} is configured. The returned config retains {@code + * gcs.oauth2.token-expires-at} so callers (for example the IRC catalog cache) can expire the + * catalog before the token becomes invalid. + * + * @param icebergConfig original catalog config + * @return the same instance when no token is injected; otherwise a new config with token fields + */ + public static IcebergConfig withGcsServiceAccountCredentials(IcebergConfig icebergConfig) { + Map properties = new HashMap<>(icebergConfig.getAllConfig()); + applyGcsServiceAccountCredentials(properties); + if (properties.equals(icebergConfig.getAllConfig())) { + return icebergConfig; + } + return new IcebergConfig(properties); + } + + private static AccessToken loadAccessTokenFromFile(String serviceAccountFile) { + Path credentialsFilePath = Paths.get(serviceAccountFile); + try (InputStream inputStream = Files.newInputStream(credentialsFilePath)) { + GoogleCredentials credentials = + GoogleCredentials.fromStream(inputStream).createScoped(GCS_CLOUD_PLATFORM_SCOPE); + credentials.refreshIfExpired(); + return credentials.getAccessToken(); + } catch (NoSuchFileException e) { + throw new UncheckedIOException( + "GCS service account file does not exist: " + serviceAccountFile, e); + } catch (IOException e) { + throw new UncheckedIOException( + "Failed to load GCS service account file: " + serviceAccountFile, e); + } } @VisibleForTesting diff --git a/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/utils/TestIcebergCatalogUtil.java b/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/utils/TestIcebergCatalogUtil.java index 89cf6f78971..337f8684dac 100644 --- a/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/utils/TestIcebergCatalogUtil.java +++ b/iceberg/iceberg-common/src/test/java/org/apache/gravitino/iceberg/common/utils/TestIcebergCatalogUtil.java @@ -19,12 +19,14 @@ package org.apache.gravitino.iceberg.common.utils; +import java.io.UncheckedIOException; import java.nio.file.Path; import java.util.HashMap; import java.util.Map; import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergCatalogBackend; import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants; import org.apache.gravitino.iceberg.common.IcebergConfig; +import org.apache.gravitino.storage.GCSProperties; import org.apache.iceberg.CatalogProperties; import org.apache.iceberg.Schema; import org.apache.iceberg.catalog.Catalog; @@ -312,6 +314,74 @@ void testApplyDefaultResolvingFileIODoesNotOverrideExplicitIOImpl() { "org.apache.iceberg.aws.s3.S3FileIO", properties.get(IcebergConstants.IO_IMPL)); } + @Test + void testApplyGcsServiceAccountCredentialsSkipsWhenTokenAlreadyPresent() { + Map properties = new HashMap<>(); + properties.put(GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE, "/tmp/gcs-key.json"); + properties.put(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN, "existing-token"); + + IcebergCatalogUtil.applyGcsServiceAccountCredentials(properties); + + Assertions.assertEquals( + "existing-token", properties.get(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN)); + Assertions.assertNull(properties.get(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN_EXPIRES_AT)); + } + + @Test + void testApplyGcsServiceAccountCredentialsNoOpWithoutServiceAccountFile() { + Map properties = new HashMap<>(); + properties.put(IcebergConstants.IO_IMPL, "org.apache.iceberg.gcp.gcs.GCSFileIO"); + + IcebergCatalogUtil.applyGcsServiceAccountCredentials(properties); + + Assertions.assertNull(properties.get(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN)); + } + + @Test + void testApplyGcsServiceAccountCredentialsFailsWhenFileMissing() { + Map properties = new HashMap<>(); + properties.put( + GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE, "/tmp/gravitino-missing-gcs-key.json"); + + UncheckedIOException thrown = + Assertions.assertThrows( + UncheckedIOException.class, + () -> IcebergCatalogUtil.applyGcsServiceAccountCredentials(properties)); + Assertions.assertTrue(thrown.getMessage().contains("does not exist")); + } + + @Test + void testWithGcsServiceAccountCredentialsReturnsSameConfigWhenNoServiceAccountFile() { + IcebergConfig config = new IcebergConfig(Map.of(IcebergConstants.CATALOG_BACKEND, "memory")); + Assertions.assertSame(config, IcebergCatalogUtil.withGcsServiceAccountCredentials(config)); + } + + @Test + void testWithGcsServiceAccountCredentialsReturnsSameConfigWhenTokenAlreadyPresent() { + Map properties = new HashMap<>(); + properties.put(GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE, "/tmp/gcs-key.json"); + properties.put(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN, "existing-token"); + IcebergConfig config = new IcebergConfig(properties); + + Assertions.assertSame(config, IcebergCatalogUtil.withGcsServiceAccountCredentials(config)); + } + + @Test + void testApplyDefaultResolvingFileIOInjectsGcsToken() { + Map properties = new HashMap<>(); + properties.put(IcebergConstants.WAREHOUSE, "gs://bucket/warehouse"); + properties.put(GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE, "/tmp/gcs-key.json"); + + // Pre-set token so applyDefaultResolvingFileIO skips loading a real service account file. + properties.put(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN, "pre-set"); + IcebergCatalogUtil.applyDefaultResolvingFileIO(properties); + + Assertions.assertEquals( + org.apache.iceberg.io.ResolvingFileIO.class.getName(), + properties.get(IcebergConstants.IO_IMPL)); + Assertions.assertEquals("pre-set", properties.get(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN)); + } + @Test void testApplyRestCatalogHttpTimeoutPropertiesUsesDefaults() { Map properties = new HashMap<>(); diff --git a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/FederatedCatalogWrapper.java b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/FederatedCatalogWrapper.java index cbe57249214..eb3892004a4 100644 --- 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 @@ -21,28 +21,20 @@ import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Preconditions; +import com.google.common.collect.ImmutableMap; 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.function.Function; 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; @@ -50,10 +42,6 @@ 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; @@ -64,7 +52,6 @@ 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; @@ -76,12 +63,12 @@ * {@link RESTCatalog}). * *

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. + * instanceof RESTCatalog} checks scattered across the base class. Table load, create and register + * use authenticated REST calls so {@code X-Iceberg-Access-Delegation} can be forwarded when the + * client requested credential vending. Update still uses the Iceberg Catalog API. This wrapper + * never injects Gravitino-generated credentials. * - *

Portions of the table create and update handling are derived from Apache Iceberg's {@code + *

Portions of the table update handling are derived from Apache Iceberg's {@code * org.apache.iceberg.rest.CatalogHandlers}: * https://github.com/apache/iceberg/blob/2abac79fcae94b5ad039bd09f7235be191b0761e/core/src/main/java/org/apache/iceberg/rest/CatalogHandlers.java */ @@ -89,6 +76,8 @@ public class FederatedCatalogWrapper extends CatalogWrapperForREST { private static final String FORMAT_VERSION = "format-version"; private static final Schema EMPTY_SCHEMA = new Schema(); + private static final String X_ICEBERG_ACCESS_DELEGATION = "X-Iceberg-Access-Delegation"; + private static final String VENDED_CREDENTIALS = "vended-credentials"; /** * Creates a federated wrapper. @@ -100,24 +89,63 @@ public FederatedCatalogWrapper(String catalogName, IcebergConfig config) { super(catalogName, config); } + /** + * Creates a table on the remote REST catalog. + * + *

Always uses a dedicated REST POST, including staged create, rather than Iceberg's Catalog + * API. When credential vending is requested the {@code X-Iceberg-Access-Delegation: + * vended-credentials} header is forwarded so the remote catalog returns {@code + * storage-credentials} inline. Upstream credential refresh endpoints are rewritten to this IRC + * catalog. + * + * @param namespace the namespace that will own the table. + * @param request the create-table request. + * @param requestCredential whether the client requested vended credentials. + * @return the create response, including rewritten remote credentials when requested. + */ @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); + return createTableViaREST(namespace, request, requestCredential); } + /** + * Loads a table from the remote REST catalog. + * + *

Always uses a dedicated REST GET rather than Iceberg's {@link RESTCatalog#loadTable}, which + * cannot send {@code X-Iceberg-Access-Delegation}. When credential vending is requested the + * header is forwarded so the remote catalog returns {@code storage-credentials} inline. Upstream + * credential refresh endpoints are rewritten to this IRC catalog. The {@code privilege} is + * ignored because the remote catalog decides what to vend. + * + * @param identifier the table identifier. + * @param requestCredential whether the client requested vended credentials. + * @param privilege ignored; the remote REST catalog vends its own credentials. + * @return the load-table response, including rewritten remote credentials when requested. + */ @Override public LoadTableResponse loadTable( TableIdentifier identifier, boolean requestCredential, CredentialPrivilege privilege) { - return loadTableInternal(identifier); + return loadTableViaREST(identifier, requestCredential); } + /** + * Registers a table on the remote REST catalog. + * + *

Always uses a dedicated REST POST rather than Iceberg's Catalog API. When credential vending + * is requested the {@code X-Iceberg-Access-Delegation: vended-credentials} header is forwarded so + * the remote catalog returns {@code storage-credentials} inline. Upstream credential refresh + * endpoints are rewritten to this IRC catalog. + * + * @param namespace the namespace that will own the table. + * @param request the register-table request. + * @param requestCredential whether the client requested vended credentials. + * @return the register response, including rewritten remote credentials when requested. + */ @Override public LoadTableResponse registerTable( Namespace namespace, RegisterTableRequest request, boolean requestCredential) { - return registerTableInternal(namespace, request); + return registerTableViaREST(namespace, request, requestCredential); } @Override @@ -159,6 +187,34 @@ private static LoadCredentialsResponse getRESTTableCredentials( String credentialsPath = ResourcePaths.forCatalogProperties(properties).table(identifier) + "/credentials"; + return callRemoteCatalog( + restCatalog, + String.format("loading credentials for table: %s", identifier), + client -> + client.get( + credentialsPath, + LoadCredentialsResponse.class, + Collections.emptyMap(), + ErrorHandlers.tableErrorHandler())); + } + + /** + * Runs an action against the remote REST catalog through a short-lived authenticated client. + * + *

Centralizes the auth manager, HTTP client and auth session lifecycle shared by the federated + * credential and load/create/register requests. Resources are closed in reverse order of + * creation, and a close failure on one does not prevent the others from being closed. + * + * @param restCatalog the underlying REST catalog whose properties supply the URI and auth config. + * @param description what the action is doing, used in close-failure log messages. + * @param action invoked with a client bound to an authenticated session. + * @param the action's result type. + * @return the action's result. + */ + private static T callRemoteCatalog( + RESTCatalog restCatalog, String description, Function action) { + Map properties = Maps.newHashMap(restCatalog.properties()); + AuthManager authManager = null; RESTClient client = null; AuthSession authSession = null; @@ -170,145 +226,146 @@ private static LoadCredentialsResponse getRESTTableCredentials( .withHeaders(RESTUtil.configHeaders(properties)) .build(); authSession = authManager.catalogSession(client, properties); - return client - .withAuthSession(authSession) - .get( - credentialsPath, - LoadCredentialsResponse.class, - Collections.emptyMap(), - ErrorHandlers.tableErrorHandler()); + return action.apply(client.withAuthSession(authSession)); } finally { - if (authSession != null) { - try { - authSession.close(); - } catch (Exception e) { - LOG.warn( - "Failed to close auth session when loading credentials for table: {}", identifier, e); - } - } + closeQuietly(authSession, "auth session", description); + closeQuietly(client, "REST client", description); + closeQuietly(authManager, "auth manager", description); + } + } - if (client != null) { - try { - client.close(); - } catch (Exception e) { - LOG.warn( - "Failed to close REST client when loading credentials for table: {}", identifier, e); - } - } + private static void closeQuietly( + AutoCloseable closeable, String resourceName, String description) { + if (closeable == null) { + return; + } - if (authManager != null) { - try { - authManager.close(); - } catch (Exception e) { - LOG.warn( - "Failed to close auth manager when loading credentials for table: {}", identifier, e); - } - } + try { + closeable.close(); + } catch (Exception e) { + LOG.warn("Failed to close {} when {}", resourceName, description, e); } } /** - * Federation-aware {@code createTable}: creates the table on the underlying (remote) catalog and - * extracts client-facing FileIO/credential properties from {@code table.io()}. + * Sends a {@code GET {table}} request to the remote REST catalog. + * + *

Follows the same HTTP client lifecycle as {@link #getRESTTableCredentials}. When credential + * vending is requested, the {@code X-Iceberg-Access-Delegation: vended-credentials} header is + * included so the remote catalog returns credentials inline in the load-table response. + * + * @param restCatalog the underlying REST catalog whose properties supply the URI and auth config. + * @param identifier the table to load. + * @param requestCredentialVending whether to include the access-delegation header. + * @return the load-table response from the remote catalog. */ - 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); - } + private static LoadTableResponse getRESTLoadTable( + RESTCatalog restCatalog, TableIdentifier identifier, boolean requestCredentialVending) { + Map properties = Maps.newHashMap(restCatalog.properties()); + String tablePath = ResourcePaths.forCatalogProperties(properties).table(identifier); + Map queryParams = ImmutableMap.of("snapshots", IcebergRESTUtils.SNAPSHOT_ALL); + + return callRemoteCatalog( + restCatalog, + String.format("loading table: %s", identifier), + client -> + client.get( + tablePath, + queryParams, + LoadTableResponse.class, + accessDelegationHeaders(requestCredentialVending), + ErrorHandlers.tableErrorHandler())); + } - throw new IllegalStateException("Cannot wrap catalog that does not produce BaseTable"); + /** + * Sends a {@code POST {namespace}/tables} request to the remote REST catalog. + * + * @param restCatalog the underlying REST catalog whose properties supply the URI and auth config. + * @param namespace the namespace that will own the table. + * @param request the create-table request (including staged create). + * @param requestCredentialVending whether to include the access-delegation header. + * @return the create response from the remote catalog. + */ + private static LoadTableResponse getRESTCreateTable( + RESTCatalog restCatalog, + Namespace namespace, + CreateTableRequest request, + boolean requestCredentialVending) { + Map properties = Maps.newHashMap(restCatalog.properties()); + String tablesPath = ResourcePaths.forCatalogProperties(properties).tables(namespace); + + return callRemoteCatalog( + restCatalog, + String.format("creating table: %s.%s", namespace, request.name()), + client -> + client.post( + tablesPath, + request, + LoadTableResponse.class, + accessDelegationHeaders(requestCredentialVending), + ErrorHandlers.createTableErrorHandler())); } - 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); - } + /** + * Sends a {@code POST {namespace}/register} request to the remote REST catalog. + * + * @param restCatalog the underlying REST catalog whose properties supply the URI and auth config. + * @param namespace the namespace that will own the table. + * @param request the register-table request. + * @param requestCredentialVending whether to include the access-delegation header. + * @return the register response from the remote catalog. + */ + private static LoadTableResponse getRESTRegisterTable( + RESTCatalog restCatalog, + Namespace namespace, + RegisterTableRequest request, + boolean requestCredentialVending) { + Map properties = Maps.newHashMap(restCatalog.properties()); + String registerPath = ResourcePaths.forCatalogProperties(properties).register(namespace); + + return callRemoteCatalog( + restCatalog, + String.format("registering table: %s.%s", namespace, request.name()), + client -> + client.post( + registerPath, + request, + LoadTableResponse.class, + accessDelegationHeaders(requestCredentialVending), + ErrorHandlers.tableErrorHandler())); + } - Map properties = Maps.newHashMap(); - properties.put("created-at", OffsetDateTime.now(ZoneOffset.UTC).toString()); - properties.putAll(request.properties()); - - Map 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(); - } + private static Map accessDelegationHeaders(boolean requestCredentialVending) { + return requestCredentialVending + ? ImmutableMap.of(X_ICEBERG_ACCESS_DELEGATION, VENDED_CREDENTIALS) + : Collections.emptyMap(); + } - Map tableProperties = retrieveFileIOProperties(table.io()); - Map 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 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 createTableViaREST( + Namespace namespace, CreateTableRequest request, boolean requestCredential) { + LoadTableResponse upstream = + getRESTCreateTable((RESTCatalog) getCatalog(), namespace, request, requestCredential); + return rewriteRemoteLoadTable(TableIdentifier.of(namespace, request.name()), upstream); } - /** - * 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()); + private LoadTableResponse loadTableViaREST( + TableIdentifier identifier, boolean requestCredential) { + LoadTableResponse upstream = + getRESTLoadTable((RESTCatalog) getCatalog(), identifier, requestCredential); + return rewriteRemoteLoadTable(identifier, upstream); + } - if (table instanceof BaseTable) { - return buildLoadTableResponseFromFileIo(ident, (BaseTable) table); - } + private LoadTableResponse registerTableViaREST( + Namespace namespace, RegisterTableRequest request, boolean requestCredential) { + LoadTableResponse upstream = + getRESTRegisterTable((RESTCatalog) getCatalog(), namespace, request, requestCredential); + return rewriteRemoteLoadTable(TableIdentifier.of(namespace, request.name()), upstream); + } - throw new IllegalStateException("Cannot wrap catalog that does not produce BaseTable"); + private LoadTableResponse rewriteRemoteLoadTable( + TableIdentifier identifier, LoadTableResponse upstream) { + return IcebergRESTUtils.rewriteLoadTableCredentials( + catalogCredentialManager.catalogName(), identifier, upstream); } /** @@ -361,52 +418,6 @@ private LoadTableResponse tableUpdateInternal(TableIdentifier ident, UpdateTable } } - /** - * 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 properties = retrieveFileIOProperties(table.io()); - Map 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() @@ -501,10 +512,4 @@ static boolean shouldApplyMetadataUpdateAfterBuilder(MetadataUpdate update) { return true; } - - private static Map 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 3a3c518f304..a960f175e0d 100644 --- a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergCatalogWrapperManager.java +++ b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergCatalogWrapperManager.java @@ -20,6 +20,7 @@ import com.github.benmanes.caffeine.cache.Cache; import com.github.benmanes.caffeine.cache.Caffeine; +import com.github.benmanes.caffeine.cache.Expiry; import com.github.benmanes.caffeine.cache.Scheduler; import com.google.common.annotations.VisibleForTesting; import com.google.common.util.concurrent.ThreadFactoryBuilder; @@ -28,6 +29,7 @@ import java.util.Optional; import java.util.concurrent.ScheduledThreadPoolExecutor; import java.util.concurrent.TimeUnit; +import org.apache.commons.lang3.StringUtils; import org.apache.gravitino.GravitinoEnv; import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergCatalogBackend; import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants; @@ -37,6 +39,7 @@ import org.apache.gravitino.iceberg.common.authentication.SupportsKerberos; import org.apache.gravitino.iceberg.common.ops.IcebergCatalogWrapper; import org.apache.gravitino.iceberg.common.ops.KerberosAwareIcebergCatalogProxy; +import org.apache.gravitino.iceberg.common.utils.IcebergCatalogUtil; import org.apache.gravitino.iceberg.service.authorization.IcebergRESTServerContext; import org.apache.gravitino.iceberg.service.provider.DynamicIcebergConfigProvider; import org.apache.gravitino.iceberg.service.provider.IcebergConfigProvider; @@ -47,6 +50,12 @@ public class IcebergCatalogWrapperManager implements AutoCloseable { public static final Logger LOG = LoggerFactory.getLogger(IcebergCatalogWrapperManager.class); + /** + * Evict a cached catalog this long before its minted GCS OAuth2 token expires, so the next + * request recreates the catalog and refreshes the token. + */ + @VisibleForTesting static final long GCS_TOKEN_REFRESH_BUFFER_MS = TimeUnit.MINUTES.toMillis(5); + private final Cache catalogWrapperCache; private final IcebergConfigProvider configProvider; @@ -57,17 +66,20 @@ public IcebergCatalogWrapperManager( boolean auxMode, String metalakeName) { this.configProvider = configProvider; + long accessEvictionNanos = + TimeUnit.MILLISECONDS.toNanos( + new IcebergConfig(properties) + .get(IcebergConfig.ICEBERG_REST_CATALOG_CACHE_EVICTION_INTERVAL)); this.catalogWrapperCache = Caffeine.newBuilder() - .expireAfterAccess( - (new IcebergConfig(properties)) - .get(IcebergConfig.ICEBERG_REST_CATALOG_CACHE_EVICTION_INTERVAL), - TimeUnit.MILLISECONDS) + .expireAfter(new CatalogWrapperExpiry(accessEvictionNanos)) .removalListener( - (k, v, c) -> { - String catalogName = (String) k; - LOG.info("Remove IcebergCatalogWrapper cache {}.", catalogName); - closeIcebergCatalogWrapper((IcebergCatalogWrapper) v); + (catalogName, catalogWrapper, cause) -> { + LOG.debug( + "Removing IcebergCatalogWrapper from cache: catalog={}, cause={}", + catalogName, + cause); + closeIcebergCatalogWrapper(catalogWrapper); }) .scheduler( Scheduler.forScheduledExecutorService( @@ -137,22 +149,26 @@ private CatalogWrapperForREST createCatalogWrapper(String catalogName) { @VisibleForTesting protected CatalogWrapperForREST createCatalogWrapper( String catalogName, IcebergConfig icebergConfig) { + // Mint GCS OAuth2 tokens into the config before constructing the wrapper so the IRC catalog + // cache can expire the entry before gcs.oauth2.token-expires-at. + IcebergConfig enrichedConfig = + IcebergCatalogUtil.withGcsServiceAccountCredentials(icebergConfig); // When the backend is a federated Iceberg REST catalog, use FederatedCatalogWrapper so // federation-aware behavior (FileIO property extraction, remote credential vending, remote // /v1/config defaults) is applied through polymorphic dispatch rather than scattered // instanceof checks. All other backends use the base CatalogWrapperForREST. IcebergCatalogBackend backend = IcebergCatalogBackend.valueOf( - icebergConfig.get(IcebergConfig.CATALOG_BACKEND).toUpperCase(Locale.ROOT)); + enrichedConfig.get(IcebergConfig.CATALOG_BACKEND).toUpperCase(Locale.ROOT)); CatalogWrapperForREST rest = backend == IcebergCatalogBackend.REST - ? new FederatedCatalogWrapper(catalogName, icebergConfig) - : new CatalogWrapperForREST(catalogName, icebergConfig); + ? new FederatedCatalogWrapper(catalogName, enrichedConfig) + : new CatalogWrapperForREST(catalogName, enrichedConfig); AuthenticationConfig authenticationConfig = - new AuthenticationConfig(icebergConfig.getAllConfig()); + new AuthenticationConfig(enrichedConfig.getAllConfig()); if (authenticationConfig.isKerberosAuth() && rest.getCatalog() instanceof SupportsKerberos) { return (CatalogWrapperForREST) - new KerberosAwareIcebergCatalogProxy(rest).getProxy(catalogName, icebergConfig); + new KerberosAwareIcebergCatalogProxy(rest).getProxy(catalogName, enrichedConfig); } return rest; @@ -166,8 +182,73 @@ private void closeIcebergCatalogWrapper(IcebergCatalogWrapper catalogWrapper) { } } + /** + * Computes how long a catalog wrapper may stay in the IRC cache. + * + *

Uses the configured access-based eviction interval, capped by the time until a minted GCS + * OAuth2 token should be refreshed ({@code gcs.oauth2.token-expires-at} minus {@link + * #GCS_TOKEN_REFRESH_BUFFER_MS}). When no token expiry is present, returns {@code + * accessEvictionNanos}. + * + * @param config catalog config that may contain {@code gcs.oauth2.token-expires-at} + * @param accessEvictionNanos default expire-after-access duration in nanoseconds + * @param nowEpochMillis current wall-clock time + * @return cache duration in nanoseconds; {@code 0} means expire immediately + */ + @VisibleForTesting + static long computeCacheDurationNanos( + IcebergConfig config, long accessEvictionNanos, long nowEpochMillis) { + String expiresAt = + config.getAllConfig().get(IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN_EXPIRES_AT); + if (StringUtils.isBlank(expiresAt)) { + return accessEvictionNanos; + } + + long expiresAtMs; + try { + expiresAtMs = Long.parseLong(expiresAt); + } catch (NumberFormatException e) { + LOG.warn("Invalid {}: {}", IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN_EXPIRES_AT, expiresAt); + return accessEvictionNanos; + } + + long remainingMs = expiresAtMs - GCS_TOKEN_REFRESH_BUFFER_MS - nowEpochMillis; + if (remainingMs <= 0) { + return 0L; + } + return Math.min(accessEvictionNanos, TimeUnit.MILLISECONDS.toNanos(remainingMs)); + } + @Override public void close() throws Exception { catalogWrapperCache.invalidateAll(); } + + private static final class CatalogWrapperExpiry implements Expiry { + + private final long accessEvictionNanos; + + CatalogWrapperExpiry(long accessEvictionNanos) { + this.accessEvictionNanos = accessEvictionNanos; + } + + @Override + public long expireAfterCreate(String key, CatalogWrapperForREST value, long currentTime) { + return computeCacheDurationNanos( + value.getIcebergConfig(), accessEvictionNanos, System.currentTimeMillis()); + } + + @Override + public long expireAfterUpdate( + String key, CatalogWrapperForREST value, long currentTime, long currentDuration) { + return expireAfterCreate(key, value, currentTime); + } + + @Override + public long expireAfterRead( + String key, CatalogWrapperForREST value, long currentTime, long currentDuration) { + // Preserve expire-after-access, but never extend past the GCS token refresh deadline. + return expireAfterCreate(key, value, currentTime); + } + } } diff --git a/iceberg/iceberg-rest-server/src/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 9a4d74091dc..bedd609d371 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 @@ -21,6 +21,7 @@ import com.fasterxml.jackson.databind.ObjectMapper; import com.google.common.base.Preconditions; import com.google.common.collect.ImmutableMap; +import com.google.common.collect.ImmutableSet; import java.io.IOException; import java.nio.charset.StandardCharsets; import java.security.MessageDigest; @@ -36,6 +37,7 @@ import java.util.List; import java.util.Map; import java.util.Optional; +import java.util.Set; import java.util.stream.Stream; import javax.servlet.http.HttpServletRequest; import javax.ws.rs.core.EntityTag; @@ -77,6 +79,18 @@ public class IcebergRESTUtils { public static final String SNAPSHOT_REFS = "refs"; + /** + * Iceberg refresh-endpoint keys that may appear in {@link LoadTableResponse#config()}. Kept in + * sync with {@link CredentialPropertyUtils#buildRefreshProps}; they are not retained by {@link + * CredentialPropertyUtils#filterCredentialProperties}, so top-level config must drop them + * explicitly before IRC-local endpoints are re-applied. + */ + private static final Set REFRESH_CREDENTIALS_ENDPOINT_KEYS = + ImmutableSet.of( + "client.refresh-credentials-endpoint", + "gcs.oauth2.refresh-credentials-endpoint", + "adls.refresh-credentials-endpoint"); + /** Snapshot modes for the Iceberg loadTable endpoint. */ public enum SnapshotMode { ALL(SNAPSHOT_ALL), @@ -232,6 +246,49 @@ private static org.apache.iceberg.rest.credentials.Credential rewriteCredential( return toRESTCredential(prefix, ImmutableMap.copyOf(filteredConfig)); } + /** + * Rewrites credentials in a federated {@link LoadTableResponse} so their {@code + * refresh-credentials-endpoint} entries, including any flattened into {@code config}, point at + * this IRC instance instead of the upstream catalog. + * + *

Upstream refresh endpoints are removed from top-level {@code config} before IRC-local ones + * are re-applied, matching {@link #rewriteCredential}. Without that step a refresh URL that lived + * only in {@code config} (with tokens only in {@code storage-credentials}) would leak. + * + * @param catalogName IRC catalog name used to build refresh paths + * @param tableIdentifier table receiving the credentials + * @param upstream the load-table response returned by the upstream REST catalog + * @return a load-table response with IRC-local refresh endpoints + */ + public static LoadTableResponse rewriteLoadTableCredentials( + String catalogName, TableIdentifier tableIdentifier, LoadTableResponse upstream) { + Map config = new HashMap<>(); + if (upstream.config() != null) { + config.putAll(upstream.config()); + } + // filterCredentialProperties drops refresh endpoints from its return value but putAll does not + // remove keys already copied from upstream.config(). Drop them first so a refresh URL that + // lived only in config (tokens only in storage-credentials) cannot leak to clients. + config.keySet().removeAll(REFRESH_CREDENTIALS_ENDPOINT_KEYS); + Map filteredCredentialProperties = + CredentialPropertyUtils.filterCredentialProperties(config); + config.putAll(filteredCredentialProperties); + config.putAll(buildRefreshProps(catalogName, tableIdentifier, filteredCredentialProperties)); + + LoadTableResponse.Builder builder = + LoadTableResponse.builder() + .withTableMetadata(upstream.tableMetadata()) + .addAllConfig(config); + if (upstream.credentials() != null) { + for (org.apache.iceberg.rest.credentials.Credential credential : upstream.credentials()) { + builder.addCredential( + rewriteCredential( + catalogName, tableIdentifier, credential.prefix(), credential.config())); + } + } + return builder.build(); + } + public static 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/main/java/org/apache/gravitino/iceberg/service/rest/IcebergTableOperations.java b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/rest/IcebergTableOperations.java index 24d567e6c79..1782fe8ec92 100644 --- a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/rest/IcebergTableOperations.java +++ b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/rest/IcebergTableOperations.java @@ -567,10 +567,14 @@ static LoadTableResponse filterSnapshotsByRefs(LoadTableResponse loadTableRespon } TableMetadata filteredMetadata = TableMetadata.buildFrom(metadata).suppressHistoricalSnapshots().build(); - return LoadTableResponse.builder() - .withTableMetadata(filteredMetadata) - .addAllConfig(loadTableResponse.config()) - .build(); + LoadTableResponse.Builder builder = + LoadTableResponse.builder() + .withTableMetadata(filteredMetadata) + .addAllConfig(loadTableResponse.config()); + if (loadTableResponse.credentials() != null) { + builder.addAllCredentials(loadTableResponse.credentials()); + } + return builder.build(); } private static Response buildResponseWithETag(LoadTableResponse loadTableResponse) { 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 e3721705856..612a6185244 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 @@ -20,14 +20,12 @@ package org.apache.gravitino.iceberg.service; import static org.mockito.Mockito.any; -import static org.mockito.Mockito.anyMap; import static org.mockito.Mockito.anyString; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; -import static org.mockito.Mockito.withSettings; import com.google.common.collect.ImmutableMap; import com.sun.net.httpserver.HttpServer; @@ -47,17 +45,15 @@ import org.apache.gravitino.credential.CredentialPrivilege; import org.apache.gravitino.iceberg.common.IcebergConfig; import org.apache.gravitino.iceberg.service.extension.DummyCredentialProvider; -import org.apache.iceberg.BaseTable; import org.apache.iceberg.BaseTransaction; import org.apache.iceberg.CatalogProperties; import org.apache.iceberg.MetadataUpdate; import org.apache.iceberg.PartitionSpec; import org.apache.iceberg.Schema; import org.apache.iceberg.SortOrder; -import org.apache.iceberg.Table; import org.apache.iceberg.TableMetadata; +import org.apache.iceberg.TableMetadataParser; 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; @@ -67,10 +63,7 @@ import org.apache.iceberg.exceptions.NoSuchTableException; import org.apache.iceberg.exceptions.NotAuthorizedException; import org.apache.iceberg.exceptions.ServiceFailureException; -import org.apache.iceberg.io.FileIO; import org.apache.iceberg.io.ResolvingFileIO; -import org.apache.iceberg.io.StorageCredential; -import org.apache.iceberg.io.SupportsStorageCredentials; import org.apache.iceberg.rest.RESTCatalog; import org.apache.iceberg.rest.auth.AuthProperties; import org.apache.iceberg.rest.credentials.Credential; @@ -80,6 +73,7 @@ import org.apache.iceberg.rest.requests.UpdateTableRequest; import org.apache.iceberg.rest.responses.LoadCredentialsResponse; import org.apache.iceberg.rest.responses.LoadTableResponse; +import org.apache.iceberg.rest.responses.LoadTableResponseParser; import org.apache.iceberg.types.Types; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; @@ -431,70 +425,93 @@ void testValidateCredentialLocation() { } @Test - void testLoadTableRefreshEndpoint() { + void testLoadTableRefreshEndpoint() throws Exception { TableIdentifier ident = TableIdentifier.of(Namespace.of("db"), "tbl"); - RESTCatalog catalog = mock(RESTCatalog.class); - BaseTable baseTable = mock(BaseTable.class); - TableOperations ops = mock(TableOperations.class); - FileIO fileIO = mock(FileIO.class); TableMetadata metadata = - TableMetadata.newTableMetadata( - new Schema(Types.NestedField.required(1, "id", Types.IntegerType.get())), - PartitionSpec.unpartitioned(), - SortOrder.unsorted(), - "s3://bucket/db/tbl", - Collections.emptyMap()); - - when(catalog.loadTable(ident)).thenReturn(baseTable); - when(baseTable.operations()).thenReturn(ops); - when(ops.current()).thenReturn(metadata); - when(baseTable.io()).thenReturn(fileIO); - when(fileIO.properties()) - .thenReturn( - ImmutableMap.of( - "s3.session-token", - "token", - "s3.session-token-expires-at-ms", - "123", - "client.refresh-credentials-endpoint", - "v1/upstream/namespaces/db/tables/tbl/credentials")); + TableMetadataParser.fromJson( + "s3://bucket/db/tbl/metadata/v1.metadata.json", + TableMetadataParser.toJson( + TableMetadata.newTableMetadata( + new Schema(Types.NestedField.required(1, "id", Types.IntegerType.get())), + PartitionSpec.unpartitioned(), + SortOrder.unsorted(), + "s3://bucket/db/tbl", + Collections.emptyMap()))); + LoadTableResponse upstreamResponse = + LoadTableResponse.builder() + .withTableMetadata(metadata) + .addAllConfig( + ImmutableMap.of( + "s3.session-token", + "token", + "s3.session-token-expires-at-ms", + "123", + "client.refresh-credentials-endpoint", + "v1/upstream/namespaces/db/tables/tbl/credentials")) + .build(); + String upstreamJson = LoadTableResponseParser.toJson(upstreamResponse); - IcebergConfig config = - new IcebergConfig( - ImmutableMap.of( - IcebergConstants.CATALOG_BACKEND, - "memory", - IcebergConstants.WAREHOUSE, - "/tmp/warehouse")); - CatalogWrapperForREST wrapper = new StaticCatalogWrapperForREST("irc1", config, catalog); + HttpServer server = HttpServer.create(new InetSocketAddress(0), 0); + server.createContext( + "/", + exchange -> { + byte[] body = upstreamJson.getBytes(StandardCharsets.UTF_8); + exchange.getResponseHeaders().add("Content-Type", "application/json"); + exchange.sendResponseHeaders(200, body.length); + try (OutputStream os = exchange.getResponseBody()) { + os.write(body); + } + }); + server.start(); + try { + String uri = "http://127.0.0.1:" + server.getAddress().getPort(); + RESTCatalog catalog = mock(RESTCatalog.class); + when(catalog.name()).thenReturn("upstream"); + when(catalog.properties()) + .thenReturn( + ImmutableMap.of( + CatalogProperties.URI, + uri, + AuthProperties.AUTH_TYPE, + AuthProperties.AUTH_TYPE_NONE, + "prefix", + "upstream")); + + IcebergConfig config = + new IcebergConfig( + ImmutableMap.of( + IcebergConstants.CATALOG_BACKEND, + "memory", + IcebergConstants.WAREHOUSE, + "/tmp/warehouse")); + CatalogWrapperForREST wrapper = new StaticCatalogWrapperForREST("irc1", config, catalog); - LoadTableResponse response = wrapper.loadTable(ident, false, CredentialPrivilege.READ); + LoadTableResponse response = wrapper.loadTable(ident, true, CredentialPrivilege.READ); - Assertions.assertEquals( - "v1/irc1/namespaces/db/tables/tbl/credentials", - response.config().get("client.refresh-credentials-endpoint")); - Assertions.assertEquals("token", response.config().get("s3.session-token")); + Assertions.assertEquals( + "v1/irc1/namespaces/db/tables/tbl/credentials", + response.config().get("client.refresh-credentials-endpoint")); + Assertions.assertEquals("token", response.config().get("s3.session-token")); + } finally { + server.stop(0); + } } @Test - void testLoadTableStorageCreds() { + void testLoadTableStorageCreds() throws Exception { TableIdentifier ident = TableIdentifier.of(Namespace.of("db"), "tbl"); - RESTCatalog catalog = mock(RESTCatalog.class); - BaseTable baseTable = mock(BaseTable.class); - TableOperations ops = mock(TableOperations.class); - FileIO fileIO = - mock(FileIO.class, withSettings().extraInterfaces(SupportsStorageCredentials.class)); - SupportsStorageCredentials storageCredentialsFileIO = (SupportsStorageCredentials) fileIO; TableMetadata metadata = - TableMetadata.newTableMetadata( - new Schema(Types.NestedField.required(1, "id", Types.IntegerType.get())), - PartitionSpec.unpartitioned(), - SortOrder.unsorted(), - "s3://bucket/db/tbl", - Collections.emptyMap()); - - StorageCredential upstreamCredential = - StorageCredential.create( + TableMetadataParser.fromJson( + "s3://bucket/db/tbl/metadata/v1.metadata.json", + TableMetadataParser.toJson( + TableMetadata.newTableMetadata( + new Schema(Types.NestedField.required(1, "id", Types.IntegerType.get())), + PartitionSpec.unpartitioned(), + SortOrder.unsorted(), + "s3://bucket/db/tbl", + Collections.emptyMap()))); + Credential upstreamCredential = + IcebergRESTUtils.toRESTCredential( "s3://bucket/db/tbl/", ImmutableMap.of( "s3.access-key-id", @@ -507,33 +524,61 @@ void testLoadTableStorageCreds() { "123", "client.refresh-credentials-endpoint", "v1/upstream/namespaces/db/tables/tbl/credentials")); + LoadTableResponse upstreamResponse = + LoadTableResponse.builder() + .withTableMetadata(metadata) + .addCredential(upstreamCredential) + .build(); + String upstreamJson = LoadTableResponseParser.toJson(upstreamResponse); - when(catalog.loadTable(ident)).thenReturn(baseTable); - when(baseTable.operations()).thenReturn(ops); - when(ops.current()).thenReturn(metadata); - when(baseTable.io()).thenReturn(fileIO); - when(fileIO.properties()).thenReturn(Collections.emptyMap()); - when(storageCredentialsFileIO.credentials()).thenReturn(List.of(upstreamCredential)); + HttpServer server = HttpServer.create(new InetSocketAddress(0), 0); + server.createContext( + "/", + exchange -> { + byte[] body = upstreamJson.getBytes(StandardCharsets.UTF_8); + exchange.getResponseHeaders().add("Content-Type", "application/json"); + exchange.sendResponseHeaders(200, body.length); + try (OutputStream os = exchange.getResponseBody()) { + os.write(body); + } + }); + server.start(); + try { + String uri = "http://127.0.0.1:" + server.getAddress().getPort(); + RESTCatalog catalog = mock(RESTCatalog.class); + when(catalog.name()).thenReturn("upstream"); + when(catalog.properties()) + .thenReturn( + ImmutableMap.of( + CatalogProperties.URI, + uri, + AuthProperties.AUTH_TYPE, + AuthProperties.AUTH_TYPE_NONE, + "prefix", + "upstream")); - IcebergConfig config = - new IcebergConfig( - ImmutableMap.of( - IcebergConstants.CATALOG_BACKEND, - "memory", - IcebergConstants.WAREHOUSE, - "/tmp/warehouse")); - CatalogWrapperForREST wrapper = new StaticCatalogWrapperForREST("irc1", config, catalog); + IcebergConfig config = + new IcebergConfig( + ImmutableMap.of( + IcebergConstants.CATALOG_BACKEND, + "memory", + IcebergConstants.WAREHOUSE, + "/tmp/warehouse")); + CatalogWrapperForREST wrapper = new StaticCatalogWrapperForREST("irc1", config, catalog); - LoadTableResponse response = wrapper.loadTable(ident, false, CredentialPrivilege.READ); + LoadTableResponse response = wrapper.loadTable(ident, true, CredentialPrivilege.READ); - Assertions.assertEquals(1, response.credentials().size()); - Credential credential = response.credentials().get(0); - Assertions.assertEquals("s3://bucket/db/tbl/", credential.prefix()); - Assertions.assertEquals("upstream-token", credential.config().get("s3.session-token")); - Assertions.assertEquals( - "v1/irc1/namespaces/db/tables/tbl/credentials", - credential.config().get("client.refresh-credentials-endpoint")); - Assertions.assertFalse(response.config().containsKey("client.refresh-credentials-endpoint")); + Assertions.assertEquals(1, response.credentials().size()); + Credential credential = response.credentials().get(0); + Assertions.assertEquals("s3://bucket/db/tbl/", credential.prefix()); + Assertions.assertEquals("upstream-token", credential.config().get("s3.session-token")); + Assertions.assertEquals( + "v1/irc1/namespaces/db/tables/tbl/credentials", + credential.config().get("client.refresh-credentials-endpoint")); + Assertions.assertFalse(response.config().containsKey("client.refresh-credentials-endpoint")); + } finally { + server.stop(0); + } } @Test @@ -621,47 +666,6 @@ void testCatalogClientConfigRejectsBadDataAccess() { () -> 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, - "memory", - IcebergConstants.WAREHOUSE, - "/tmp/warehouse")); - CatalogWrapperForREST wrapper = new StaticCatalogWrapperForREST("test", 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 void testWrapperLazyLoadsCatalog() { IcebergConfig config = @@ -680,102 +684,6 @@ void testWrapperLazyLoadsCatalog() { } } - @Test - void testStageCreateWithLocationIncludesFileIo() throws Exception { - RESTCatalog catalog = mock(RESTCatalog.class); - Catalog.TableBuilder tableBuilder = mock(Catalog.TableBuilder.class); - Transaction transaction = mock(Transaction.class); - Table table = mock(Table.class); - FileIO fileIO = mock(FileIO.class); - when(catalog.buildTable(any(TableIdentifier.class), any())).thenReturn(tableBuilder); - when(tableBuilder.withPartitionSpec(any())).thenReturn(tableBuilder); - when(tableBuilder.withSortOrder(any())).thenReturn(tableBuilder); - when(tableBuilder.withProperties(anyMap())).thenReturn(tableBuilder); - when(tableBuilder.withLocation("s3://bucket/warehouse/table")).thenReturn(tableBuilder); - when(tableBuilder.createTransaction()).thenReturn(transaction); - when(transaction.table()).thenReturn(table); - when(table.io()).thenReturn(fileIO); - when(table.location()).thenReturn("s3://bucket/warehouse/table"); - 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, - "memory", - IcebergConstants.WAREHOUSE, - "/tmp/warehouse")); - CatalogWrapperForREST wrapper = new StaticCatalogWrapperForREST("test", config, catalog); - - Schema schema = new Schema(Types.NestedField.required(1, "id", Types.IntegerType.get())); - CreateTableRequest request = - CreateTableRequest.builder() - .withName("tbl") - .withSchema(schema) - .withLocation("s3://bucket/warehouse/table") - .stageCreate() - .build(); - - LoadTableResponse response = wrapper.createTable(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)); - verify(tableBuilder).withLocation("s3://bucket/warehouse/table"); - } - - @Test - void testStageCreateNullLocationSkipsWithLocation() { - RESTCatalog catalog = mock(RESTCatalog.class); - Catalog.TableBuilder tableBuilder = mock(Catalog.TableBuilder.class); - Transaction transaction = mock(Transaction.class); - Table table = mock(Table.class); - FileIO fileIO = mock(FileIO.class); - when(catalog.buildTable(any(TableIdentifier.class), any())).thenReturn(tableBuilder); - when(tableBuilder.withPartitionSpec(any())).thenReturn(tableBuilder); - when(tableBuilder.withSortOrder(any())).thenReturn(tableBuilder); - when(tableBuilder.withProperties(anyMap())).thenReturn(tableBuilder); - when(tableBuilder.createTransaction()).thenReturn(transaction); - when(transaction.table()).thenReturn(table); - when(table.io()).thenReturn(fileIO); - when(table.location()).thenReturn("s3://bucket/warehouse/default-location"); - 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, - "memory", - IcebergConstants.WAREHOUSE, - "/tmp/warehouse")); - CatalogWrapperForREST wrapper = new StaticCatalogWrapperForREST("test", config, catalog); - - Schema schema = new Schema(Types.NestedField.required(1, "id", Types.IntegerType.get())); - CreateTableRequest request = - CreateTableRequest.builder().withName("tbl").withSchema(schema).stageCreate().build(); - - LoadTableResponse response = wrapper.createTable(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)); - verify(tableBuilder, never()).withLocation(any()); - } - @Test void testStagedCreateRejectsExtraRequirements() { RESTCatalog catalog = mock(RESTCatalog.class); @@ -1027,4 +935,668 @@ public Catalog getCatalog() { return catalog; } } + + @Test + void testFederatedLoadTableDelegatesToRemote() throws Exception { + TableIdentifier table = TableIdentifier.of(Namespace.of("db"), "tbl"); + String expectedPath = "/v1/upstream/namespaces/db/tables/tbl"; + + TableMetadata metadata = + TableMetadataParser.fromJson( + "s3://bucket/db/tbl/metadata/v1.metadata.json", + TableMetadataParser.toJson( + TableMetadata.newTableMetadata( + new Schema(Types.NestedField.required(1, "id", Types.IntegerType.get())), + PartitionSpec.unpartitioned(), + SortOrder.unsorted(), + "s3://bucket/db/tbl", + Collections.emptyMap()))); + org.apache.iceberg.rest.credentials.Credential cred = + IcebergRESTUtils.toRESTCredential( + "s3://bucket/db/tbl/", + ImmutableMap.of( + "s3.access-key-id", "upstream-key", + "s3.secret-access-key", "upstream-secret", + "s3.session-token", "upstream-token", + "client.refresh-credentials-endpoint", + "v1/upstream/namespaces/db/tables/tbl/credentials")); + LoadTableResponse upstreamResponse = + LoadTableResponse.builder() + .withTableMetadata(metadata) + .addAllConfig(ImmutableMap.of("io-impl", "org.apache.iceberg.aws.s3.S3FileIO")) + .addCredential(cred) + .build(); + String upstreamJson = LoadTableResponseParser.toJson(upstreamResponse); + + AtomicReference requestPath = new AtomicReference<>(); + AtomicReference requestMethod = new AtomicReference<>(); + AtomicReference requestQuery = new AtomicReference<>(); + AtomicReference accessDelegationHeader = new AtomicReference<>(); + HttpServer server = HttpServer.create(new InetSocketAddress(0), 0); + server.createContext( + "/", + exchange -> { + requestPath.set(exchange.getRequestURI().getPath()); + requestMethod.set(exchange.getRequestMethod()); + requestQuery.set(exchange.getRequestURI().getQuery()); + accessDelegationHeader.set( + exchange.getRequestHeaders().getFirst("X-Iceberg-Access-Delegation")); + byte[] body = upstreamJson.getBytes(StandardCharsets.UTF_8); + exchange.getResponseHeaders().add("Content-Type", "application/json"); + exchange.sendResponseHeaders(200, body.length); + try (OutputStream os = exchange.getResponseBody()) { + os.write(body); + } + }); + server.start(); + try { + String uri = "http://127.0.0.1:" + server.getAddress().getPort(); + RESTCatalog restCatalog = mock(RESTCatalog.class); + when(restCatalog.name()).thenReturn("upstream"); + when(restCatalog.properties()) + .thenReturn( + ImmutableMap.of( + CatalogProperties.URI, + uri, + AuthProperties.AUTH_TYPE, + AuthProperties.AUTH_TYPE_NONE, + "prefix", + "upstream")); + + IcebergConfig config = + new IcebergConfig( + ImmutableMap.of( + IcebergConstants.CATALOG_BACKEND, + "memory", + IcebergConstants.WAREHOUSE, + "/tmp/warehouse")); + CatalogWrapperForREST wrapper = new StaticCatalogWrapperForREST("local", config, restCatalog); + + LoadTableResponse response = wrapper.loadTable(table, true, CredentialPrivilege.READ); + + Assertions.assertEquals(expectedPath, requestPath.get()); + Assertions.assertEquals("GET", requestMethod.get()); + Assertions.assertEquals("snapshots=all", requestQuery.get()); + Assertions.assertEquals("vended-credentials", accessDelegationHeader.get()); + verify(restCatalog, never()).loadTable(table); + Assertions.assertEquals(1, response.credentials().size()); + Credential credential = response.credentials().get(0); + Assertions.assertEquals("s3://bucket/db/tbl/", credential.prefix()); + Assertions.assertEquals("upstream-key", credential.config().get("s3.access-key-id")); + Assertions.assertEquals("upstream-token", credential.config().get("s3.session-token")); + Assertions.assertEquals( + "v1/local/namespaces/db/tables/tbl/credentials", + credential.config().get("client.refresh-credentials-endpoint")); + Assertions.assertEquals( + "org.apache.iceberg.aws.s3.S3FileIO", response.config().get("io-impl")); + } finally { + server.stop(0); + } + } + + @Test + void testFederatedLoadTableNoCredentials() throws Exception { + TableIdentifier table = TableIdentifier.of(Namespace.of("db"), "tbl"); + TableMetadata metadata = + TableMetadataParser.fromJson( + "s3://bucket/db/tbl/metadata/v1.metadata.json", + TableMetadataParser.toJson( + TableMetadata.newTableMetadata( + new Schema(Types.NestedField.required(1, "id", Types.IntegerType.get())), + PartitionSpec.unpartitioned(), + SortOrder.unsorted(), + "s3://bucket/db/tbl", + Collections.emptyMap()))); + LoadTableResponse upstreamResponse = + LoadTableResponse.builder() + .withTableMetadata(metadata) + .addAllConfig(ImmutableMap.of("io-impl", "org.apache.iceberg.aws.s3.S3FileIO")) + .build(); + String upstreamJson = LoadTableResponseParser.toJson(upstreamResponse); + + AtomicReference accessDelegationHeader = new AtomicReference<>(); + HttpServer server = HttpServer.create(new InetSocketAddress(0), 0); + server.createContext( + "/", + exchange -> { + accessDelegationHeader.set( + exchange.getRequestHeaders().getFirst("X-Iceberg-Access-Delegation")); + byte[] body = upstreamJson.getBytes(StandardCharsets.UTF_8); + exchange.getResponseHeaders().add("Content-Type", "application/json"); + exchange.sendResponseHeaders(200, body.length); + try (OutputStream os = exchange.getResponseBody()) { + os.write(body); + } + }); + server.start(); + try { + String uri = "http://127.0.0.1:" + server.getAddress().getPort(); + RESTCatalog restCatalog = mock(RESTCatalog.class); + when(restCatalog.name()).thenReturn("upstream"); + when(restCatalog.properties()) + .thenReturn( + ImmutableMap.of( + CatalogProperties.URI, + uri, + AuthProperties.AUTH_TYPE, + AuthProperties.AUTH_TYPE_NONE, + "prefix", + "upstream")); + + IcebergConfig config = + new IcebergConfig( + ImmutableMap.of( + IcebergConstants.CATALOG_BACKEND, + "memory", + IcebergConstants.WAREHOUSE, + "/tmp/warehouse")); + CatalogWrapperForREST wrapper = new StaticCatalogWrapperForREST("local", config, restCatalog); + + LoadTableResponse response = wrapper.loadTable(table, false, CredentialPrivilege.READ); + + Assertions.assertNull( + accessDelegationHeader.get(), + "X-Iceberg-Access-Delegation header should not be sent without credential vending"); + verify(restCatalog, never()).loadTable(table); + Assertions.assertTrue( + response.credentials() == null || response.credentials().isEmpty(), + "Non-vended request should not return remote storage-credentials"); + Assertions.assertEquals( + "org.apache.iceberg.aws.s3.S3FileIO", response.config().get("io-impl")); + } finally { + server.stop(0); + } + } + + @Test + void testFederatedCreateTableWithCredentials() throws Exception { + Namespace namespace = Namespace.of("db"); + TableMetadata metadata = + TableMetadataParser.fromJson( + "s3://bucket/db/tbl/metadata/v1.metadata.json", + TableMetadataParser.toJson( + TableMetadata.newTableMetadata( + new Schema(Types.NestedField.required(1, "id", Types.IntegerType.get())), + PartitionSpec.unpartitioned(), + SortOrder.unsorted(), + "s3://bucket/db/tbl", + Collections.emptyMap()))); + Credential cred = + IcebergRESTUtils.toRESTCredential( + "s3://bucket/db/tbl/", + ImmutableMap.of( + "s3.session-token", + "upstream-token", + "client.refresh-credentials-endpoint", + "v1/upstream/namespaces/db/tables/tbl/credentials")); + LoadTableResponse upstreamResponse = + LoadTableResponse.builder().withTableMetadata(metadata).addCredential(cred).build(); + String upstreamJson = LoadTableResponseParser.toJson(upstreamResponse); + + AtomicReference requestPath = new AtomicReference<>(); + AtomicReference requestMethod = new AtomicReference<>(); + AtomicReference accessDelegationHeader = new AtomicReference<>(); + HttpServer server = HttpServer.create(new InetSocketAddress(0), 0); + server.createContext( + "/", + exchange -> { + requestPath.set(exchange.getRequestURI().getPath()); + requestMethod.set(exchange.getRequestMethod()); + accessDelegationHeader.set( + exchange.getRequestHeaders().getFirst("X-Iceberg-Access-Delegation")); + byte[] body = upstreamJson.getBytes(StandardCharsets.UTF_8); + exchange.getResponseHeaders().add("Content-Type", "application/json"); + exchange.sendResponseHeaders(200, body.length); + try (OutputStream os = exchange.getResponseBody()) { + os.write(body); + } + }); + server.start(); + try { + String uri = "http://127.0.0.1:" + server.getAddress().getPort(); + RESTCatalog restCatalog = mock(RESTCatalog.class); + when(restCatalog.name()).thenReturn("upstream"); + when(restCatalog.properties()) + .thenReturn( + ImmutableMap.of( + CatalogProperties.URI, + uri, + AuthProperties.AUTH_TYPE, + AuthProperties.AUTH_TYPE_NONE, + "prefix", + "upstream")); + + IcebergConfig config = + new IcebergConfig( + ImmutableMap.of( + IcebergConstants.CATALOG_BACKEND, + "memory", + IcebergConstants.WAREHOUSE, + "/tmp/warehouse")); + CatalogWrapperForREST wrapper = new StaticCatalogWrapperForREST("local", config, restCatalog); + CreateTableRequest request = + CreateTableRequest.builder() + .withName("tbl") + .withSchema(new Schema(Types.NestedField.required(1, "id", Types.IntegerType.get()))) + .build(); + + LoadTableResponse response = wrapper.createTable(namespace, request, true); + + Assertions.assertEquals("/v1/upstream/namespaces/db/tables", requestPath.get()); + Assertions.assertEquals("POST", requestMethod.get()); + Assertions.assertEquals("vended-credentials", accessDelegationHeader.get()); + Assertions.assertEquals(1, response.credentials().size()); + Assertions.assertEquals( + "v1/local/namespaces/db/tables/tbl/credentials", + response.credentials().get(0).config().get("client.refresh-credentials-endpoint")); + } finally { + server.stop(0); + } + } + + @Test + void testFederatedCreateTableNoCredentials() throws Exception { + Namespace namespace = Namespace.of("db"); + TableMetadata metadata = + TableMetadataParser.fromJson( + "s3://bucket/db/tbl/metadata/v1.metadata.json", + TableMetadataParser.toJson( + TableMetadata.newTableMetadata( + new Schema(Types.NestedField.required(1, "id", Types.IntegerType.get())), + PartitionSpec.unpartitioned(), + SortOrder.unsorted(), + "s3://bucket/db/tbl", + Collections.emptyMap()))); + LoadTableResponse upstreamResponse = + LoadTableResponse.builder() + .withTableMetadata(metadata) + .addAllConfig(ImmutableMap.of("io-impl", "org.apache.iceberg.aws.s3.S3FileIO")) + .build(); + String upstreamJson = LoadTableResponseParser.toJson(upstreamResponse); + + AtomicReference requestPath = new AtomicReference<>(); + AtomicReference requestMethod = new AtomicReference<>(); + AtomicReference accessDelegationHeader = new AtomicReference<>(); + HttpServer server = HttpServer.create(new InetSocketAddress(0), 0); + server.createContext( + "/", + exchange -> { + requestPath.set(exchange.getRequestURI().getPath()); + requestMethod.set(exchange.getRequestMethod()); + accessDelegationHeader.set( + exchange.getRequestHeaders().getFirst("X-Iceberg-Access-Delegation")); + byte[] body = upstreamJson.getBytes(StandardCharsets.UTF_8); + exchange.getResponseHeaders().add("Content-Type", "application/json"); + exchange.sendResponseHeaders(200, body.length); + try (OutputStream os = exchange.getResponseBody()) { + os.write(body); + } + }); + server.start(); + try { + String uri = "http://127.0.0.1:" + server.getAddress().getPort(); + RESTCatalog restCatalog = mock(RESTCatalog.class); + when(restCatalog.name()).thenReturn("upstream"); + when(restCatalog.properties()) + .thenReturn( + ImmutableMap.of( + CatalogProperties.URI, + uri, + AuthProperties.AUTH_TYPE, + AuthProperties.AUTH_TYPE_NONE, + "prefix", + "upstream")); + + IcebergConfig config = + new IcebergConfig( + ImmutableMap.of( + IcebergConstants.CATALOG_BACKEND, + "memory", + IcebergConstants.WAREHOUSE, + "/tmp/warehouse")); + CatalogWrapperForREST wrapper = new StaticCatalogWrapperForREST("local", config, restCatalog); + CreateTableRequest request = + CreateTableRequest.builder() + .withName("tbl") + .withSchema(new Schema(Types.NestedField.required(1, "id", Types.IntegerType.get()))) + .build(); + + LoadTableResponse response = wrapper.createTable(namespace, request, false); + + Assertions.assertEquals("/v1/upstream/namespaces/db/tables", requestPath.get()); + Assertions.assertEquals("POST", requestMethod.get()); + Assertions.assertNull( + accessDelegationHeader.get(), + "X-Iceberg-Access-Delegation header should not be sent without credential vending"); + Assertions.assertTrue( + response.credentials() == null || response.credentials().isEmpty(), + "Non-vended request should not return remote storage-credentials"); + Assertions.assertEquals( + "org.apache.iceberg.aws.s3.S3FileIO", response.config().get("io-impl")); + } finally { + server.stop(0); + } + } + + @Test + void testFederatedCreateTableForwardsStageCreate() throws Exception { + Namespace namespace = Namespace.of("db"); + TableMetadata metadata = + TableMetadataParser.fromJson( + "s3://bucket/db/tbl/metadata/v1.metadata.json", + TableMetadataParser.toJson( + TableMetadata.newTableMetadata( + new Schema(Types.NestedField.required(1, "id", Types.IntegerType.get())), + PartitionSpec.unpartitioned(), + SortOrder.unsorted(), + "s3://bucket/db/tbl", + Collections.emptyMap()))); + LoadTableResponse upstreamResponse = + LoadTableResponse.builder().withTableMetadata(metadata).build(); + String upstreamJson = LoadTableResponseParser.toJson(upstreamResponse); + + AtomicReference requestBody = new AtomicReference<>(); + HttpServer server = HttpServer.create(new InetSocketAddress(0), 0); + server.createContext( + "/", + exchange -> { + requestBody.set( + new String(exchange.getRequestBody().readAllBytes(), StandardCharsets.UTF_8)); + byte[] body = upstreamJson.getBytes(StandardCharsets.UTF_8); + exchange.getResponseHeaders().add("Content-Type", "application/json"); + exchange.sendResponseHeaders(200, body.length); + try (OutputStream os = exchange.getResponseBody()) { + os.write(body); + } + }); + server.start(); + try { + String uri = "http://127.0.0.1:" + server.getAddress().getPort(); + RESTCatalog restCatalog = mock(RESTCatalog.class); + when(restCatalog.name()).thenReturn("upstream"); + when(restCatalog.properties()) + .thenReturn( + ImmutableMap.of( + CatalogProperties.URI, + uri, + AuthProperties.AUTH_TYPE, + AuthProperties.AUTH_TYPE_NONE, + "prefix", + "upstream")); + + IcebergConfig config = + new IcebergConfig( + ImmutableMap.of( + IcebergConstants.CATALOG_BACKEND, + "memory", + IcebergConstants.WAREHOUSE, + "/tmp/warehouse")); + CatalogWrapperForREST wrapper = new StaticCatalogWrapperForREST("local", config, restCatalog); + Schema schema = new Schema(Types.NestedField.required(1, "id", Types.IntegerType.get())); + CreateTableRequest request = + CreateTableRequest.builder() + .withName("tbl") + .withSchema(schema) + .withLocation("s3://bucket/warehouse/table") + .stageCreate() + .build(); + + wrapper.createTable(namespace, request, false); + + CreateTableRequest forwarded = + IcebergObjectMapper.getInstance().readValue(requestBody.get(), CreateTableRequest.class); + Assertions.assertTrue(forwarded.stageCreate()); + Assertions.assertEquals("s3://bucket/warehouse/table", forwarded.location()); + } finally { + server.stop(0); + } + } + + @Test + void testFederatedRegisterTableWithCredentials() throws Exception { + Namespace namespace = Namespace.of("db"); + TableMetadata metadata = + TableMetadataParser.fromJson( + "s3://bucket/db/tbl/metadata/v1.metadata.json", + TableMetadataParser.toJson( + TableMetadata.newTableMetadata( + new Schema(Types.NestedField.required(1, "id", Types.IntegerType.get())), + PartitionSpec.unpartitioned(), + SortOrder.unsorted(), + "s3://bucket/db/tbl", + Collections.emptyMap()))); + Credential cred = + IcebergRESTUtils.toRESTCredential( + "s3://bucket/db/tbl/", + ImmutableMap.of( + "s3.session-token", + "upstream-token", + "client.refresh-credentials-endpoint", + "v1/upstream/namespaces/db/tables/tbl/credentials")); + LoadTableResponse upstreamResponse = + LoadTableResponse.builder().withTableMetadata(metadata).addCredential(cred).build(); + String upstreamJson = LoadTableResponseParser.toJson(upstreamResponse); + + AtomicReference requestPath = new AtomicReference<>(); + AtomicReference requestMethod = new AtomicReference<>(); + AtomicReference accessDelegationHeader = new AtomicReference<>(); + HttpServer server = HttpServer.create(new InetSocketAddress(0), 0); + server.createContext( + "/", + exchange -> { + requestPath.set(exchange.getRequestURI().getPath()); + requestMethod.set(exchange.getRequestMethod()); + accessDelegationHeader.set( + exchange.getRequestHeaders().getFirst("X-Iceberg-Access-Delegation")); + byte[] body = upstreamJson.getBytes(StandardCharsets.UTF_8); + exchange.getResponseHeaders().add("Content-Type", "application/json"); + exchange.sendResponseHeaders(200, body.length); + try (OutputStream os = exchange.getResponseBody()) { + os.write(body); + } + }); + server.start(); + try { + String uri = "http://127.0.0.1:" + server.getAddress().getPort(); + RESTCatalog restCatalog = mock(RESTCatalog.class); + when(restCatalog.name()).thenReturn("upstream"); + when(restCatalog.properties()) + .thenReturn( + ImmutableMap.of( + CatalogProperties.URI, + uri, + AuthProperties.AUTH_TYPE, + AuthProperties.AUTH_TYPE_NONE, + "prefix", + "upstream")); + + IcebergConfig config = + new IcebergConfig( + ImmutableMap.of( + IcebergConstants.CATALOG_BACKEND, + "memory", + IcebergConstants.WAREHOUSE, + "/tmp/warehouse")); + CatalogWrapperForREST wrapper = new StaticCatalogWrapperForREST("local", config, restCatalog); + RegisterTableRequest request = + ImmutableRegisterTableRequest.builder() + .name("tbl") + .metadataLocation("s3://bucket/db/tbl/metadata/v1.metadata.json") + .build(); + + LoadTableResponse response = wrapper.registerTable(namespace, request, true); + + Assertions.assertEquals("/v1/upstream/namespaces/db/register", requestPath.get()); + Assertions.assertEquals("POST", requestMethod.get()); + Assertions.assertEquals("vended-credentials", accessDelegationHeader.get()); + Assertions.assertEquals(1, response.credentials().size()); + Assertions.assertEquals( + "v1/local/namespaces/db/tables/tbl/credentials", + response.credentials().get(0).config().get("client.refresh-credentials-endpoint")); + } finally { + server.stop(0); + } + } + + @Test + void testFederatedRegisterTableIncludesRemoteConfig() throws Exception { + Namespace namespace = Namespace.of("db"); + TableMetadata metadata = + TableMetadataParser.fromJson( + "s3://bucket/db/tbl/metadata/v1.metadata.json", + TableMetadataParser.toJson( + TableMetadata.newTableMetadata( + new Schema(Types.NestedField.required(1, "id", Types.IntegerType.get())), + PartitionSpec.unpartitioned(), + SortOrder.unsorted(), + "s3://bucket/db/tbl", + Collections.emptyMap()))); + LoadTableResponse upstreamResponse = + LoadTableResponse.builder() + .withTableMetadata(metadata) + .addAllConfig( + ImmutableMap.of( + IcebergConstants.IO_IMPL, + "org.apache.iceberg.aws.s3.S3FileIO", + IcebergConstants.ICEBERG_S3_ENDPOINT, + "http://localhost:9000")) + .build(); + String upstreamJson = LoadTableResponseParser.toJson(upstreamResponse); + + AtomicReference requestPath = new AtomicReference<>(); + AtomicReference requestMethod = new AtomicReference<>(); + AtomicReference accessDelegationHeader = new AtomicReference<>(); + HttpServer server = HttpServer.create(new InetSocketAddress(0), 0); + server.createContext( + "/", + exchange -> { + requestPath.set(exchange.getRequestURI().getPath()); + requestMethod.set(exchange.getRequestMethod()); + accessDelegationHeader.set( + exchange.getRequestHeaders().getFirst("X-Iceberg-Access-Delegation")); + byte[] body = upstreamJson.getBytes(StandardCharsets.UTF_8); + exchange.getResponseHeaders().add("Content-Type", "application/json"); + exchange.sendResponseHeaders(200, body.length); + try (OutputStream os = exchange.getResponseBody()) { + os.write(body); + } + }); + server.start(); + try { + String uri = "http://127.0.0.1:" + server.getAddress().getPort(); + RESTCatalog catalog = mock(RESTCatalog.class); + when(catalog.name()).thenReturn("upstream"); + when(catalog.properties()) + .thenReturn( + ImmutableMap.of( + CatalogProperties.URI, + uri, + AuthProperties.AUTH_TYPE, + AuthProperties.AUTH_TYPE_NONE, + "prefix", + "upstream")); + + IcebergConfig config = + new IcebergConfig( + ImmutableMap.of( + IcebergConstants.CATALOG_BACKEND, + "memory", + IcebergConstants.WAREHOUSE, + "/tmp/warehouse")); + CatalogWrapperForREST wrapper = new StaticCatalogWrapperForREST("test", config, catalog); + + RegisterTableRequest request = + ImmutableRegisterTableRequest.builder() + .name("tbl") + .metadataLocation("s3://bucket/warehouse/tbl/metadata/v1.metadata.json") + .build(); + + LoadTableResponse response = wrapper.registerTable(namespace, request, false); + + Assertions.assertEquals("/v1/upstream/namespaces/db/register", requestPath.get()); + Assertions.assertEquals("POST", requestMethod.get()); + Assertions.assertNull(accessDelegationHeader.get()); + 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)); + } finally { + server.stop(0); + } + } + + @Test + void testFederatedRegisterTableOverwrite() throws Exception { + Namespace namespace = Namespace.of("db"); + TableMetadata metadata = + TableMetadataParser.fromJson( + "s3://bucket/db/tbl/metadata/v1.metadata.json", + TableMetadataParser.toJson( + TableMetadata.newTableMetadata( + new Schema(Types.NestedField.required(1, "id", Types.IntegerType.get())), + PartitionSpec.unpartitioned(), + SortOrder.unsorted(), + "s3://bucket/db/tbl", + Collections.emptyMap()))); + LoadTableResponse upstreamResponse = + LoadTableResponse.builder().withTableMetadata(metadata).build(); + String upstreamJson = LoadTableResponseParser.toJson(upstreamResponse); + + AtomicReference requestBody = new AtomicReference<>(); + HttpServer server = HttpServer.create(new InetSocketAddress(0), 0); + server.createContext( + "/", + exchange -> { + requestBody.set( + new String(exchange.getRequestBody().readAllBytes(), StandardCharsets.UTF_8)); + byte[] body = upstreamJson.getBytes(StandardCharsets.UTF_8); + exchange.getResponseHeaders().add("Content-Type", "application/json"); + exchange.sendResponseHeaders(200, body.length); + try (OutputStream os = exchange.getResponseBody()) { + os.write(body); + } + }); + server.start(); + try { + String uri = "http://127.0.0.1:" + server.getAddress().getPort(); + RESTCatalog catalog = mock(RESTCatalog.class); + when(catalog.name()).thenReturn("upstream"); + when(catalog.properties()) + .thenReturn( + ImmutableMap.of( + CatalogProperties.URI, + uri, + AuthProperties.AUTH_TYPE, + AuthProperties.AUTH_TYPE_NONE, + "prefix", + "upstream")); + + IcebergConfig config = + new IcebergConfig( + ImmutableMap.of( + IcebergConstants.CATALOG_BACKEND, + "memory", + IcebergConstants.WAREHOUSE, + "/tmp/warehouse")); + CatalogWrapperForREST wrapper = new StaticCatalogWrapperForREST("test", config, catalog); + + RegisterTableRequest request = + ImmutableRegisterTableRequest.builder() + .name("tbl") + .metadataLocation("s3://bucket/warehouse/tbl/metadata/v2.metadata.json") + .overwrite(true) + .build(); + + wrapper.registerTable(namespace, request, false); + + RegisterTableRequest forwarded = + IcebergObjectMapper.getInstance() + .readValue(requestBody.get(), RegisterTableRequest.class); + Assertions.assertTrue(forwarded.overwrite()); + Assertions.assertEquals( + "s3://bucket/warehouse/tbl/metadata/v2.metadata.json", forwarded.metadataLocation()); + } finally { + server.stop(0); + } + } } diff --git a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergCatalogWrapperManagerForREST.java b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergCatalogWrapperManagerForREST.java index 45eb62f69ee..3a5e8925e12 100644 --- a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergCatalogWrapperManagerForREST.java +++ b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestIcebergCatalogWrapperManagerForREST.java @@ -22,6 +22,7 @@ import com.google.common.collect.Maps; import java.util.Map; import java.util.Optional; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; import java.util.function.Consumer; import org.apache.commons.lang3.StringUtils; @@ -189,6 +190,70 @@ public void testDefaultCatalogAliasInvalidatedWhenCatalogRemoved() throws Except } } + @Test + public void testComputeCacheDurationNanosWithoutTokenExpiryUsesAccessEviction() { + IcebergConfig config = + new IcebergConfig(ImmutableMap.of(IcebergConstants.CATALOG_BACKEND, "memory")); + long accessEvictionNanos = TimeUnit.HOURS.toNanos(1); + Assertions.assertEquals( + accessEvictionNanos, + IcebergCatalogWrapperManager.computeCacheDurationNanos( + config, accessEvictionNanos, System.currentTimeMillis())); + } + + @Test + public void testComputeCacheDurationNanosCapsByGcsTokenExpiry() { + long now = 1_700_000_000_000L; + long expiresAt = now + TimeUnit.HOURS.toMillis(1); // token valid for 1h + IcebergConfig config = + new IcebergConfig( + ImmutableMap.of( + IcebergConstants.CATALOG_BACKEND, + "memory", + IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN_EXPIRES_AT, + String.valueOf(expiresAt))); + long accessEvictionNanos = TimeUnit.HOURS.toNanos(2); + long expected = + TimeUnit.MILLISECONDS.toNanos( + TimeUnit.HOURS.toMillis(1) - IcebergCatalogWrapperManager.GCS_TOKEN_REFRESH_BUFFER_MS); + Assertions.assertEquals( + expected, + IcebergCatalogWrapperManager.computeCacheDurationNanos(config, accessEvictionNanos, now)); + } + + @Test + public void testComputeCacheDurationNanosExpiresImmediatelyWhenPastRefreshDeadline() { + long now = 1_700_000_000_000L; + long expiresAt = now + TimeUnit.MINUTES.toMillis(2); // within 5-minute buffer + IcebergConfig config = + new IcebergConfig( + ImmutableMap.of( + IcebergConstants.CATALOG_BACKEND, + "memory", + IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN_EXPIRES_AT, + String.valueOf(expiresAt))); + Assertions.assertEquals( + 0L, + IcebergCatalogWrapperManager.computeCacheDurationNanos( + config, TimeUnit.HOURS.toNanos(1), now)); + } + + @Test + public void testComputeCacheDurationNanosIgnoresInvalidExpiresAt() { + IcebergConfig config = + new IcebergConfig( + ImmutableMap.of( + IcebergConstants.CATALOG_BACKEND, + "memory", + IcebergConstants.ICEBERG_GCS_OAUTH2_TOKEN_EXPIRES_AT, + "not-a-number")); + long accessEvictionNanos = TimeUnit.MINUTES.toNanos(30); + Assertions.assertEquals( + accessEvictionNanos, + IcebergCatalogWrapperManager.computeCacheDurationNanos( + config, accessEvictionNanos, System.currentTimeMillis())); + } + private static IcebergCatalogWrapperManager newManager() { Map config = Maps.newHashMap(); IcebergConfigProvider configProvider = IcebergConfigProviderFactory.create(config); 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 2ad1f85c3df..e3e4f3225d6 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 @@ -24,6 +24,7 @@ import static org.mockito.Mockito.withSettings; import com.google.common.collect.ImmutableMap; +import java.util.Collections; import java.util.List; import java.util.Map; import org.apache.gravitino.NameIdentifier; @@ -36,7 +37,9 @@ import org.apache.gravitino.credential.S3TokenCredential; import org.apache.gravitino.iceberg.service.authorization.IcebergRESTServerContext; import org.apache.gravitino.iceberg.service.provider.IcebergConfigProvider; +import org.apache.iceberg.PartitionSpec; import org.apache.iceberg.Schema; +import org.apache.iceberg.SortOrder; import org.apache.iceberg.TableMetadata; import org.apache.iceberg.catalog.Namespace; import org.apache.iceberg.catalog.TableIdentifier; @@ -47,6 +50,7 @@ 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.rest.responses.LoadTableResponse; import org.apache.iceberg.types.Types.IntegerType; import org.apache.iceberg.types.Types.NestedField; import org.apache.iceberg.types.Types.StringType; @@ -350,4 +354,99 @@ void testRewriteTableCredentials() { "v1/irc1/namespaces/db/tables/tbl/credentials", credential.config().get("client.refresh-credentials-endpoint")); } + + @Test + void testRewriteLoadTableCredentials() { + TableIdentifier table = TableIdentifier.of(Namespace.of("db"), "tbl"); + TableMetadata metadata = + TableMetadata.newTableMetadata( + new Schema(NestedField.required(1, "id", IntegerType.get())), + PartitionSpec.unpartitioned(), + SortOrder.unsorted(), + "s3://bucket/db/tbl", + Collections.emptyMap()); + LoadTableResponse upstream = + LoadTableResponse.builder() + .withTableMetadata(metadata) + .addAllConfig( + ImmutableMap.of( + "io-impl", + "org.apache.iceberg.aws.s3.S3FileIO", + "s3.session-token", + "upstream-token", + "client.refresh-credentials-endpoint", + "v1/upstream/namespaces/db/tables/tbl/credentials")) + .addCredential( + IcebergRESTUtils.toRESTCredential( + "s3://bucket/db/tbl/", + ImmutableMap.of( + "s3.session-token", + "upstream-token", + "client.refresh-credentials-endpoint", + "v1/upstream/namespaces/db/tables/tbl/credentials"))) + .build(); + + LoadTableResponse rewritten = + IcebergRESTUtils.rewriteLoadTableCredentials("irc1", table, upstream); + + Assertions.assertEquals("s3://bucket/db/tbl", rewritten.tableMetadata().location()); + Assertions.assertEquals( + "org.apache.iceberg.aws.s3.S3FileIO", rewritten.config().get("io-impl")); + Assertions.assertEquals("upstream-token", rewritten.config().get("s3.session-token")); + Assertions.assertEquals( + "v1/irc1/namespaces/db/tables/tbl/credentials", + rewritten.config().get("client.refresh-credentials-endpoint")); + 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")); + Assertions.assertEquals( + "v1/irc1/namespaces/db/tables/tbl/credentials", + credential.config().get("client.refresh-credentials-endpoint")); + } + + @Test + void testRewriteLoadTableCredentialsDropsConfigRefreshWithoutTokens() { + TableIdentifier table = TableIdentifier.of(Namespace.of("db"), "tbl"); + TableMetadata metadata = + TableMetadata.newTableMetadata( + new Schema(NestedField.required(1, "id", IntegerType.get())), + PartitionSpec.unpartitioned(), + SortOrder.unsorted(), + "s3://bucket/db/tbl", + Collections.emptyMap()); + // Upstream put the refresh URL in top-level config while the session token lives only in + // storage-credentials. The config refresh key must be dropped, not left pointing upstream. + LoadTableResponse upstream = + LoadTableResponse.builder() + .withTableMetadata(metadata) + .addAllConfig( + ImmutableMap.of( + "io-impl", + "org.apache.iceberg.aws.s3.S3FileIO", + "client.refresh-credentials-endpoint", + "v1/upstream/namespaces/db/tables/tbl/credentials")) + .addCredential( + IcebergRESTUtils.toRESTCredential( + "s3://bucket/db/tbl/", + ImmutableMap.of( + "s3.session-token", + "upstream-token", + "client.refresh-credentials-endpoint", + "v1/upstream/namespaces/db/tables/tbl/credentials"))) + .build(); + + LoadTableResponse rewritten = + IcebergRESTUtils.rewriteLoadTableCredentials("irc1", table, upstream); + + Assertions.assertEquals( + "org.apache.iceberg.aws.s3.S3FileIO", rewritten.config().get("io-impl")); + Assertions.assertFalse( + rewritten.config().containsKey("client.refresh-credentials-endpoint"), + "Top-level config must not keep an upstream refresh endpoint when tokens are absent"); + Assertions.assertEquals(1, rewritten.credentials().size()); + Assertions.assertEquals( + "v1/irc1/namespaces/db/tables/tbl/credentials", + rewritten.credentials().get(0).config().get("client.refresh-credentials-endpoint")); + } } diff --git a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/TestIcebergTableOperations.java b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/TestIcebergTableOperations.java index 14d2782b4f7..8cb0557837f 100644 --- a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/TestIcebergTableOperations.java +++ b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/TestIcebergTableOperations.java @@ -542,6 +542,11 @@ private Response doLoadTable(Namespace ns, String name) { } private Response doLoadTableWithSnapshots(Namespace ns, String name, String snapshots) { + return doLoadTableWithSnapshots(ns, name, snapshots, false); + } + + private Response doLoadTableWithSnapshots( + Namespace ns, String name, String snapshots, boolean credentialVending) { String path = IcebergRestTestUtil.NAMESPACE_PATH + "/" @@ -549,7 +554,12 @@ private Response doLoadTableWithSnapshots(Namespace ns, String name, String snap + "/tables/" + name; Map queryParams = ImmutableMap.of("snapshots", snapshots); - return getIcebergClientBuilder(path, Optional.of(queryParams)).get(); + Invocation.Builder builder = getIcebergClientBuilder(path, Optional.of(queryParams)); + if (credentialVending) { + builder = + builder.header(IcebergTableOperations.X_ICEBERG_ACCESS_DELEGATION, "vended-credentials"); + } + return builder.get(); } private Response doPlanTableScan(Namespace ns, String tableName, PlanTableScanRequest request) { @@ -1083,6 +1093,73 @@ void testLoadTableSnapshotsRefsFiltering(Namespace namespace) { "Refs should be preserved in filtered response"); } + @ParameterizedTest + @MethodSource("org.apache.gravitino.iceberg.service.rest.IcebergRestTestUtil#testNamespaces") + void testLoadTableSnapshotsRefsKeepsCredentials(Namespace namespace) { + verifyCreateNamespaceSucc(namespace); + String tableName = "snapshots_refs_creds"; + CreateTableRequest createTableRequest = + CreateTableRequest.builder() + .withName(tableName) + .withSchema(tableSchema) + .withLocation("s3://bucket/" + tableName) + .setProperties( + ImmutableMap.of( + CatalogWrapperForTest.GENERATE_PLAN_TASKS_DATA_PROP, Boolean.TRUE.toString())) + .build(); + Response createResponse = + getTableClientBuilder(namespace, Optional.empty()) + .header(IcebergTableOperations.X_ICEBERG_ACCESS_DELEGATION, "vended-credentials") + .post(Entity.entity(createTableRequest, MediaType.APPLICATION_JSON_TYPE)); + Assertions.assertEquals(Status.OK.getStatusCode(), createResponse.getStatus()); + + Response refsResponse = doLoadTableWithSnapshots(namespace, tableName, "refs", true); + Assertions.assertEquals(Status.OK.getStatusCode(), refsResponse.getStatus()); + LoadTableResponse refsTableResponse = refsResponse.readEntity(LoadTableResponse.class); + + Assertions.assertTrue( + refsTableResponse.tableMetadata().snapshots().size() >= 1, + "Filtered response should keep at least the current ref snapshot"); + Assertions.assertEquals( + DummyCredentialProvider.DUMMY_CREDENTIAL_TYPE, + refsTableResponse.config().get(Credential.CREDENTIAL_TYPE), + "snapshots=refs must keep vended credentials after filtering"); + Assertions.assertFalse( + refsTableResponse.credentials().isEmpty(), + "snapshots=refs must keep storage-credentials after filtering"); + } + + @Test + void testFilterSnapshotsByRefsKeepsCredentials() { + TableMetadata metadata = + TableMetadata.newTableMetadata( + tableSchema, + org.apache.iceberg.PartitionSpec.unpartitioned(), + "s3://bucket/db/tbl", + ImmutableMap.of()); + org.apache.iceberg.rest.credentials.Credential credential = + IcebergRESTUtils.toRESTCredential( + "s3://bucket/db/tbl/", + ImmutableMap.of( + "s3.session-token", + "token", + "client.refresh-credentials-endpoint", + "v1/c/ns/t/credentials")); + LoadTableResponse original = + LoadTableResponse.builder() + .withTableMetadata(metadata) + .addAllConfig(ImmutableMap.of("io-impl", "org.apache.iceberg.aws.s3.S3FileIO")) + .addCredential(credential) + .build(); + + LoadTableResponse filtered = IcebergTableOperations.filterSnapshotsByRefs(original); + + Assertions.assertEquals(1, filtered.credentials().size()); + Assertions.assertEquals( + "token", filtered.credentials().get(0).config().get("s3.session-token")); + Assertions.assertEquals("org.apache.iceberg.aws.s3.S3FileIO", filtered.config().get("io-impl")); + } + @ParameterizedTest @MethodSource("org.apache.gravitino.iceberg.service.rest.IcebergRestTestUtil#testNamespaces") void testLoadTableSnapshotsAllReturnsAllSnapshots(Namespace namespace) {