Remove PolarisConfiguration.loadConfig (v2) (apache#1858)

(cherry picked from commit 8942f68)

Author: eric-maynard

persistence/eclipselink/src/main/java/org/apache/polaris/extension/persistence/impl/eclipselink/EclipseLinkPolarisMetaStoreManagerFactory.java

@@ -25,6 +25,7 @@
 import jakarta.inject.Inject;
 import java.nio.file.Path;
 import org.apache.polaris.core.PolarisDiagnostics;
+import org.apache.polaris.core.config.PolarisConfigurationStore;
 import org.apache.polaris.core.context.RealmContext;
 import org.apache.polaris.core.persistence.LocalPolarisMetaStoreManagerFactory;
 import org.apache.polaris.core.persistence.PolarisMetaStoreManager;
@@ -46,12 +47,13 @@ public class EclipseLinkPolarisMetaStoreManagerFactory
   @Inject PolarisStorageIntegrationProvider storageIntegrationProvider;
 
   protected EclipseLinkPolarisMetaStoreManagerFactory() {
-    this(null);
+    this(null, null);
   }
 
   @Inject
-  protected EclipseLinkPolarisMetaStoreManagerFactory(PolarisDiagnostics diagnostics) {
-    super(diagnostics);
+  protected EclipseLinkPolarisMetaStoreManagerFactory(
+      PolarisDiagnostics diagnostics, PolarisConfigurationStore configurationStore) {
+    super(diagnostics, configurationStore);
   }
 
   @Override

persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/JdbcMetaStoreManagerFactory.java

@@ -32,6 +32,7 @@
 import org.apache.polaris.core.PolarisCallContext;
 import org.apache.polaris.core.PolarisDefaultDiagServiceImpl;
 import org.apache.polaris.core.PolarisDiagnostics;
+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.entity.PolarisEntity;
@@ -74,6 +75,7 @@ public class JdbcMetaStoreManagerFactory implements MetaStoreManagerFactory {
   @Inject PolarisStorageIntegrationProvider storageIntegrationProvider;
   @Inject Instance<DataSource> dataSource;
   @Inject RelationalJdbcConfiguration relationalJdbcConfiguration;
+  @Inject PolarisConfigurationStore configurationStore;
 
   protected JdbcMetaStoreManagerFactory() {}
 
@@ -205,7 +207,8 @@ public synchronized StorageCredentialCache getOrCreateStorageCredentialCache(
       RealmContext realmContext) {
     if (!storageCredentialCacheMap.containsKey(realmContext.getRealmIdentifier())) {
       storageCredentialCacheMap.put(
-          realmContext.getRealmIdentifier(), new StorageCredentialCache());
+          realmContext.getRealmIdentifier(),
+          new StorageCredentialCache(realmContext, configurationStore));
     }
 
     return storageCredentialCacheMap.get(realmContext.getRealmIdentifier());
@@ -216,7 +219,8 @@ public synchronized EntityCache getOrCreateEntityCache(RealmContext realmContext
     if (!entityCacheMap.containsKey(realmContext.getRealmIdentifier())) {
       PolarisMetaStoreManager metaStoreManager = getOrCreateMetaStoreManager(realmContext);
       entityCacheMap.put(
-          realmContext.getRealmIdentifier(), new InMemoryEntityCache(metaStoreManager));
+          realmContext.getRealmIdentifier(),
+          new InMemoryEntityCache(realmContext, configurationStore, metaStoreManager));
     }
 
     return entityCacheMap.get(realmContext.getRealmIdentifier());

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

@@ -24,7 +24,6 @@
 import java.util.function.Function;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
-import org.apache.polaris.core.context.CallContext;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -210,25 +209,6 @@ public BehaviorChangeConfiguration<T> buildBehaviorChangeConfiguration() {
     }
   }
 
-  /**
-   * Returns the value of a `PolarisConfiguration`, or the default if it cannot be loaded. This
-   * method does not need to be used when a `CallContext` is already available
-   */
-  public static <T> T loadConfig(PolarisConfiguration<T> configuration) {
-    var callContext = CallContext.getCurrentContext();
-    if (callContext == null) {
-      LOGGER.warn(
-          String.format(
-              "Unable to load current call context; using %s = %s",
-              configuration.key, configuration.defaultValue));
-      return configuration.defaultValue;
-    }
-    return callContext
-        .getPolarisCallContext()
-        .getConfigurationStore()
-        .getConfiguration(callContext.getRealmContext(), configuration);
-  }
-
   public static <T> Builder<T> builder() {
     return new Builder<>();
   }

polaris-core/src/main/java/org/apache/polaris/core/entity/CatalogEntity.java

@@ -43,6 +43,7 @@
 import org.apache.polaris.core.admin.model.StorageConfigInfo;
 import org.apache.polaris.core.config.BehaviorChangeConfiguration;
 import org.apache.polaris.core.connection.ConnectionConfigInfoDpo;
+import org.apache.polaris.core.context.CallContext;
 import org.apache.polaris.core.secrets.UserSecretReference;
 import org.apache.polaris.core.storage.FileStorageConfigurationInfo;
 import org.apache.polaris.core.storage.PolarisStorageConfigurationInfo;
@@ -80,7 +81,7 @@ public static CatalogEntity of(PolarisBaseEntity sourceEntity) {
     return null;
   }
 
-  public static CatalogEntity fromCatalog(Catalog catalog) {
+  public static CatalogEntity fromCatalog(CallContext callContext, Catalog catalog) {
     Builder builder =
         new Builder()
             .setName(catalog.getName())
@@ -90,7 +91,7 @@ public static CatalogEntity fromCatalog(Catalog catalog) {
     internalProperties.put(CATALOG_TYPE_PROPERTY, catalog.getType().name());
     builder.setInternalProperties(internalProperties);
     builder.setStorageConfigurationInfo(
-        catalog.getStorageConfigInfo(), getDefaultBaseLocation(catalog));
+        callContext, catalog.getStorageConfigInfo(), getDefaultBaseLocation(catalog));
     return builder.build();
   }
 
@@ -247,7 +248,7 @@ public Builder setReplaceNewLocationPrefixWithCatalogDefault(String value) {
     }
 
     public Builder setStorageConfigurationInfo(
-        StorageConfigInfo storageConfigModel, String defaultBaseLocation) {
+        CallContext callContext, StorageConfigInfo storageConfigModel, String defaultBaseLocation) {
       if (storageConfigModel != null) {
         PolarisStorageConfigurationInfo config;
         Set<String> allowedLocations = new HashSet<>(storageConfigModel.getAllowedLocations());
@@ -261,7 +262,7 @@ public Builder setStorageConfigurationInfo(
           throw new BadRequestException("Must specify default base location");
         }
         allowedLocations.add(defaultBaseLocation);
-        validateMaxAllowedLocations(allowedLocations);
+        validateMaxAllowedLocations(callContext, allowedLocations);
         switch (storageConfigModel.getStorageType()) {
           case S3:
             AwsStorageConfigInfo awsConfigModel = (AwsStorageConfigInfo) storageConfigModel;
@@ -305,10 +306,15 @@ public Builder setStorageConfigurationInfo(
     }
 
     /** Validate the number of allowed locations not exceeding the max value. */
-    private void validateMaxAllowedLocations(Collection<String> allowedLocations) {
+    private void validateMaxAllowedLocations(
+        CallContext callContext, Collection<String> allowedLocations) {
       int maxAllowedLocations =
-          BehaviorChangeConfiguration.loadConfig(
-              BehaviorChangeConfiguration.STORAGE_CONFIGURATION_MAX_LOCATIONS);
+          callContext
+              .getPolarisCallContext()
+              .getConfigurationStore()
+              .getConfiguration(
+                  callContext.getRealmContext(),
+                  BehaviorChangeConfiguration.STORAGE_CONFIGURATION_MAX_LOCATIONS);
       if (maxAllowedLocations != -1 && allowedLocations.size() > maxAllowedLocations) {
         throw new IllegalArgumentException(
             String.format(

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

@@ -1619,7 +1619,7 @@ private void revokeGrantRecord(
     try {
       EnumMap<StorageAccessProperty, String> creds =
           storageIntegration.getSubscopedCreds(
-              callCtx.getDiagServices(),
+              callCtx,
               storageConfigurationInfo,
               allowListOperation,
               allowedReadLocations,

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

@@ -26,6 +26,7 @@
 import org.apache.polaris.core.PolarisCallContext;
 import org.apache.polaris.core.PolarisDefaultDiagServiceImpl;
 import org.apache.polaris.core.PolarisDiagnostics;
+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.entity.PolarisEntity;
@@ -63,10 +64,14 @@ public abstract class LocalPolarisMetaStoreManagerFactory<StoreType>
       LoggerFactory.getLogger(LocalPolarisMetaStoreManagerFactory.class);
 
   private final PolarisDiagnostics diagnostics;
+  private final PolarisConfigurationStore configurationStore;
   private boolean bootstrap;
 
-  protected LocalPolarisMetaStoreManagerFactory(@Nonnull PolarisDiagnostics diagnostics) {
+  protected LocalPolarisMetaStoreManagerFactory(
+      @Nonnull PolarisDiagnostics diagnostics,
+      @Nonnull PolarisConfigurationStore configurationStore) {
     this.diagnostics = diagnostics;
+    this.configurationStore = configurationStore;
   }
 
   protected abstract StoreType createBackingStore(@Nonnull PolarisDiagnostics diagnostics);
@@ -177,7 +182,8 @@ public synchronized StorageCredentialCache getOrCreateStorageCredentialCache(
       RealmContext realmContext) {
     if (!storageCredentialCacheMap.containsKey(realmContext.getRealmIdentifier())) {
       storageCredentialCacheMap.put(
-          realmContext.getRealmIdentifier(), new StorageCredentialCache());
+          realmContext.getRealmIdentifier(),
+          new StorageCredentialCache(realmContext, configurationStore));
     }
 
     return storageCredentialCacheMap.get(realmContext.getRealmIdentifier());
@@ -188,7 +194,8 @@ public synchronized EntityCache getOrCreateEntityCache(RealmContext realmContext
     if (!entityCacheMap.containsKey(realmContext.getRealmIdentifier())) {
       PolarisMetaStoreManager metaStoreManager = getOrCreateMetaStoreManager(realmContext);
       entityCacheMap.put(
-          realmContext.getRealmIdentifier(), new InMemoryEntityCache(metaStoreManager));
+          realmContext.getRealmIdentifier(),
+          new InMemoryEntityCache(realmContext, configurationStore, metaStoreManager));
     }
 
     return entityCacheMap.get(realmContext.getRealmIdentifier());

polaris-core/src/main/java/org/apache/polaris/core/persistence/cache/InMemoryEntityCache.java

@@ -30,7 +30,8 @@
 import org.apache.polaris.core.PolarisCallContext;
 import org.apache.polaris.core.config.BehaviorChangeConfiguration;
 import org.apache.polaris.core.config.FeatureConfiguration;
-import org.apache.polaris.core.config.PolarisConfiguration;
+import org.apache.polaris.core.config.PolarisConfigurationStore;
+import org.apache.polaris.core.context.RealmContext;
 import org.apache.polaris.core.entity.PolarisBaseEntity;
 import org.apache.polaris.core.entity.PolarisEntityType;
 import org.apache.polaris.core.entity.PolarisGrantRecord;
@@ -58,7 +59,10 @@ public class InMemoryEntityCache implements EntityCache {
    *
    * @param polarisMetaStoreManager the meta store manager implementation
    */
-  public InMemoryEntityCache(@Nonnull PolarisMetaStoreManager polarisMetaStoreManager) {
+  public InMemoryEntityCache(
+      @Nonnull RealmContext realmContext,
+      @Nonnull PolarisConfigurationStore configurationStore,
+      @Nonnull PolarisMetaStoreManager polarisMetaStoreManager) {
 
     // by name cache
     this.byName = new ConcurrentHashMap<>();
@@ -76,15 +80,19 @@ public InMemoryEntityCache(@Nonnull PolarisMetaStoreManager polarisMetaStoreMana
         };
 
     long weigherTarget =
-        PolarisConfiguration.loadConfig(FeatureConfiguration.ENTITY_CACHE_WEIGHER_TARGET);
+        configurationStore.getConfiguration(
+            realmContext, FeatureConfiguration.ENTITY_CACHE_WEIGHER_TARGET);
     Caffeine<Long, ResolvedPolarisEntity> byIdBuilder =
         Caffeine.newBuilder()
             .maximumWeight(weigherTarget)
             .weigher(EntityWeigher.asWeigher())
             .expireAfterAccess(1, TimeUnit.HOURS) // Expire entries after 1 hour of no access
             .removalListener(removalListener); // Set the removal listener
 
-    if (PolarisConfiguration.loadConfig(BehaviorChangeConfiguration.ENTITY_CACHE_SOFT_VALUES)) {
+    boolean useSoftValues =
+        configurationStore.getConfiguration(
+            realmContext, BehaviorChangeConfiguration.ENTITY_CACHE_SOFT_VALUES);
+    if (useSoftValues) {
       byIdBuilder.softValues();
     }
 

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

@@ -2053,7 +2053,7 @@ private PolarisEntityResolver resolveSecurableToRoleGrant(
     try {
       EnumMap<StorageAccessProperty, String> creds =
           storageIntegration.getSubscopedCreds(
-              callCtx.getDiagServices(),
+              callCtx,
               storageConfigurationInfo,
               allowListOperation,
               allowedReadLocations,

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

@@ -22,7 +22,7 @@
 import java.util.EnumMap;
 import java.util.Map;
 import java.util.Set;
-import org.apache.polaris.core.PolarisDiagnostics;
+import org.apache.polaris.core.context.CallContext;
 
 /**
  * Abstract of Polaris Storage Integration. It holds the reference to an object that having the
@@ -45,7 +45,7 @@ public String getStorageIdentifierOrId() {
   /**
    * Subscope the creds against the allowed read and write locations.
    *
-   * @param diagnostics the diagnostics service
+   * @param callContext the call context
    * @param storageConfig storage configuration
    * @param allowListOperation whether to allow LIST on all the provided allowed read/write
    *     locations
@@ -54,7 +54,7 @@ public String getStorageIdentifierOrId() {
    * @return An enum map including the scoped credentials
    */
   public abstract EnumMap<StorageAccessProperty, String> getSubscopedCreds(
-      @Nonnull PolarisDiagnostics diagnostics,
+      @Nonnull CallContext callContext,
       @Nonnull T storageConfig,
       boolean allowListOperation,
       @Nonnull Set<String> allowedReadLocations,

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

@@ -19,7 +19,6 @@
 package org.apache.polaris.core.storage.aws;
 
 import static org.apache.polaris.core.config.FeatureConfiguration.STORAGE_CREDENTIAL_DURATION_SECONDS;
-import static org.apache.polaris.core.config.PolarisConfiguration.loadConfig;
 
 import jakarta.annotation.Nonnull;
 import java.net.URI;
@@ -29,7 +28,7 @@
 import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Stream;
-import org.apache.polaris.core.PolarisDiagnostics;
+import org.apache.polaris.core.context.CallContext;
 import org.apache.polaris.core.storage.InMemoryStorageIntegration;
 import org.apache.polaris.core.storage.StorageAccessProperty;
 import org.apache.polaris.core.storage.StorageUtil;
@@ -63,11 +62,16 @@ public AwsCredentialsStorageIntegration(
   /** {@inheritDoc} */
   @Override
   public EnumMap<StorageAccessProperty, String> getSubscopedCreds(
-      @Nonnull PolarisDiagnostics diagnostics,
+      @Nonnull CallContext callContext,
       @Nonnull AwsStorageConfigurationInfo storageConfig,
       boolean allowListOperation,
       @Nonnull Set<String> allowedReadLocations,
       @Nonnull Set<String> allowedWriteLocations) {
+    int storageCredentialDurationSeconds =
+        callContext
+            .getPolarisCallContext()
+            .getConfigurationStore()
+            .getConfiguration(callContext.getRealmContext(), STORAGE_CREDENTIAL_DURATION_SECONDS);
     AssumeRoleRequest.Builder request =
         AssumeRoleRequest.builder()
             .externalId(storageConfig.getExternalId())
@@ -80,7 +84,7 @@ public EnumMap<StorageAccessProperty, String> getSubscopedCreds(
                         allowedReadLocations,
                         allowedWriteLocations)
                     .toJson())
-            .durationSeconds(loadConfig(STORAGE_CREDENTIAL_DURATION_SECONDS));
+            .durationSeconds(storageCredentialDurationSeconds);
     credentialsProvider.ifPresent(
         cp -> request.overrideConfiguration(b -> b.credentialsProvider(cp)));
     AssumeRoleResponse response = stsClient.assumeRole(request.build());