AWS: Inject Glue catalog ID via interceptor instead of per-call

aws/src/main/java/org/apache/iceberg/aws/AssumeRoleAwsClientFactory.java

@@ -81,6 +81,7 @@ public GlueClient glue() {
     return GlueClient.builder()
         .applyMutation(this::applyAssumeRoleConfigurations)
         .applyMutation(httpClientProperties::applyHttpClientConfigurations)
+        .applyMutation(awsProperties::applyGlueCatalogIdConfigurations)
         .applyMutation(awsClientProperties::applyRetryConfigurations)
         .build();
   }

aws/src/main/java/org/apache/iceberg/aws/AwsClientFactories.java

@@ -147,6 +147,7 @@ public GlueClient glue() {
           .applyMutation(awsClientProperties::applyClientRegionConfiguration)
           .applyMutation(httpClientProperties::applyHttpClientConfigurations)
           .applyMutation(awsProperties::applyGlueEndpointConfigurations)
+          .applyMutation(awsProperties::applyGlueCatalogIdConfigurations)
           .applyMutation(awsClientProperties::applyClientCredentialConfigurations)
           .applyMutation(awsClientProperties::applyRetryConfigurations)
           .build();

aws/src/main/java/org/apache/iceberg/aws/AwsProperties.java

@@ -410,6 +410,22 @@ public <T extends GlueClientBuilder> void applyGlueEndpointConfigurations(T buil
     configureEndpoint(builder, glueEndpoint);
   }
 
+  /**
+   * Add the interceptor to assign a catalog id to a glue client.
+   *
+   * <p>Sample usage:
+   *
+   * <pre>
+   *     GlueClient.builder().applyMutation(awsProperties::applyGlueCatalogIdConfigurations)
+   * </pre>
+   */
+  public <T extends GlueClientBuilder> void applyGlueCatalogIdConfigurations(T builder) {
+    if (!Strings.isNullOrEmpty(glueCatalogId)) {
+      builder.overrideConfiguration(
+          c -> c.addExecutionInterceptor(new GlueCatalogIdInterceptor(glueCatalogId)));
+    }
+  }
+
   /**
    * Override the endpoint for a dynamoDb client.
    *

aws/src/main/java/org/apache/iceberg/aws/GlueCatalogIdInterceptor.java

@@ -0,0 +1,80 @@
+/*
+ * 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.iceberg.aws;
+
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.base.Strings;
+import software.amazon.awssdk.core.SdkRequest;
+import software.amazon.awssdk.core.interceptor.Context.ModifyRequest;
+import software.amazon.awssdk.core.interceptor.ExecutionAttributes;
+import software.amazon.awssdk.core.interceptor.ExecutionInterceptor;
+import software.amazon.awssdk.services.glue.model.CreateDatabaseRequest;
+import software.amazon.awssdk.services.glue.model.CreateTableRequest;
+import software.amazon.awssdk.services.glue.model.DeleteDatabaseRequest;
+import software.amazon.awssdk.services.glue.model.DeleteTableRequest;
+import software.amazon.awssdk.services.glue.model.GetDatabaseRequest;
+import software.amazon.awssdk.services.glue.model.GetDatabasesRequest;
+import software.amazon.awssdk.services.glue.model.GetTableRequest;
+import software.amazon.awssdk.services.glue.model.GetTablesRequest;
+import software.amazon.awssdk.services.glue.model.GlueRequest;
+import software.amazon.awssdk.services.glue.model.UpdateDatabaseRequest;
+import software.amazon.awssdk.services.glue.model.UpdateTableRequest;
+
+class GlueCatalogIdInterceptor implements ExecutionInterceptor {
+
+  private final String catalogId;
+
+  GlueCatalogIdInterceptor(String catalogId) {
+    Preconditions.checkArgument(
+        !Strings.isNullOrEmpty(catalogId), "Invalid catalog id: null or empty");
+    this.catalogId = catalogId;
+  }
+
+  @Override
+  public SdkRequest modifyRequest(ModifyRequest context, ExecutionAttributes executionAttributes) {
+    SdkRequest request = context.request();
+    if (!(request instanceof GlueRequest)) {
+      return request;
+    }
+
+    if (request instanceof GetDatabaseRequest) {
+      return ((GetDatabaseRequest) request).toBuilder().catalogId(catalogId).build();
+    } else if (request instanceof GetDatabasesRequest) {
+      return ((GetDatabasesRequest) request).toBuilder().catalogId(catalogId).build();
+    } else if (request instanceof CreateDatabaseRequest) {
+      return ((CreateDatabaseRequest) request).toBuilder().catalogId(catalogId).build();
+    } else if (request instanceof DeleteDatabaseRequest) {
+      return ((DeleteDatabaseRequest) request).toBuilder().catalogId(catalogId).build();
+    } else if (request instanceof UpdateDatabaseRequest) {
+      return ((UpdateDatabaseRequest) request).toBuilder().catalogId(catalogId).build();
+    } else if (request instanceof GetTableRequest) {
+      return ((GetTableRequest) request).toBuilder().catalogId(catalogId).build();
+    } else if (request instanceof GetTablesRequest) {
+      return ((GetTablesRequest) request).toBuilder().catalogId(catalogId).build();
+    } else if (request instanceof CreateTableRequest) {
+      return ((CreateTableRequest) request).toBuilder().catalogId(catalogId).build();
+    } else if (request instanceof UpdateTableRequest) {
+      return ((UpdateTableRequest) request).toBuilder().catalogId(catalogId).build();
+    } else if (request instanceof DeleteTableRequest) {
+      return ((DeleteTableRequest) request).toBuilder().catalogId(catalogId).build();
+    } else {
+      throw new IllegalArgumentException("Unexpected request: " + request.getClass().getName());
+    }
+  }
+}

aws/src/main/java/org/apache/iceberg/aws/glue/GlueCatalog.java

@@ -289,7 +289,6 @@ protected String defaultWarehouseLocation(TableIdentifier tableIdentifier) {
     GetDatabaseResponse response =
         glue.getDatabase(
             GetDatabaseRequest.builder()
-                .catalogId(awsProperties.glueCatalogId())
                 .name(
                     IcebergToGlueConverter.getDatabaseName(
                         tableIdentifier, awsProperties.glueCatalogSkipNameValidation()))
@@ -323,7 +322,6 @@ public List<TableIdentifier> listTables(Namespace namespace) {
       GetTablesResponse response =
           glue.getTables(
               GetTablesRequest.builder()
-                  .catalogId(awsProperties.glueCatalogId())
                   .databaseName(
                       IcebergToGlueConverter.toDatabaseName(
                           namespace, awsProperties.glueCatalogSkipNameValidation()))
@@ -366,7 +364,6 @@ public boolean dropTable(TableIdentifier identifier, boolean purge) {
       }
       glue.deleteTable(
           DeleteTableRequest.builder()
-              .catalogId(awsProperties.glueCatalogId())
               .databaseName(
                   IcebergToGlueConverter.getDatabaseName(
                       identifier, awsProperties.glueCatalogSkipNameValidation()))
@@ -415,11 +412,7 @@ public void renameTable(TableIdentifier from, TableIdentifier to) {
     try {
       GetTableResponse response =
           glue.getTable(
-              GetTableRequest.builder()
-                  .catalogId(awsProperties.glueCatalogId())
-                  .databaseName(fromTableDbName)
-                  .name(fromTableName)
-                  .build());
+              GetTableRequest.builder().databaseName(fromTableDbName).name(fromTableName).build());
       fromTable = response.table();
     } catch (EntityNotFoundException e) {
       throw new NoSuchTableException(
@@ -436,7 +429,6 @@ public void renameTable(TableIdentifier from, TableIdentifier to) {
 
     glue.createTable(
         CreateTableRequest.builder()
-            .catalogId(awsProperties.glueCatalogId())
             .databaseName(toTableDbName)
             .tableInput(tableInputBuilder.name(toTableName).build())
             .build());
@@ -452,11 +444,7 @@ public void renameTable(TableIdentifier from, TableIdentifier to) {
           to,
           e);
       glue.deleteTable(
-          DeleteTableRequest.builder()
-              .catalogId(awsProperties.glueCatalogId())
-              .databaseName(toTableDbName)
-              .name(toTableName)
-              .build());
+          DeleteTableRequest.builder().databaseName(toTableDbName).name(toTableName).build());
       throw e;
     }
 
@@ -468,7 +456,6 @@ public void createNamespace(Namespace namespace, Map<String, String> metadata) {
     try {
       glue.createDatabase(
           CreateDatabaseRequest.builder()
-              .catalogId(awsProperties.glueCatalogId())
               .databaseInput(
                   IcebergToGlueConverter.toDatabaseInput(
                       namespace, metadata, awsProperties.glueCatalogSkipNameValidation()))
@@ -496,11 +483,7 @@ public List<Namespace> listNamespaces(Namespace namespace) throws NoSuchNamespac
     List<Namespace> results = Lists.newArrayList();
     do {
       GetDatabasesResponse response =
-          glue.getDatabases(
-              GetDatabasesRequest.builder()
-                  .catalogId(awsProperties.glueCatalogId())
-                  .nextToken(nextToken)
-                  .build());
+          glue.getDatabases(GetDatabasesRequest.builder().nextToken(nextToken).build());
       nextToken = response.nextToken();
       if (response.hasDatabaseList()) {
         results.addAll(
@@ -522,12 +505,7 @@ public Map<String, String> loadNamespaceMetadata(Namespace namespace)
             namespace, awsProperties.glueCatalogSkipNameValidation());
     try {
       Database database =
-          glue.getDatabase(
-                  GetDatabaseRequest.builder()
-                      .catalogId(awsProperties.glueCatalogId())
-                      .name(databaseName)
-                      .build())
-              .database();
+          glue.getDatabase(GetDatabaseRequest.builder().name(databaseName).build()).database();
       Map<String, String> result = Maps.newHashMap(database.parameters());
 
       if (database.locationUri() != null) {
@@ -559,7 +537,6 @@ public boolean dropNamespace(Namespace namespace) throws NamespaceNotEmptyExcept
     GetTablesResponse response =
         glue.getTables(
             GetTablesRequest.builder()
-                .catalogId(awsProperties.glueCatalogId())
                 .databaseName(
                     IcebergToGlueConverter.toDatabaseName(
                         namespace, awsProperties.glueCatalogSkipNameValidation()))
@@ -578,7 +555,6 @@ public boolean dropNamespace(Namespace namespace) throws NamespaceNotEmptyExcept
 
     glue.deleteDatabase(
         DeleteDatabaseRequest.builder()
-            .catalogId(awsProperties.glueCatalogId())
             .name(
                 IcebergToGlueConverter.toDatabaseName(
                     namespace, awsProperties.glueCatalogSkipNameValidation()))
@@ -596,7 +572,6 @@ public boolean setProperties(Namespace namespace, Map<String, String> properties
     newProperties.putAll(properties);
     glue.updateDatabase(
         UpdateDatabaseRequest.builder()
-            .catalogId(awsProperties.glueCatalogId())
             .name(
                 IcebergToGlueConverter.toDatabaseName(
                     namespace, awsProperties.glueCatalogSkipNameValidation()))
@@ -619,7 +594,6 @@ public boolean removeProperties(Namespace namespace, Set<String> properties)
 
     glue.updateDatabase(
         UpdateDatabaseRequest.builder()
-            .catalogId(awsProperties.glueCatalogId())
             .name(
                 IcebergToGlueConverter.toDatabaseName(
                     namespace, awsProperties.glueCatalogSkipNameValidation()))

aws/src/main/java/org/apache/iceberg/aws/glue/GlueTableOperations.java

@@ -277,11 +277,7 @@ private Table getGlueTable() {
     try {
       GetTableResponse response =
           glue.getTable(
-              GetTableRequest.builder()
-                  .catalogId(awsProperties.glueCatalogId())
-                  .databaseName(databaseName)
-                  .name(tableName)
-                  .build());
+              GetTableRequest.builder().databaseName(databaseName).name(tableName).build());
       return response.table();
     } catch (EntityNotFoundException e) {
       return null;
@@ -311,7 +307,6 @@ void persistGlueTable(
       UpdateTableRequest.Builder updateTableRequest =
           UpdateTableRequest.builder()
               .overrideConfiguration(c -> c.addMetricPublisher(retryDetector))
-              .catalogId(awsProperties.glueCatalogId())
               .databaseName(databaseName)
               .skipArchive(awsProperties.glueCatalogSkipArchive())
               .tableInput(
@@ -335,7 +330,6 @@ void persistGlueTable(
       glue.createTable(
           CreateTableRequest.builder()
               .overrideConfiguration(c -> c.addMetricPublisher(retryDetector))
-              .catalogId(awsProperties.glueCatalogId())
               .databaseName(databaseName)
               .tableInput(
                   TableInput.builder()

aws/src/main/java/org/apache/iceberg/aws/lakeformation/LakeFormationAwsClientFactory.java

@@ -54,7 +54,6 @@ public class LakeFormationAwsClientFactory extends AssumeRoleAwsClientFactory {
 
   private String dbName;
   private String tableName;
-  private String glueCatalogId;
   private String glueAccountId;
 
   public LakeFormationAwsClientFactory() {}
@@ -70,7 +69,6 @@ public void initialize(Map<String, String> catalogProperties) {
         AwsProperties.CLIENT_ASSUME_ROLE_TAGS_PREFIX);
     this.dbName = catalogProperties.get(AwsProperties.LAKE_FORMATION_DB_NAME);
     this.tableName = catalogProperties.get(AwsProperties.LAKE_FORMATION_TABLE_NAME);
-    this.glueCatalogId = catalogProperties.get(AwsProperties.GLUE_CATALOG_ID);
     this.glueAccountId = catalogProperties.get(AwsProperties.GLUE_ACCOUNT_ID);
   }
 
@@ -114,13 +112,7 @@ private boolean isTableRegisteredWithLakeFormation() {
         tableName != null && !tableName.isEmpty(), "Table name can not be empty");
 
     GetTableResponse response =
-        glue()
-            .getTable(
-                GetTableRequest.builder()
-                    .catalogId(glueCatalogId)
-                    .databaseName(dbName)
-                    .name(tableName)
-                    .build());
+        glue().getTable(GetTableRequest.builder().databaseName(dbName).name(tableName).build());
     return response.table().isRegisteredWithLakeFormation();
   }