Allow update of AWS roleArn associated with catalog when the AWS account IDs are same

dropwizard/service/src/test/java/org/apache/polaris/service/dropwizard/admin/PolarisServiceImplIntegrationTest.java

@@ -918,14 +918,14 @@ public void testCreateListUpdateAndDeleteCatalog() {
     }
 
     // Reject update of fields that can't be currently updated
-    StorageConfigInfo modifiedStorageConfig =
+    StorageConfigInfo invalidModifiedStorageConfig =
         new AwsStorageConfigInfo(
-            "arn:aws:iam::123456789011:role/newrole", StorageConfigInfo.StorageTypeEnum.S3);
+            "arn:aws:iam::123456789012:role/newrole", StorageConfigInfo.StorageTypeEnum.S3);
     UpdateCatalogRequest badUpdateRequest =
         new UpdateCatalogRequest(
             fetchedCatalog.getEntityVersion(),
             Map.of("default-base-location", "s3://newbucket/"),
-            modifiedStorageConfig);
+            invalidModifiedStorageConfig);
     try (Response response =
         newRequest("http://localhost:%d/api/management/v1/catalogs/mycatalog")
             .put(Entity.json(badUpdateRequest))) {
@@ -939,11 +939,17 @@ public void testCreateListUpdateAndDeleteCatalog() {
           .startsWith("Cannot modify");
     }
 
+    // Allow update of fields that are supported (e.g. new default-base-location and role ARN when
+    // AWS
+    // account IDs are same)
+    StorageConfigInfo validModifiedStorageConfig =
+        new AwsStorageConfigInfo(
+            "arn:aws:iam::123456789011:role/newrole", StorageConfigInfo.StorageTypeEnum.S3);
     UpdateCatalogRequest updateRequest =
         new UpdateCatalogRequest(
             fetchedCatalog.getEntityVersion(),
             Map.of("default-base-location", "s3://newbucket/"),
-            storageConfig);
+            validModifiedStorageConfig);
 
     // 200 successful update
     try (Response response =
@@ -954,6 +960,9 @@ public void testCreateListUpdateAndDeleteCatalog() {
 
       assertThat(fetchedCatalog.getProperties().toMap())
           .isEqualTo(Map.of("default-base-location", "s3://newbucket/"));
+      assertThat(fetchedCatalog.getStorageConfigInfo())
+          .isInstanceOf(AwsStorageConfigInfo.class)
+          .hasFieldOrPropertyWithValue("roleArn", "arn:aws:iam::123456789011:role/newrole");
     }
 
     // 200 GET after update should show new properties

polaris-core/src/main/java/org/apache/polaris/core/storage/aws/AwsStorageConfigurationInfo.java

@@ -25,6 +25,7 @@
 import jakarta.annotation.Nonnull;
 import jakarta.annotation.Nullable;
 import java.util.List;
+import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 import org.apache.polaris.core.storage.PolarisStorageConfigurationInfo;
 
@@ -35,8 +36,8 @@ public class AwsStorageConfigurationInfo extends PolarisStorageConfigurationInfo
   // for allowed read and write locations for subscoping creds.
   @JsonIgnore private static final int MAX_ALLOWED_LOCATIONS = 5;
 
-  // Technically, it should be ^arn:(aws|aws-cn|aws-us-gov):iam::\d{12}:role/.+$,
-  @JsonIgnore public static final String ROLE_ARN_PATTERN = "^arn:aws:iam::\\d{12}:role/.+$";
+  // Technically, it should be ^arn:(aws|aws-cn|aws-us-gov):iam::(\d{12}):role/.+$,
+  @JsonIgnore public static final String ROLE_ARN_PATTERN = "^arn:aws:iam::(\\d{12}):role/.+$";
 
   // AWS role to be assumed
   private final @Nonnull String roleARN;
@@ -81,7 +82,7 @@ public String getFileIoImplClassName() {
     return "org.apache.iceberg.aws.s3.S3FileIO";
   }
 
-  public void validateArn(String arn) {
+  public static void validateArn(String arn) {
     if (arn == null || arn.isEmpty()) {
       throw new IllegalArgumentException("ARN cannot be null or empty");
     }
@@ -122,6 +123,22 @@ public void setRegion(@Nullable String region) {
     this.region = region;
   }
 
+  @JsonIgnore
+  public String getAwsAccountId() {
+    return parseAwsAccountId(roleARN);
+  }
+
+  private static String parseAwsAccountId(String arn) {
+    validateArn(arn);
+    Pattern pattern = Pattern.compile(ROLE_ARN_PATTERN);
+    Matcher matcher = pattern.matcher(arn);
+    if (matcher.matches()) {
+      return matcher.group(1);
+    } else {
+      throw new IllegalArgumentException("ARN does not match the expected role ARN pattern");
+    }
+  }
+
   @Override
   public String toString() {
     return MoreObjects.toStringHelper(this)

polaris-core/src/test/java/org/apache/polaris/core/storage/InMemoryStorageIntegrationTest.java

@@ -74,6 +74,21 @@ public void testValidateAccessToLocations() {
                 new PolarisStorageIntegration.ValidationResult(false, "")));
   }
 
+  @Test
+  public void testAwsAccountIdParsing() {
+    AwsStorageConfigurationInfo awsConfig =
+        new AwsStorageConfigurationInfo(
+            PolarisStorageConfigurationInfo.StorageType.S3,
+            List.of("s3://bucket/path/to/warehouse"),
+            "arn:aws:iam::012345678901:role/jdoe",
+            "us-east-2");
+
+    String expectedAccountId = "012345678901";
+    String actualAccountId = awsConfig.getAwsAccountId();
+
+    Assertions.assertThat(actualAccountId).isEqualTo(expectedAccountId);
+  }
+
   @Test
   public void testValidateAccessToLocationsWithWildcard() {
     MockInMemoryStorageIntegration storage = new MockInMemoryStorageIntegration();

service/common/src/main/java/org/apache/polaris/service/admin/PolarisAdminService.java

@@ -633,8 +633,7 @@ private void validateUpdateCatalogDiffOrThrow(
     if (currentStorageConfig instanceof AwsStorageConfigurationInfo currentAwsConfig
         && newStorageConfig instanceof AwsStorageConfigurationInfo newAwsConfig) {
 
-      if (!currentAwsConfig.getRoleARN().equals(newAwsConfig.getRoleARN())
-          || !newAwsConfig.getRoleARN().equals(currentAwsConfig.getRoleARN())) {
+      if (!currentAwsConfig.getAwsAccountId().equals(newAwsConfig.getAwsAccountId())) {
         throw new BadRequestException(
             "Cannot modify Role ARN in storage config from %s to %s",
             currentStorageConfig, newStorageConfig);