[#13012] fix(catalog-glue): Fail fast and give actionable errors on missing AWS credentials

catalogs/catalog-glue/src/main/java/org/apache/gravitino/catalog/glue/GlueCatalogOperations.java

@@ -34,6 +34,7 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.function.Consumer;
+import java.util.function.Supplier;
 import java.util.function.UnaryOperator;
 import java.util.stream.Collectors;
 import org.apache.commons.lang3.StringUtils;
@@ -69,6 +70,7 @@
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import software.amazon.awssdk.core.exception.SdkClientException;
 import software.amazon.awssdk.core.exception.SdkException;
 import software.amazon.awssdk.services.glue.GlueClient;
 import software.amazon.awssdk.services.glue.model.CreateDatabaseRequest;
@@ -180,19 +182,20 @@ public void close() throws IOException {
   public NameIdentifier[] listSchemas(Namespace namespace) throws NoSuchCatalogException {
     List<NameIdentifier> result = new ArrayList<>();
     String nextToken = null;
+    String context = "listing schemas under " + namespace;
     try {
       do {
         GetDatabasesRequest.Builder req = GetDatabasesRequest.builder();
         applyCatalogId(catalogId, req::catalogId);
         if (nextToken != null) req.nextToken(nextToken);
-        GetDatabasesResponse resp = glueClient.getDatabases(req.build());
+        GetDatabasesResponse resp = callGlue(() -> glueClient.getDatabases(req.build()), context);
         resp.databaseList().stream()
             .map(db -> NameIdentifier.of(namespace, db.name()))
             .forEach(result::add);
         nextToken = resp.nextToken();
       } while (nextToken != null);
     } catch (GlueException e) {
-      throw GlueExceptionConverter.toSchemaException(e, "listing schemas under " + namespace);
+      throw GlueExceptionConverter.toSchemaException(e, context);
     }
     return result.toArray(new NameIdentifier[0]);
   }
@@ -218,7 +221,7 @@ public GlueSchema createSchema(
     applyCatalogId(catalogId, req::catalogId);
 
     try {
-      glueClient.createDatabase(req.build());
+      callGlue(() -> glueClient.createDatabase(req.build()), "schema " + ident.name());
     } catch (GlueException e) {
       throw GlueExceptionConverter.toSchemaException(e, "schema " + ident.name());
     }
@@ -243,7 +246,9 @@ public GlueSchema loadSchema(NameIdentifier ident) throws NoSuchSchemaException
     applyCatalogId(catalogId, req::catalogId);
     try {
       GlueSchema schema =
-          GlueSchema.fromGlueDatabase(glueClient.getDatabase(req.build()).database());
+          GlueSchema.fromGlueDatabase(
+              callGlue(() -> glueClient.getDatabase(req.build()), "schema " + ident.name())
+                  .database());
       LOG.info("Loaded Glue schema (database) {}", ident.name());
       return schema;
     } catch (GlueException e) {
@@ -289,7 +294,7 @@ public GlueSchema alterSchema(NameIdentifier ident, SchemaChange... changes)
     applyCatalogId(catalogId, req::catalogId);
 
     try {
-      glueClient.updateDatabase(req.build());
+      callGlue(() -> glueClient.updateDatabase(req.build()), "schema " + ident.name());
     } catch (GlueException e) {
       throw GlueExceptionConverter.toSchemaException(e, "schema " + ident.name());
     }
@@ -311,7 +316,13 @@ public boolean dropSchema(NameIdentifier ident, boolean cascade) throws NonEmpty
           GetTablesRequest.builder().databaseName(ident.name()).maxResults(1);
       applyCatalogId(catalogId, tabReq::catalogId);
       try {
-        if (!glueClient.getTables(tabReq.build()).tableList().isEmpty()) {
+        boolean hasTables =
+            !callGlue(
+                    () -> glueClient.getTables(tabReq.build()),
+                    "checking tables in schema " + ident.name())
+                .tableList()
+                .isEmpty();
+        if (hasTables) {
           throw new NonEmptySchemaException(
               "Schema %s is not empty. Use cascade=true to drop it with its tables.", ident.name());
         }
@@ -324,7 +335,7 @@ public boolean dropSchema(NameIdentifier ident, boolean cascade) throws NonEmpty
     DeleteDatabaseRequest.Builder req = DeleteDatabaseRequest.builder().name(ident.name());
     applyCatalogId(catalogId, req::catalogId);
     try {
-      glueClient.deleteDatabase(req.build());
+      callGlue(() -> glueClient.deleteDatabase(req.build()), "schema " + ident.name());
       LOG.info("Dropped Glue schema (database) {}", ident.name());
       return true;
     } catch (EntityNotFoundException e) {
@@ -339,12 +350,13 @@ public NameIdentifier[] listTables(Namespace namespace) throws NoSuchSchemaExcep
     String dbName = schemaName(namespace);
     List<NameIdentifier> result = new ArrayList<>();
     String nextToken = null;
+    String context = "listing tables in schema " + dbName;
     try {
       do {
         GetTablesRequest.Builder req = GetTablesRequest.builder().databaseName(dbName);
         applyCatalogId(catalogId, req::catalogId);
         if (nextToken != null) req.nextToken(nextToken);
-        GetTablesResponse resp = glueClient.getTables(req.build());
+        GetTablesResponse resp = callGlue(() -> glueClient.getTables(req.build()), context);
         resp.tableList().stream()
             .filter(t -> !isView(t))
             .filter(this::matchesFormatFilter)
@@ -355,7 +367,7 @@ public NameIdentifier[] listTables(Namespace namespace) throws NoSuchSchemaExcep
     } catch (EntityNotFoundException e) {
       throw new NoSuchSchemaException(e, "Schema %s does not exist", dbName);
     } catch (GlueException e) {
-      throw GlueExceptionConverter.toSchemaException(e, "listing tables in schema " + dbName);
+      throw GlueExceptionConverter.toSchemaException(e, context);
     }
     return result.toArray(new NameIdentifier[0]);
   }
@@ -367,7 +379,7 @@ public GlueTable loadTable(NameIdentifier ident) throws NoSuchTableException {
     applyCatalogId(catalogId, req::catalogId);
     try {
       software.amazon.awssdk.services.glue.model.Table rawGlueTable =
-          glueClient.getTable(req.build()).table();
+          callGlue(() -> glueClient.getTable(req.build()), "table " + ident.name()).table();
       rejectIfView(rawGlueTable, ident, dbName);
       GlueTable table = GlueTable.fromGlueTable(rawGlueTable, typeConverter);
 
@@ -506,7 +518,8 @@ public GlueTable alterTable(NameIdentifier ident, TableChange... changes)
     applyCatalogId(catalogId, rawReq::catalogId);
     Table rawGlueTable;
     try {
-      rawGlueTable = glueClient.getTable(rawReq.build()).table();
+      rawGlueTable =
+          callGlue(() -> glueClient.getTable(rawReq.build()), "table " + ident.name()).table();
     } catch (GlueException e) {
       throw GlueExceptionConverter.toTableException(e, "table " + ident.name());
     }
@@ -652,7 +665,10 @@ public boolean dropTable(NameIdentifier ident) {
         GetTableRequest.builder().databaseName(dbName).name(ident.name());
     applyCatalogId(catalogId, getReq::catalogId);
     try {
-      rejectIfView(glueClient.getTable(getReq.build()).table(), ident, dbName);
+      rejectIfView(
+          callGlue(() -> glueClient.getTable(getReq.build()), "table " + ident.name()).table(),
+          ident,
+          dbName);
     } catch (EntityNotFoundException e) {
       return false;
     } catch (GlueException e) {
@@ -663,7 +679,7 @@ public boolean dropTable(NameIdentifier ident) {
         DeleteTableRequest.builder().databaseName(dbName).name(ident.name());
     applyCatalogId(catalogId, req::catalogId);
     try {
-      glueClient.deleteTable(req.build());
+      callGlue(() -> glueClient.deleteTable(req.build()), "table " + ident.name());
       LOG.info("Dropped Glue table {}.{}", dbName, ident.name());
       return true;
     } catch (EntityNotFoundException e) {
@@ -717,7 +733,7 @@ private void executeCreateTable(
       String dbName, NameIdentifier ident, CreateTableRequest.Builder req) {
     applyCatalogId(catalogId, req::catalogId);
     try {
-      glueClient.createTable(req.build());
+      callGlue(() -> glueClient.createTable(req.build()), "table " + ident.name());
     } catch (EntityNotFoundException e) {
       throw new NoSuchSchemaException(e, "Schema %s does not exist", dbName);
     } catch (GlueException e) {
@@ -728,7 +744,7 @@ private void executeCreateTable(
   private void executeUpdateTable(NameIdentifier ident, UpdateTableRequest.Builder req) {
     applyCatalogId(catalogId, req::catalogId);
     try {
-      glueClient.updateTable(req.build());
+      callGlue(() -> glueClient.updateTable(req.build()), "table " + ident.name());
     } catch (GlueException e) {
       throw GlueExceptionConverter.toTableException(e, "table " + ident.name());
     }
@@ -893,7 +909,9 @@ private String databaseLocationUri(String dbName) {
     GetDatabaseRequest.Builder req = GetDatabaseRequest.builder().name(dbName);
     applyCatalogId(catalogId, req::catalogId);
     try {
-      return glueClient.getDatabase(req.build()).database().locationUri();
+      return callGlue(() -> glueClient.getDatabase(req.build()), "schema " + dbName)
+          .database()
+          .locationUri();
     } catch (GlueException e) {
       throw GlueExceptionConverter.toSchemaException(e, "schema " + dbName);
     }
@@ -957,6 +975,32 @@ static void applyCatalogId(String catalogId, Consumer<String> setter) {
     if (catalogId != null) setter.accept(catalogId);
   }
 
+  /**
+   * Translates a raw AWS SDK credential-chain failure into a message naming this connector's own
+   * {@code aws-access-key-id} / {@code aws-secret-access-key} properties, so operators are not left
+   * to guess from the SDK's generic provider-chain error. Non-credential {@link
+   * SdkClientException}s (e.g. network failures) are rethrown unchanged.
+   */
+  private static RuntimeException translateCredentialFailure(SdkClientException e, String context) {
+    return GlueExceptionConverter.isCredentialFailure(e)
+        ? GlueExceptionConverter.toCredentialException(e, context)
+        : e;
+  }
+
+  /**
+   * Invokes a Glue SDK call, translating a credential-chain {@link SdkClientException} into an
+   * actionable error naming this connector's own credential properties. {@link GlueException} is
+   * left untouched so each call site's own catch block still applies its usual (e.g.
+   * not-found/already-exists) semantics.
+   */
+  private static <T> T callGlue(Supplier<T> call, String context) {
+    try {
+      return call.get();
+    } catch (SdkClientException e) {
+      throw translateCredentialFailure(e, context);
+    }
+  }
+
   /**
    * Finds the first column matching {@code name} in {@code cols}, replaces it with the result of
    * {@code updater}, and returns {@code true} if a replacement was made.

catalogs/catalog-glue/src/main/java/org/apache/gravitino/catalog/glue/GlueClientProvider.java

@@ -18,13 +18,16 @@
  */
 package org.apache.gravitino.catalog.glue;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import java.net.URI;
 import java.util.Map;
 import org.apache.commons.lang3.StringUtils;
 import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
+import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
 import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider;
 import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
+import software.amazon.awssdk.core.exception.SdkClientException;
 import software.amazon.awssdk.regions.Region;
 import software.amazon.awssdk.services.glue.GlueClient;
 import software.amazon.awssdk.services.glue.GlueClientBuilder;
@@ -52,7 +55,8 @@ private GlueClientProvider() {}
    * @param config Catalog configuration properties.
    * @return A configured and ready-to-use {@link GlueClient}.
    * @throws IllegalArgumentException if {@code aws-region} is missing or blank, if only one of the
-   *     credential keys is provided, or if {@code aws-glue-endpoint} is not a valid URI.
+   *     credential keys is provided, if {@code aws-glue-endpoint} is not a valid URI, or if no
+   *     usable AWS credential source can be resolved.
    */
   public static GlueClient buildClient(Map<String, String> config) {
     String region = config.get(GlueConstants.AWS_REGION);
@@ -76,12 +80,12 @@ public static GlueClient buildClient(Map<String, String> config) {
     String secretKey = config.get(GlueConstants.AWS_SECRET_ACCESS_KEY);
     boolean hasStaticCredentials = hasAwsStaticCredentials(accessKey, secretKey);
 
-    if (hasStaticCredentials) {
-      builder.credentialsProvider(
-          StaticCredentialsProvider.create(AwsBasicCredentials.create(accessKey, secretKey)));
-    } else {
-      builder.credentialsProvider(DefaultCredentialsProvider.builder().build());
-    }
+    AwsCredentialsProvider credentialsProvider =
+        hasStaticCredentials
+            ? StaticCredentialsProvider.create(AwsBasicCredentials.create(accessKey, secretKey))
+            : DefaultCredentialsProvider.builder().build();
+    validateCredentials(credentialsProvider);
+    builder.credentialsProvider(credentialsProvider);
 
     // Optional custom endpoint override for VPC endpoints or LocalStack testing.
     String endpoint = config.get(GlueConstants.AWS_GLUE_ENDPOINT);
@@ -92,6 +96,35 @@ public static GlueClient buildClient(Map<String, String> config) {
     return builder.build();
   }
 
+  /**
+   * Eagerly resolves {@code credentialsProvider} to confirm a usable credential source exists,
+   * instead of leaving resolution to the first real Glue API call. Without this check, a catalog
+   * created with no static credentials and no usable default-chain source (env vars, instance
+   * profile, etc.) is stored successfully and then fails on every operation with a raw AWS SDK
+   * error that never mentions this connector's own credential properties.
+   *
+   * @throws IllegalArgumentException if no credentials can be resolved
+   */
+  @VisibleForTesting
+  static void validateCredentials(AwsCredentialsProvider credentialsProvider) {
+    try {
+      credentialsProvider.resolveCredentials();
+    } catch (SdkClientException e) {
+      if (!GlueExceptionConverter.isCredentialFailure(e)) {
+        throw new IllegalArgumentException(
+            "Failed to resolve AWS credentials for the Glue catalog: " + e.getMessage(), e);
+      }
+      throw new IllegalArgumentException(
+          String.format(
+              "No usable AWS credentials found for the Glue catalog. Set both '%s' and '%s' "
+                  + "catalog properties for static authentication, or ensure the default AWS "
+                  + "credential chain (environment variables, instance profile, web identity "
+                  + "token, etc.) can resolve credentials.",
+              GlueConstants.AWS_ACCESS_KEY_ID, GlueConstants.AWS_SECRET_ACCESS_KEY),
+          e);
+    }
+  }
+
   static boolean hasAwsStaticCredentials(String accessKey, String secretKey) {
     boolean hasAccessKey = StringUtils.isNotBlank(accessKey);
     boolean hasSecretKey = StringUtils.isNotBlank(secretKey);

catalogs/catalog-glue/src/main/java/org/apache/gravitino/catalog/glue/GlueExceptionConverter.java

@@ -26,6 +26,7 @@
 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;
@@ -35,8 +36,42 @@
 /** Converts AWS Glue SDK exceptions to Gravitino exceptions. */
 final class GlueExceptionConverter {
 
+  private static final String NO_CREDENTIALS_MARKER =
+      "Unable to load credentials from any of the providers";
+
   private GlueExceptionConverter() {}
 
+  /**
+   * Whether {@code e} is the AWS SDK's default-credential-chain-exhausted error, which otherwise
+   * surfaces as a raw {@link SdkClientException} listing SDK-internal credential sources instead of
+   * this connector's own {@code aws-access-key-id} / {@code aws-secret-access-key} properties.
+   *
+   * @param e the client exception raised while calling AWS Glue
+   * @return true if {@code e} is a credential-resolution failure
+   */
+  static boolean isCredentialFailure(SdkClientException e) {
+    return e.getMessage() != null && e.getMessage().contains(NO_CREDENTIALS_MARKER);
+  }
+
+  /**
+   * Converts a credential-resolution {@link SdkClientException} into a message that names this
+   * connector's own credential properties, so operators are not left guessing which environment
+   * variable or IAM role the raw SDK message intended.
+   *
+   * @param e the credential-resolution failure
+   * @param context description of the operation context for error messages
+   * @return a Gravitino runtime exception with an actionable message
+   */
+  static RuntimeException toCredentialException(SdkClientException e, String context) {
+    return new RuntimeException(
+        String.format(
+            "Failed to authenticate with AWS Glue for %s. No usable AWS credentials were "
+                + "found. Set both '%s' and '%s' catalog properties, or ensure the default AWS "
+                + "credential chain can resolve credentials.",
+            context, GlueConstants.AWS_ACCESS_KEY_ID, GlueConstants.AWS_SECRET_ACCESS_KEY),
+        e);
+  }
+
   /**
    * Converts a {@link GlueException} to the appropriate Gravitino schema exception.
    *

catalogs/catalog-glue/src/test/java/org/apache/gravitino/catalog/glue/TestGlueCatalogSchemaOperations.java

@@ -125,6 +125,31 @@ void testListSchemasEmpty() {
     assertEquals(0, result.length);
   }
 
+  @Test
+  void testListSchemasMapsCredentialFailureToActionableMessage() {
+    Namespace ns = Namespace.of("metalake", "catalog");
+    SdkClientException cause =
+        SdkClientException.create("Unable to load credentials from any of the providers");
+    when(mockClient.getDatabases(any(GetDatabasesRequest.class))).thenThrow(cause);
+
+    RuntimeException ex = assertThrows(RuntimeException.class, () -> ops.listSchemas(ns));
+
+    assertEquals(cause, ex.getCause());
+    assertTrue(ex.getMessage().contains("aws-access-key-id"));
+    assertTrue(ex.getMessage().contains("aws-secret-access-key"));
+  }
+
+  @Test
+  void testListSchemasRethrowsNonCredentialSdkClientException() {
+    Namespace ns = Namespace.of("metalake", "catalog");
+    SdkClientException cause = SdkClientException.create("connection refused");
+    when(mockClient.getDatabases(any(GetDatabasesRequest.class))).thenThrow(cause);
+
+    SdkClientException ex = assertThrows(SdkClientException.class, () -> ops.listSchemas(ns));
+
+    assertEquals(cause, ex);
+  }
+
   // -------------------------------------------------------------------------
   // createSchema
   // -------------------------------------------------------------------------