Use application-scoped StorageCredentialCache

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

@@ -54,7 +54,6 @@
 import org.apache.polaris.core.persistence.dao.entity.EntityResult;
 import org.apache.polaris.core.persistence.dao.entity.PrincipalSecretsResult;
 import org.apache.polaris.core.storage.PolarisStorageIntegrationProvider;
-import org.apache.polaris.core.storage.cache.StorageCredentialCache;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -70,7 +69,6 @@ public class JdbcMetaStoreManagerFactory implements MetaStoreManagerFactory {
   private static final Logger LOGGER = LoggerFactory.getLogger(JdbcMetaStoreManagerFactory.class);
 
   final Map<String, PolarisMetaStoreManager> metaStoreManagerMap = new HashMap<>();
-  final Map<String, StorageCredentialCache> storageCredentialCacheMap = new HashMap<>();
   final Map<String, EntityCache> entityCacheMap = new HashMap<>();
   final Map<String, Supplier<BasePersistence>> sessionSupplierMap = new HashMap<>();
   protected final PolarisDiagnostics diagServices = new PolarisDefaultDiagServiceImpl();
@@ -183,7 +181,6 @@ public Map<String, BaseResult> purgeRealms(Iterable<String> realms) {
       BaseResult result = metaStoreManager.purge(callContext);
       results.put(realm, result);
 
-      storageCredentialCacheMap.remove(realm);
       sessionSupplierMap.remove(realm);
       metaStoreManagerMap.remove(realm);
     }
@@ -218,17 +215,6 @@ public synchronized Supplier<BasePersistence> getOrCreateSessionSupplier(
     return sessionSupplierMap.get(realmContext.getRealmIdentifier());
   }
 
-  @Override
-  public synchronized StorageCredentialCache getOrCreateStorageCredentialCache(
-      RealmContext realmContext, RealmConfig realmConfig) {
-    if (!storageCredentialCacheMap.containsKey(realmContext.getRealmIdentifier())) {
-      storageCredentialCacheMap.put(
-          realmContext.getRealmIdentifier(), new StorageCredentialCache());
-    }
-
-    return storageCredentialCacheMap.get(realmContext.getRealmIdentifier());
-  }
-
   @Override
   public synchronized EntityCache getOrCreateEntityCache(
       RealmContext realmContext, RealmConfig realmConfig) {

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

@@ -41,7 +41,6 @@
 import org.apache.polaris.core.persistence.dao.entity.PrincipalSecretsResult;
 import org.apache.polaris.core.persistence.transactional.TransactionalMetaStoreManagerImpl;
 import org.apache.polaris.core.persistence.transactional.TransactionalPersistence;
-import org.apache.polaris.core.storage.cache.StorageCredentialCache;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -54,7 +53,6 @@ public abstract class LocalPolarisMetaStoreManagerFactory<StoreType>
     implements MetaStoreManagerFactory {
 
   final Map<String, PolarisMetaStoreManager> metaStoreManagerMap = new HashMap<>();
-  final Map<String, StorageCredentialCache> storageCredentialCacheMap = new HashMap<>();
   final Map<String, EntityCache> entityCacheMap = new HashMap<>();
   final Map<String, StoreType> backingStoreMap = new HashMap<>();
   final Map<String, Supplier<TransactionalPersistence>> sessionSupplierMap = new HashMap<>();
@@ -138,7 +136,6 @@ public Map<String, BaseResult> purgeRealms(Iterable<String> realms) {
       BaseResult result = metaStoreManager.purge(callContext);
       results.put(realm, result);
 
-      storageCredentialCacheMap.remove(realm);
       backingStoreMap.remove(realm);
       sessionSupplierMap.remove(realm);
       metaStoreManagerMap.remove(realm);
@@ -172,17 +169,6 @@ public synchronized Supplier<TransactionalPersistence> getOrCreateSessionSupplie
     return sessionSupplierMap.get(realmContext.getRealmIdentifier());
   }
 
-  @Override
-  public synchronized StorageCredentialCache getOrCreateStorageCredentialCache(
-      RealmContext realmContext, RealmConfig realmConfig) {
-    if (!storageCredentialCacheMap.containsKey(realmContext.getRealmIdentifier())) {
-      storageCredentialCacheMap.put(
-          realmContext.getRealmIdentifier(), new StorageCredentialCache());
-    }
-
-    return storageCredentialCacheMap.get(realmContext.getRealmIdentifier());
-  }
-
   @Override
   public synchronized EntityCache getOrCreateEntityCache(
       RealmContext realmContext, RealmConfig realmConfig) {

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

@@ -27,7 +27,6 @@
 import org.apache.polaris.core.persistence.cache.EntityCache;
 import org.apache.polaris.core.persistence.dao.entity.BaseResult;
 import org.apache.polaris.core.persistence.dao.entity.PrincipalSecretsResult;
-import org.apache.polaris.core.storage.cache.StorageCredentialCache;
 
 /** Configuration interface for configuring the {@link PolarisMetaStoreManager}. */
 public interface MetaStoreManagerFactory {
@@ -36,9 +35,6 @@ public interface MetaStoreManagerFactory {
 
   Supplier<? extends BasePersistence> getOrCreateSessionSupplier(RealmContext realmContext);
 
-  StorageCredentialCache getOrCreateStorageCredentialCache(
-      RealmContext realmContext, RealmConfig realmConfig);
-
   EntityCache getOrCreateEntityCache(RealmContext realmContext, RealmConfig realmConfig);
 
   Map<String, PrincipalSecretsResult> bootstrapRealms(

polaris-core/src/main/java/org/apache/polaris/core/storage/cache/StorageCredentialCache.java

@@ -46,15 +46,13 @@ public class StorageCredentialCache {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(StorageCredentialCache.class);
 
-  private static final long CACHE_MAX_NUMBER_OF_ENTRIES = 10_000L;
-
   private final LoadingCache<StorageCredentialCacheKey, StorageCredentialCacheEntry> cache;
 
   /** Initialize the creds cache */
-  public StorageCredentialCache() {
+  public StorageCredentialCache(StorageCredentialCacheConfig cacheConfig) {
     cache =
         Caffeine.newBuilder()
-            .maximumSize(CACHE_MAX_NUMBER_OF_ENTRIES)
+            .maximumSize(cacheConfig.maxEntries())
             .expireAfter(
                 Expiry.creating(
                     (StorageCredentialCacheKey key, StorageCredentialCacheEntry entry) -> {

polaris-core/src/main/java/org/apache/polaris/core/storage/cache/StorageCredentialCacheConfig.java

@@ -0,0 +1,25 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.polaris.core.storage.cache;
+
+public interface StorageCredentialCacheConfig {
+
+  long maxEntries();
+}

polaris-core/src/test/java/org/apache/polaris/core/storage/cache/StorageCredentialCacheTest.java

@@ -54,10 +54,8 @@
 
 public class StorageCredentialCacheTest {
 
-  // polaris call context
   private final PolarisCallContext callCtx;
-
-  // the meta store manager
+  private final StorageCredentialCacheConfig storageCredentialCacheConfig;
   private final PolarisMetaStoreManager metaStoreManager;
 
   private StorageCredentialCache storageCredentialCache;
@@ -71,12 +69,13 @@ public StorageCredentialCacheTest() {
     TransactionalPersistence metaStore =
         new TreeMapTransactionalPersistenceImpl(store, Mockito.mock(), RANDOM_SECRETS);
     callCtx = new PolarisCallContext(() -> "testRealm", metaStore, diagServices);
+    storageCredentialCacheConfig = () -> 10_000;
     metaStoreManager = Mockito.mock(PolarisMetaStoreManager.class);
     storageCredentialCache = newStorageCredentialCache();
   }
 
   private StorageCredentialCache newStorageCredentialCache() {
-    return new StorageCredentialCache();
+    return new StorageCredentialCache(storageCredentialCacheConfig);
   }
 
   @Test

runtime/service/src/main/java/org/apache/polaris/service/quarkus/config/QuarkusProducers.java

@@ -52,6 +52,7 @@
 import org.apache.polaris.core.secrets.UserSecretsManager;
 import org.apache.polaris.core.secrets.UserSecretsManagerFactory;
 import org.apache.polaris.core.storage.cache.StorageCredentialCache;
+import org.apache.polaris.core.storage.cache.StorageCredentialCacheConfig;
 import org.apache.polaris.service.auth.ActiveRolesProvider;
 import org.apache.polaris.service.auth.AuthenticationType;
 import org.apache.polaris.service.auth.Authenticator;
@@ -100,8 +101,9 @@ public Clock clock() {
 
   @Produces
   @ApplicationScoped
-  public StorageCredentialCache storageCredentialCache() {
-    return new StorageCredentialCache();
+  public StorageCredentialCache storageCredentialCache(
+      StorageCredentialCacheConfig storageCredentialCacheConfig) {
+    return new StorageCredentialCache(storageCredentialCacheConfig);
   }
 
   @Produces

runtime/service/src/main/java/org/apache/polaris/service/quarkus/storage/QuarkusStorageCredentialCacheConfig.java

@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.polaris.service.quarkus.storage;
+
+import io.smallrye.config.ConfigMapping;
+import io.smallrye.config.WithDefault;
+import io.smallrye.config.WithName;
+import jakarta.validation.constraints.Min;
+import org.apache.polaris.core.storage.cache.StorageCredentialCacheConfig;
+
+@ConfigMapping(prefix = "polaris.storage-credential-cache")
+public interface QuarkusStorageCredentialCacheConfig extends StorageCredentialCacheConfig {
+  @WithName("max-entries")
+  @WithDefault("10000")
+  @Min(0)
+  @Override
+  long maxEntries();
+}

runtime/service/src/test/java/org/apache/polaris/service/quarkus/catalog/IcebergCatalogHandlerAuthzTest.java

@@ -1791,7 +1791,7 @@ public void testSendNotificationSufficientPrivileges() {
 
     PolarisCallContextCatalogFactory factory =
         new PolarisCallContextCatalogFactory(
-            new RealmEntityManagerFactory(null, null) {
+            new RealmEntityManagerFactory(null, null, null) {
               @Override
               public PolarisEntityManager getOrCreateEntityManager(RealmContext realmContext) {
                 return entityManager;

runtime/service/src/test/java/org/apache/polaris/service/quarkus/catalog/IcebergCatalogTest.java

@@ -126,6 +126,7 @@
 import org.apache.polaris.core.storage.aws.AwsCredentialsStorageIntegration;
 import org.apache.polaris.core.storage.aws.AwsStorageConfigurationInfo;
 import org.apache.polaris.core.storage.cache.StorageCredentialCache;
+import org.apache.polaris.core.storage.cache.StorageCredentialCacheConfig;
 import org.apache.polaris.service.admin.PolarisAdminService;
 import org.apache.polaris.service.catalog.PolarisPassthroughResolutionView;
 import org.apache.polaris.service.catalog.iceberg.CatalogHandlerUtils;
@@ -237,6 +238,7 @@ public Map<String, String> getConfigOverrides() {
 
   @Inject MetaStoreManagerFactory metaStoreManagerFactory;
   @Inject PolarisConfigurationStore configurationStore;
+  @Inject StorageCredentialCacheConfig storageCredentialCacheConfig;
   @Inject PolarisStorageIntegrationProvider storageIntegrationProvider;
   @Inject UserSecretsManagerFactory userSecretsManagerFactory;
   @Inject PolarisDiagnostics diagServices;
@@ -248,6 +250,7 @@ public Map<String, String> getConfigOverrides() {
   private UserSecretsManager userSecretsManager;
   private PolarisCallContext polarisContext;
   private PolarisAdminService adminService;
+  private StorageCredentialCache storageCredentialCache;
   private PolarisEntityManager entityManager;
   private FileIOFactory fileIOFactory;
   private InMemoryFileIO fileIO;
@@ -286,10 +289,12 @@ public void before(TestInfo testInfo) {
             configurationStore,
             Clock.systemDefaultZone());
 
+    storageCredentialCache = new StorageCredentialCache(storageCredentialCacheConfig);
+
     entityManager =
         new PolarisEntityManager(
             metaStoreManager,
-            new StorageCredentialCache(),
+            storageCredentialCache,
             createEntityCache(polarisContext.getRealmConfig(), metaStoreManager));
 
     PrincipalEntity rootEntity =
@@ -352,7 +357,8 @@ public void before(TestInfo testInfo) {
                     .asCatalog()));
 
     RealmEntityManagerFactory realmEntityManagerFactory =
-        new RealmEntityManagerFactory(metaStoreManagerFactory, configurationStore);
+        new RealmEntityManagerFactory(
+            metaStoreManagerFactory, configurationStore, storageCredentialCache);
     this.fileIOFactory =
         new DefaultFileIOFactory(realmEntityManagerFactory, metaStoreManagerFactory);
 
@@ -986,7 +992,8 @@ public void testValidateNotificationFailToCreateFileIO() {
     FileIOFactory fileIOFactory =
         spy(
             new DefaultFileIOFactory(
-                new RealmEntityManagerFactory(metaStoreManagerFactory, configurationStore),
+                new RealmEntityManagerFactory(
+                    metaStoreManagerFactory, configurationStore, storageCredentialCache),
                 metaStoreManagerFactory));
     IcebergCatalog catalog =
         new IcebergCatalog(
@@ -1877,7 +1884,8 @@ public void testDropTableWithPurge() {
     FileIO fileIO =
         new TaskFileIOSupplier(
                 new DefaultFileIOFactory(
-                    new RealmEntityManagerFactory(metaStoreManagerFactory, configurationStore),
+                    new RealmEntityManagerFactory(
+                        metaStoreManagerFactory, configurationStore, storageCredentialCache),
                     metaStoreManagerFactory))
             .apply(taskEntity, polarisContext);
     Assertions.assertThat(fileIO).isNotNull().isInstanceOf(ExceptionMappingFileIO.class);
@@ -2021,7 +2029,8 @@ public void testFileIOWrapper() {
 
     MeasuredFileIOFactory measured =
         new MeasuredFileIOFactory(
-            new RealmEntityManagerFactory(metaStoreManagerFactory, configurationStore),
+            new RealmEntityManagerFactory(
+                metaStoreManagerFactory, configurationStore, storageCredentialCache),
             metaStoreManagerFactory);
     IcebergCatalog catalog =
         new IcebergCatalog(

runtime/service/src/test/java/org/apache/polaris/service/quarkus/catalog/IcebergCatalogViewTest.java

@@ -62,6 +62,7 @@
 import org.apache.polaris.core.secrets.UserSecretsManager;
 import org.apache.polaris.core.secrets.UserSecretsManagerFactory;
 import org.apache.polaris.core.storage.cache.StorageCredentialCache;
+import org.apache.polaris.core.storage.cache.StorageCredentialCacheConfig;
 import org.apache.polaris.service.admin.PolarisAdminService;
 import org.apache.polaris.service.catalog.PolarisPassthroughResolutionView;
 import org.apache.polaris.service.catalog.iceberg.IcebergCatalog;
@@ -130,6 +131,7 @@ public Map<String, String> getConfigOverrides() {
   @Inject MetaStoreManagerFactory metaStoreManagerFactory;
   @Inject UserSecretsManagerFactory userSecretsManagerFactory;
   @Inject PolarisConfigurationStore configurationStore;
+  @Inject StorageCredentialCacheConfig storageCredentialCacheConfig;
   @Inject PolarisDiagnostics diagServices;
   @Inject PolarisEventListener polarisEventListener;
 
@@ -176,10 +178,12 @@ public void before(TestInfo testInfo) {
             configurationStore,
             Clock.systemDefaultZone());
 
+    StorageCredentialCache storageCredentialCache =
+        new StorageCredentialCache(storageCredentialCacheConfig);
     PolarisEntityManager entityManager =
         new PolarisEntityManager(
             metaStoreManager,
-            new StorageCredentialCache(),
+            storageCredentialCache,
             new InMemoryEntityCache(polarisContext.getRealmConfig(), metaStoreManager));
 
     CallContext.setCurrentContext(polarisContext);
@@ -236,7 +240,8 @@ public void before(TestInfo testInfo) {
             polarisContext, entityManager, securityContext, CATALOG_NAME);
     FileIOFactory fileIOFactory =
         new DefaultFileIOFactory(
-            new RealmEntityManagerFactory(metaStoreManagerFactory, configurationStore),
+            new RealmEntityManagerFactory(
+                metaStoreManagerFactory, configurationStore, storageCredentialCache),
             metaStoreManagerFactory);
 
     testPolarisEventListener = (TestPolarisEventListener) polarisEventListener;

runtime/service/src/test/java/org/apache/polaris/service/quarkus/catalog/PolarisGenericTableCatalogTest.java

@@ -67,6 +67,7 @@
 import org.apache.polaris.core.storage.aws.AwsCredentialsStorageIntegration;
 import org.apache.polaris.core.storage.aws.AwsStorageConfigurationInfo;
 import org.apache.polaris.core.storage.cache.StorageCredentialCache;
+import org.apache.polaris.core.storage.cache.StorageCredentialCacheConfig;
 import org.apache.polaris.service.admin.PolarisAdminService;
 import org.apache.polaris.service.catalog.PolarisPassthroughResolutionView;
 import org.apache.polaris.service.catalog.generic.PolarisGenericTableCatalog;
@@ -120,6 +121,7 @@ public Map<String, String> getConfigOverrides() {
   @Inject MetaStoreManagerFactory metaStoreManagerFactory;
   @Inject UserSecretsManagerFactory userSecretsManagerFactory;
   @Inject PolarisConfigurationStore configurationStore;
+  @Inject StorageCredentialCacheConfig storageCredentialCacheConfig;
   @Inject PolarisStorageIntegrationProvider storageIntegrationProvider;
   @Inject PolarisDiagnostics diagServices;
 
@@ -168,10 +170,12 @@ public void before(TestInfo testInfo) {
             diagServices,
             configurationStore,
             Clock.systemDefaultZone());
+    StorageCredentialCache storageCredentialCache =
+        new StorageCredentialCache(storageCredentialCacheConfig);
     entityManager =
         new PolarisEntityManager(
             metaStoreManager,
-            new StorageCredentialCache(),
+            storageCredentialCache,
             new InMemoryEntityCache(polarisContext.getRealmConfig(), metaStoreManager));
 
     PrincipalEntity rootEntity =
@@ -237,7 +241,8 @@ public void before(TestInfo testInfo) {
             polarisContext, entityManager, securityContext, CATALOG_NAME);
     TaskExecutor taskExecutor = Mockito.mock();
     RealmEntityManagerFactory realmEntityManagerFactory =
-        new RealmEntityManagerFactory(metaStoreManagerFactory, configurationStore);
+        new RealmEntityManagerFactory(
+            metaStoreManagerFactory, configurationStore, storageCredentialCache);
     this.fileIOFactory =
         new DefaultFileIOFactory(realmEntityManagerFactory, metaStoreManagerFactory);