Review comments

Author: fivetran-ashokborra

polaris-core/src/main/java/org/apache/polaris/core/config/FeatureConfiguration.java

@@ -353,6 +353,7 @@ public static void enforceFeatureEnabledOrThrow(
   public static final FeatureConfiguration<Boolean> ENABLE_KMS_SUPPORT_FOR_S3 =
       PolarisConfiguration.<Boolean>builder()
           .key("ENABLE_KMS_SUPPORT_FOR_S3")
+          .catalogConfig("polaris.config.enable-kms-support-for-s3")
           .description("If true, enables KMS support for S3 storage integration")
           .defaultValue(false)
           .buildFeatureConfiguration();

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

@@ -18,6 +18,7 @@
  */
 package org.apache.polaris.core.storage.aws;
 
+import static org.apache.polaris.core.config.FeatureConfiguration.ENABLE_KMS_SUPPORT_FOR_S3;
 import static org.apache.polaris.core.config.FeatureConfiguration.STORAGE_CREDENTIAL_DURATION_SECONDS;
 
 import jakarta.annotation.Nonnull;
@@ -84,10 +85,11 @@ public AccessConfig getSubscopedCreds(
             .roleSessionName("PolarisAwsCredentialsStorageIntegration")
             .policy(
                 policyString(
-                        storageConfig.getRoleARN(),
+                        storageConfig,
                         allowListOperation,
                         allowedReadLocations,
-                        allowedWriteLocations)
+                        allowedWriteLocations,
+                        callContext)
                     .toJson())
             .durationSeconds(storageCredentialDurationSeconds);
     credentialsProvider.ifPresent(
@@ -148,7 +150,8 @@ private IamPolicy policyString(
       AwsStorageConfigurationInfo awsStorageConfigurationInfo,
       boolean allowList,
       Set<String> readLocations,
-      Set<String> writeLocations) {
+      Set<String> writeLocations,
+      CallContext callContext) {
     IamPolicy.Builder policyBuilder = IamPolicy.builder();
     IamStatement.Builder allowGetObjectStatementBuilder =
         IamStatement.builder()
@@ -225,7 +228,7 @@ private IamPolicy policyString(
 
     policyBuilder.addStatement(allowGetObjectStatementBuilder.build());
 
-    if (isKMSSupported()) {
+    if (isKMSSupported(callContext)) {
       policyBuilder.addStatement(
           IamStatement.builder()
               .effect(IamEffect.ALLOW)
@@ -273,7 +276,10 @@ private String getArnPrefixFor(String roleArn) {
     return path;
   }
 
-  private boolean isKMSSupported() {
-    return PolarisConfiguration.loadConfig(FeatureConfiguration.ENABLE_KMS_SUPPORT_FOR_S3);
+  private boolean isKMSSupported(CallContext callContext) {
+    return callContext
+        .getPolarisCallContext()
+        .getConfigurationStore()
+        .getConfiguration(callContext.getRealmContext(), ENABLE_KMS_SUPPORT_FOR_S3);
   }
 }

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

@@ -19,15 +19,37 @@
 
 package org.apache.polaris.core.storage;
 
+import jakarta.annotation.Nonnull;
+import jakarta.annotation.Nullable;
+import java.time.Clock;
 import org.apache.polaris.core.PolarisCallContext;
 import org.apache.polaris.core.PolarisDiagnostics;
+import org.apache.polaris.core.config.FeatureConfiguration;
+import org.apache.polaris.core.config.PolarisConfigurationStore;
 import org.apache.polaris.core.context.CallContext;
+import org.apache.polaris.core.context.RealmContext;
 import org.apache.polaris.core.persistence.BasePersistence;
 import org.mockito.Mockito;
 
 public abstract class BaseStorageIntegrationTest {
   protected CallContext newCallContext() {
+    PolarisConfigurationStore configStore =
+        new PolarisConfigurationStore() {
+          @Override
+          public <T> @Nullable T getConfiguration(
+              @Nonnull RealmContext realmContext, String configName) {
+            if (FeatureConfiguration.ENABLE_KMS_SUPPORT_FOR_S3.key.equals(configName)) {
+              return (T) Boolean.TRUE;
+            }
+            return PolarisConfigurationStore.super.getConfiguration(realmContext, configName);
+          }
+        };
+
     return new PolarisCallContext(
-        () -> "realm", Mockito.mock(BasePersistence.class), Mockito.mock(PolarisDiagnostics.class));
+        () -> "realm",
+        Mockito.mock(BasePersistence.class),
+        Mockito.mock(PolarisDiagnostics.class),
+        configStore,
+        Clock.systemDefaultZone());
   }
 }

polaris-core/src/test/java/org/apache/polaris/service/storage/aws/AwsCredentialsStorageIntegrationTest.java

@@ -533,6 +533,8 @@ public void testGetSubscopedCredsInlinePolicyWithoutList() {
     String warehouseKeyPrefix = "path/to/warehouse";
     String firstPath = warehouseKeyPrefix + "/namespace/table";
     String secondPath = warehouseKeyPrefix + "/oldnamespace/table";
+    String region = "us-east-2";
+    String accountId = "012345678901";
     Mockito.when(stsClient.assumeRole(Mockito.isA(AssumeRoleRequest.class)))
         .thenAnswer(
             invocation -> {
@@ -807,6 +809,8 @@ public void testGetSubscopedCredsInlinePolicyWithoutWrites() {
     String warehouseKeyPrefix = "path/to/warehouse";
     String firstPath = warehouseKeyPrefix + "/namespace/table";
     String secondPath = warehouseKeyPrefix + "/oldnamespace/table";
+    String region = "us-east-2";
+    String accountId = "012345678901";
     Mockito.when(stsClient.assumeRole(Mockito.isA(AssumeRoleRequest.class)))
         .thenAnswer(
             invocation -> {
@@ -1075,6 +1079,8 @@ public void testGetSubscopedCredsInlinePolicyWithEmptyReadAndWrite() {
     String bucket = "bucket";
     String warehouseKeyPrefix = "path/to/warehouse";
     String region = "us-east-2";
+    String accountId = "012345678901";
+    String region = "us-east-2";
     Mockito.when(stsClient.assumeRole(Mockito.isA(AssumeRoleRequest.class)))
         .thenAnswer(
             invocation -> {