Remove config parameter from PolarisStorageIntegration#getSubscopedCreds (#2235)

Instances of `PolarisStorageIntegration` are created for a particular `PolarisStorageConfigurationInfo`, the same value is then passed into `PSI.getSubscopedCreds()`.

This change removes the config parameter, as it's already known when `PolarisStorageIntegration` instances are created.

Author: snazy

polaris-core/src/main/java/org/apache/polaris/core/persistence/AtomicOperationMetaStoreManager.java

@@ -1607,14 +1607,10 @@ private void revokeGrantRecord(
             catalogId,
             entityId);
 
-    PolarisStorageConfigurationInfo storageConfigurationInfo =
-        BaseMetaStoreManager.extractStorageConfiguration(
-            callCtx.getDiagServices(), reloadedEntity.getEntity());
     try {
       AccessConfig accessConfig =
           storageIntegration.getSubscopedCreds(
               callCtx.getRealmConfig(),
-              storageConfigurationInfo,
               allowListOperation,
               allowedReadLocations,
               allowedWriteLocations);

polaris-core/src/main/java/org/apache/polaris/core/persistence/transactional/TransactionalMetaStoreManagerImpl.java

@@ -2054,14 +2054,10 @@ private PolarisEntityResolver resolveSecurableToRoleGrant(
             catalogId,
             entityId);
 
-    PolarisStorageConfigurationInfo storageConfigurationInfo =
-        BaseMetaStoreManager.extractStorageConfiguration(
-            callCtx.getDiagServices(), reloadedEntity.getEntity());
     try {
       AccessConfig accessConfig =
           storageIntegration.getSubscopedCreds(
               callCtx.getRealmConfig(),
-              storageConfigurationInfo,
               allowListOperation,
               allowedReadLocations,
               allowedWriteLocations);

polaris-core/src/main/java/org/apache/polaris/core/storage/InMemoryStorageIntegration.java

@@ -39,8 +39,8 @@
 public abstract class InMemoryStorageIntegration<T extends PolarisStorageConfigurationInfo>
     extends PolarisStorageIntegration<T> {
 
-  public InMemoryStorageIntegration(String identifierOrId) {
-    super(identifierOrId);
+  protected InMemoryStorageIntegration(T config, String identifierOrId) {
+    super(config, identifierOrId);
   }
 
   /**

polaris-core/src/main/java/org/apache/polaris/core/storage/PolarisStorageIntegration.java

@@ -32,11 +32,17 @@
 public abstract class PolarisStorageIntegration<T extends PolarisStorageConfigurationInfo> {
 
   private final String integrationIdentifierOrId;
+  private final T config;
 
-  public PolarisStorageIntegration(String identifierOrId) {
+  public PolarisStorageIntegration(T config, String identifierOrId) {
+    this.config = config;
     this.integrationIdentifierOrId = identifierOrId;
   }
 
+  protected T config() {
+    return config;
+  }
+
   public String getStorageIdentifierOrId() {
     return integrationIdentifierOrId;
   }
@@ -45,7 +51,6 @@ public String getStorageIdentifierOrId() {
    * Subscope the creds against the allowed read and write locations.
    *
    * @param realmConfig the call context
-   * @param storageConfig storage configuration
    * @param allowListOperation whether to allow LIST on all the provided allowed read/write
    *     locations
    * @param allowedReadLocations a set of allowed to read locations
@@ -54,7 +59,6 @@ public String getStorageIdentifierOrId() {
    */
   public abstract AccessConfig getSubscopedCreds(
       @Nonnull RealmConfig realmConfig,
-      @Nonnull T storageConfig,
       boolean allowListOperation,
       @Nonnull Set<String> allowedReadLocations,
       @Nonnull Set<String> allowedWriteLocations);

polaris-core/src/main/java/org/apache/polaris/core/storage/PolarisStorageIntegrationProvider.java

@@ -25,7 +25,6 @@
  * PolarisStorageConfigurationInfo}.
  */
 public interface PolarisStorageIntegrationProvider {
-  @SuppressWarnings("unchecked")
   <T extends PolarisStorageConfigurationInfo>
       @Nullable PolarisStorageIntegration<T> getStorageIntegrationForConfig(
           PolarisStorageConfigurationInfo polarisStorageConfigurationInfo);

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

@@ -49,17 +49,21 @@ public class AwsCredentialsStorageIntegration
   private final StsClientProvider stsClientProvider;
   private final Optional<AwsCredentialsProvider> credentialsProvider;
 
-  public AwsCredentialsStorageIntegration(StsClient fixedClient) {
-    this((destination) -> fixedClient);
+  public AwsCredentialsStorageIntegration(
+      AwsStorageConfigurationInfo config, StsClient fixedClient) {
+    this(config, (destination) -> fixedClient);
   }
 
-  public AwsCredentialsStorageIntegration(StsClientProvider stsClientProvider) {
-    this(stsClientProvider, Optional.empty());
+  public AwsCredentialsStorageIntegration(
+      AwsStorageConfigurationInfo config, StsClientProvider stsClientProvider) {
+    this(config, stsClientProvider, Optional.empty());
   }
 
   public AwsCredentialsStorageIntegration(
-      StsClientProvider stsClientProvider, Optional<AwsCredentialsProvider> credentialsProvider) {
-    super(AwsCredentialsStorageIntegration.class.getName());
+      AwsStorageConfigurationInfo config,
+      StsClientProvider stsClientProvider,
+      Optional<AwsCredentialsProvider> credentialsProvider) {
+    super(config, AwsCredentialsStorageIntegration.class.getName());
     this.stsClientProvider = stsClientProvider;
     this.credentialsProvider = credentialsProvider;
   }
@@ -68,12 +72,12 @@ public AwsCredentialsStorageIntegration(
   @Override
   public AccessConfig getSubscopedCreds(
       @Nonnull RealmConfig realmConfig,
-      @Nonnull AwsStorageConfigurationInfo storageConfig,
       boolean allowListOperation,
       @Nonnull Set<String> allowedReadLocations,
       @Nonnull Set<String> allowedWriteLocations) {
     int storageCredentialDurationSeconds =
         realmConfig.getConfig(STORAGE_CREDENTIAL_DURATION_SECONDS);
+    AwsStorageConfigurationInfo storageConfig = config();
     AssumeRoleRequest.Builder request =
         AssumeRoleRequest.builder()
             .externalId(storageConfig.getExternalId())

polaris-core/src/main/java/org/apache/polaris/core/storage/azure/AzureCredentialsStorageIntegration.java

@@ -64,8 +64,8 @@ public class AzureCredentialsStorageIntegration
 
   final DefaultAzureCredential defaultAzureCredential;
 
-  public AzureCredentialsStorageIntegration() {
-    super(AzureCredentialsStorageIntegration.class.getName());
+  public AzureCredentialsStorageIntegration(AzureStorageConfigurationInfo config) {
+    super(config, AzureCredentialsStorageIntegration.class.getName());
     // The DefaultAzureCredential will by default load the environment variables for client id,
     // client secret, tenant id
     defaultAzureCredential = new DefaultAzureCredentialBuilder().build();
@@ -74,7 +74,6 @@ public AzureCredentialsStorageIntegration() {
   @Override
   public AccessConfig getSubscopedCreds(
       @Nonnull RealmConfig realmConfig,
-      @Nonnull AzureStorageConfigurationInfo storageConfig,
       boolean allowListOperation,
       @Nonnull Set<String> allowedReadLocations,
       @Nonnull Set<String> allowedWriteLocations) {
@@ -119,7 +118,7 @@ public AccessConfig getSubscopedCreds(
         OffsetDateTime.ofInstant(
             start.plusSeconds(3600), ZoneOffset.UTC); // 1 hr to sync with AWS and GCP Access token
 
-    AccessToken accessToken = getAccessToken(storageConfig.getTenantId());
+    AccessToken accessToken = getAccessToken(config().getTenantId());
     // Get user delegation key.
     // Set the new generated user delegation key expiry to 7 days and minute 1 min
     // Azure strictly requires the end time to be <= 7 days from the current time, -1 min to avoid

polaris-core/src/main/java/org/apache/polaris/core/storage/gcp/GcpCredentialsStorageIntegration.java

@@ -59,8 +59,10 @@ public class GcpCredentialsStorageIntegration
   private final HttpTransportFactory transportFactory;
 
   public GcpCredentialsStorageIntegration(
-      GoogleCredentials sourceCredentials, HttpTransportFactory transportFactory) {
-    super(GcpCredentialsStorageIntegration.class.getName());
+      GcpStorageConfigurationInfo config,
+      GoogleCredentials sourceCredentials,
+      HttpTransportFactory transportFactory) {
+    super(config, GcpCredentialsStorageIntegration.class.getName());
     // Needed for when environment variable GOOGLE_APPLICATION_CREDENTIALS points to google service
     // account key json
     this.sourceCredentials =
@@ -71,7 +73,6 @@ public GcpCredentialsStorageIntegration(
   @Override
   public AccessConfig getSubscopedCreds(
       @Nonnull RealmConfig realmConfig,
-      @Nonnull GcpStorageConfigurationInfo storageConfig,
       boolean allowListOperation,
       @Nonnull Set<String> allowedReadLocations,
       @Nonnull Set<String> allowedWriteLocations) {

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

@@ -33,6 +33,7 @@
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.CsvSource;
 import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.Mockito;
 
 class InMemoryStorageIntegrationTest {
 
@@ -203,13 +204,14 @@ public void testValidateAccessToLocationsWithPrefixOfAllowedLocation() {
   private static final class MockInMemoryStorageIntegration
       extends InMemoryStorageIntegration<PolarisStorageConfigurationInfo> {
     public MockInMemoryStorageIntegration() {
-      super(MockInMemoryStorageIntegration.class.getName());
+      super(
+          Mockito.mock(PolarisStorageConfigurationInfo.class),
+          MockInMemoryStorageIntegration.class.getName());
     }
 
     @Override
     public AccessConfig getSubscopedCreds(
         @Nonnull RealmConfig realmConfig,
-        @Nonnull PolarisStorageConfigurationInfo storageConfig,
         boolean allowListOperation,
         @Nonnull Set<String> allowedReadLocations,
         @Nonnull Set<String> allowedWriteLocations) {