[Cherry-pick to branch-1.3] Resolve #13051 #13052 #13054 #13034

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";

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<String, Str
         throw new SchemaAlreadyExistsException("Schema %s already exists", ident);
       }
     } catch (IOException ioe) {
-      throw new RuntimeException("Failed to check if schema " + ident + " exists", ioe);
+      throw ExceptionMessages.wrap("Failed to check if schema " + ident + " exists", ioe);
     }
 
     Map<String, Path> schemaPaths = getAndCheckSchemaPaths(ident.name(), properties);
@@ -804,7 +808,7 @@ public Schema createSchema(NameIdentifier ident, String comment, Map<String, Str
               }
 
             } catch (IOException ioe) {
-              throw new RuntimeException(
+              throw ExceptionMessages.wrap(
                   "Failed to create schema " + ident + " location " + schemaPath, ioe);
             }
           }
@@ -821,7 +825,7 @@ public Schema alterSchema(NameIdentifier ident, SchemaChange... changes)
         throw new NoSuchSchemaException(SCHEMA_DOES_NOT_EXIST_MSG, ident);
       }
     } catch (IOException ioe) {
-      throw new RuntimeException("Failed to check if schema " + ident + " exists", ioe);
+      throw ExceptionMessages.wrap("Failed to check if schema " + ident + " exists", ioe);
     }
 
     // note: we need to invalidate the related fileset cache when the schema rename change is
@@ -958,7 +962,7 @@ public boolean dropSchema(NameIdentifier ident, boolean cascade) throws NonEmpty
       LOG.warn("Schema {} does not exist", ident);
       return false;
     } catch (IOException ioe) {
-      throw new RuntimeException("Failed to delete schema " + ident + " location", ioe);
+      throw ExceptionMessages.wrap("Failed to delete schema " + ident + " location", ioe);
     }
   }
 
@@ -1118,7 +1122,7 @@ private Map<String, Path> getAndCheckCatalogStorageLocations(Map<String, String>
                           + 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<String, String> 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) {

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);
   }

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);
     }
   }
 

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);

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);
     }
   }

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 "