apache · eric-maynard · Jul 21, 2025 · Jul 18, 2025
diff --git a/polaris-core/src/main/java/org/apache/polaris/core/persistence/PolarisEntityManager.java b/polaris-core/src/main/java/org/apache/polaris/core/persistence/PolarisEntityManager.java
@@ -32,7 +32,6 @@
 import org.apache.polaris.core.persistence.cache.EntityCache;
 import org.apache.polaris.core.persistence.resolver.PolarisResolutionManifest;
 import org.apache.polaris.core.persistence.resolver.Resolver;
-import org.apache.polaris.core.storage.cache.StorageCredentialCache;
 
 /**
  * Wraps logic of handling name-caching and entity-caching against a concrete underlying entity
@@ -43,22 +42,16 @@ public class PolarisEntityManager {
   private final PolarisMetaStoreManager metaStoreManager;
   private final EntityCache entityCache;
 
-  private final StorageCredentialCache credentialCache;
-
   // Lazily instantiated only a single time per entity manager.
   private ResolvedPolarisEntity implicitResolvedRootContainerEntity = null;
 
   /**
    * @param metaStoreManager the metastore manager for the current realm
-   * @param credentialCache the storage credential cache for the current realm
    * @param entityCache the entity cache to use (it may be {@code null}).
    */
   public PolarisEntityManager(
-      @Nonnull PolarisMetaStoreManager metaStoreManager,
-      @Nonnull StorageCredentialCache credentialCache,
-      @Nullable EntityCache entityCache) {
+      @Nonnull PolarisMetaStoreManager metaStoreManager, @Nullable EntityCache entityCache) {
     this.metaStoreManager = metaStoreManager;
-    this.credentialCache = credentialCache;
     this.entityCache = entityCache;
   }
 
@@ -133,8 +126,4 @@ private synchronized ResolvedPolarisEntity getSimulatedResolvedRootContainerEnti
     }
     return implicitResolvedRootContainerEntity;
   }
-
-  public StorageCredentialCache getCredentialCache() {
-    return credentialCache;
-  }
 }
diff --git a/polaris-core/src/main/java/org/apache/polaris/core/storage/cache/StorageCredentialCache.java b/polaris-core/src/main/java/org/apache/polaris/core/storage/cache/StorageCredentialCache.java
@@ -169,4 +169,9 @@ private boolean isTypeSupported(PolarisEntityType type) {
   public long getEstimatedSize() {
     return this.cache.estimatedSize();
   }
+
+  @VisibleForTesting
+  public void invalidateAll() {
+    this.cache.invalidateAll();
+  }
 }
diff --git a/.../service/src/test/java/org/apache/polaris/service/quarkus/admin/PolarisAuthzTestBase.java b/.../service/src/test/java/org/apache/polaris/service/quarkus/admin/PolarisAuthzTestBase.java
@@ -75,6 +75,7 @@
 import org.apache.polaris.core.policy.PredefinedPolicyTypes;
 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.service.admin.PolarisAdminService;
 import org.apache.polaris.service.catalog.PolarisPassthroughResolutionView;
 import org.apache.polaris.service.catalog.generic.PolarisGenericTableCatalog;
@@ -191,6 +192,7 @@ public Map<String, String> getConfigOverrides() {
   @Inject protected PolarisEventListener polarisEventListener;
   @Inject protected CatalogHandlerUtils catalogHandlerUtils;
   @Inject protected PolarisConfigurationStore configurationStore;
+  @Inject protected StorageCredentialCache storageCredentialCache;
 
   protected IcebergCatalog baseCatalog;
   protected PolarisGenericTableCatalog genericTableCatalog;
@@ -220,6 +222,8 @@ public static void setUpMocks() {
 
   @BeforeEach
   public void before(TestInfo testInfo) {
+    storageCredentialCache.invalidateAll();
+
     RealmContext realmContext = testInfo::getDisplayName;
     QuarkusMock.installMockForType(realmContext, RealmContext.class);
     metaStoreManager = managerFactory.getOrCreateMetaStoreManager(realmContext);
@@ -463,6 +467,7 @@ private void initBaseCatalog() {
             callContext, entityManager, securityContext, CATALOG_NAME);
     this.baseCatalog =
         new IcebergCatalog(
+            storageCredentialCache,
             entityManager,
             metaStoreManager,
             callContext,
@@ -492,16 +497,16 @@ public TestPolarisCallContextCatalogFactory() {
 
     @Inject
     public TestPolarisCallContextCatalogFactory(
+        StorageCredentialCache storageCredentialCache,
         RealmEntityManagerFactory entityManagerFactory,
         MetaStoreManagerFactory metaStoreManagerFactory,
-        UserSecretsManagerFactory userSecretsManagerFactory,
         TaskExecutor taskExecutor,
         FileIOFactory fileIOFactory,
         PolarisEventListener polarisEventListener) {
       super(
+          storageCredentialCache,
           entityManagerFactory,
           metaStoreManagerFactory,
-          userSecretsManagerFactory,
           taskExecutor,
           fileIOFactory,
           polarisEventListener);

diff --git a/...src/test/java/org/apache/polaris/service/quarkus/admin/PolarisS3InteroperabilityTest.java b/...src/test/java/org/apache/polaris/service/quarkus/admin/PolarisS3InteroperabilityTest.java
@@ -73,7 +73,7 @@ private static String makeTableLocation(
 
   public PolarisS3InteroperabilityTest() {
     TestServices.FileIOFactorySupplier fileIOFactorySupplier =
-        (entityManagerFactory, metaStoreManagerFactory) ->
+        (storageCredentialCache, metaStoreManagerFactory) ->
             (FileIOFactory)
                 (callContext,
                     ioImplClassName,

diff --git a/.../src/test/java/org/apache/polaris/service/quarkus/catalog/AbstractIcebergCatalogTest.java b/.../src/test/java/org/apache/polaris/service/quarkus/catalog/AbstractIcebergCatalogTest.java
@@ -125,7 +125,6 @@
 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;
@@ -134,7 +133,6 @@
 import org.apache.polaris.service.catalog.io.ExceptionMappingFileIO;
 import org.apache.polaris.service.catalog.io.FileIOFactory;
 import org.apache.polaris.service.catalog.io.MeasuredFileIOFactory;
-import org.apache.polaris.service.config.RealmEntityManagerFactory;
 import org.apache.polaris.service.config.ReservedProperties;
 import org.apache.polaris.service.events.AfterTableCommitedEvent;
 import org.apache.polaris.service.events.AfterTableRefreshedEvent;
@@ -225,7 +223,7 @@ public Map<String, String> getConfigOverrides() {
 
   @Inject MetaStoreManagerFactory metaStoreManagerFactory;
   @Inject PolarisConfigurationStore configurationStore;
-  @Inject StorageCredentialCacheConfig storageCredentialCacheConfig;
+  @Inject StorageCredentialCache storageCredentialCache;
   @Inject PolarisStorageIntegrationProvider storageIntegrationProvider;
   @Inject UserSecretsManagerFactory userSecretsManagerFactory;
   @Inject PolarisDiagnostics diagServices;
@@ -237,8 +235,6 @@ public Map<String, String> getConfigOverrides() {
   private UserSecretsManager userSecretsManager;
   private PolarisCallContext polarisContext;
   private PolarisAdminService adminService;
-  private StorageCredentialCache storageCredentialCache;
-  private RealmEntityManagerFactory realmEntityManagerFactory;
   private PolarisEntityManager entityManager;
   private FileIOFactory fileIOFactory;
   private InMemoryFileIO fileIO;
@@ -263,6 +259,8 @@ protected void bootstrapRealm(String realmName) {}
   @BeforeEach
   @SuppressWarnings("unchecked")
   public void before(TestInfo testInfo) {
+    storageCredentialCache.invalidateAll();
+
     realmName =
         "realm_%s_%s"
             .formatted(
@@ -281,13 +279,8 @@ public void before(TestInfo testInfo) {
             configurationStore,
             Clock.systemDefaultZone());
 
-    storageCredentialCache = new StorageCredentialCache(storageCredentialCacheConfig);
-
-    entityManager =
-        new PolarisEntityManager(
-            metaStoreManager,
-            storageCredentialCache,
-            createEntityCache(polarisContext.getRealmConfig(), metaStoreManager));
+    EntityCache entityCache = createEntityCache(polarisContext.getRealmConfig(), metaStoreManager);
+    entityManager = new PolarisEntityManager(metaStoreManager, entityCache);
 
     // LocalPolarisMetaStoreManagerFactory.bootstrapServiceAndCreatePolarisPrincipalForRealm sets
     // the CallContext.setCurrentContext() but never clears it, whereas the NoSQL one resets it.
@@ -352,11 +345,7 @@ public void before(TestInfo testInfo) {
                     .build()
                     .asCatalog()));
 
-    realmEntityManagerFactory =
-        new RealmEntityManagerFactory(
-            metaStoreManagerFactory, configurationStore, storageCredentialCache);
-    this.fileIOFactory =
-        new DefaultFileIOFactory(realmEntityManagerFactory, metaStoreManagerFactory);
+    this.fileIOFactory = new DefaultFileIOFactory(storageCredentialCache, metaStoreManagerFactory);
 
     StsClient stsClient = Mockito.mock(StsClient.class);
     when(stsClient.assumeRole(isA(AssumeRoleRequest.class)))
@@ -447,6 +436,7 @@ protected IcebergCatalog newIcebergCatalog(
             polarisContext, entityManager, securityContext, catalogName);
     TaskExecutor taskExecutor = Mockito.mock(TaskExecutor.class);
     return new IcebergCatalog(
+        storageCredentialCache,
         entityManager,
         metaStoreManager,
         polarisContext,
@@ -996,7 +986,7 @@ public void testValidateNotificationFailToCreateFileIO() {
     final String tableLocation = "s3://externally-owned-bucket/validate_table/";
     final String tableMetadataLocation = tableLocation + "metadata/";
     FileIOFactory fileIOFactory =
-        spy(new DefaultFileIOFactory(realmEntityManagerFactory, metaStoreManagerFactory));
+        spy(new DefaultFileIOFactory(storageCredentialCache, metaStoreManagerFactory));
     IcebergCatalog catalog = newIcebergCatalog(catalog().name(), metaStoreManager, fileIOFactory);
     catalog.initialize(
         CATALOG_NAME,
@@ -1849,7 +1839,7 @@ public void testDropTableWithPurge() {
         .containsEntry(StorageAccessProperty.AWS_TOKEN, SESSION_TOKEN);
     FileIO fileIO =
         new TaskFileIOSupplier(
-                new DefaultFileIOFactory(realmEntityManagerFactory, metaStoreManagerFactory))
+                new DefaultFileIOFactory(storageCredentialCache, metaStoreManagerFactory))
             .apply(taskEntity, polarisContext);
     Assertions.assertThat(fileIO).isNotNull().isInstanceOf(ExceptionMappingFileIO.class);
     Assertions.assertThat(((ExceptionMappingFileIO) fileIO).getInnerIo())
@@ -1976,7 +1966,7 @@ static Stream<Arguments> testRetriableException() {
   @Test
   public void testFileIOWrapper() {
     MeasuredFileIOFactory measured =
-        new MeasuredFileIOFactory(realmEntityManagerFactory, metaStoreManagerFactory);
+        new MeasuredFileIOFactory(storageCredentialCache, metaStoreManagerFactory);
     IcebergCatalog catalog = newIcebergCatalog(CATALOG_NAME, metaStoreManager, measured);
     catalog.initialize(
         CATALOG_NAME,

diff --git a/.../test/java/org/apache/polaris/service/quarkus/catalog/AbstractIcebergCatalogViewTest.java b/.../test/java/org/apache/polaris/service/quarkus/catalog/AbstractIcebergCatalogViewTest.java
@@ -55,16 +55,15 @@
 import org.apache.polaris.core.persistence.MetaStoreManagerFactory;
 import org.apache.polaris.core.persistence.PolarisEntityManager;
 import org.apache.polaris.core.persistence.PolarisMetaStoreManager;
+import org.apache.polaris.core.persistence.cache.EntityCache;
 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;
 import org.apache.polaris.service.catalog.io.DefaultFileIOFactory;
 import org.apache.polaris.service.catalog.io.FileIOFactory;
-import org.apache.polaris.service.config.RealmEntityManagerFactory;
 import org.apache.polaris.service.config.ReservedProperties;
 import org.apache.polaris.service.events.AfterViewCommitedEvent;
 import org.apache.polaris.service.events.AfterViewRefreshedEvent;
@@ -115,7 +114,7 @@ public Map<String, String> getConfigOverrides() {
   @Inject MetaStoreManagerFactory metaStoreManagerFactory;
   @Inject UserSecretsManagerFactory userSecretsManagerFactory;
   @Inject PolarisConfigurationStore configurationStore;
-  @Inject StorageCredentialCacheConfig storageCredentialCacheConfig;
+  @Inject StorageCredentialCache storageCredentialCache;
   @Inject PolarisDiagnostics diagServices;
   @Inject PolarisEventListener polarisEventListener;
 
@@ -147,6 +146,8 @@ protected void bootstrapRealm(String realmName) {}
 
   @BeforeEach
   public void before(TestInfo testInfo) {
+    storageCredentialCache.invalidateAll();
+
     realmName =
         "realm_%s_%s"
             .formatted(
@@ -165,14 +166,10 @@ public void before(TestInfo testInfo) {
             configurationStore,
             Clock.systemDefaultZone());
 
-    StorageCredentialCache storageCredentialCache =
-        new StorageCredentialCache(storageCredentialCacheConfig);
-    PolarisEntityManager entityManager =
-        new PolarisEntityManager(
-            metaStoreManager,
-            storageCredentialCache,
-            metaStoreManagerFactory.getOrCreateEntityCache(
-                polarisContext.getRealmContext(), polarisContext.getRealmConfig()));
+    EntityCache entityCache =
+        metaStoreManagerFactory.getOrCreateEntityCache(
+            polarisContext.getRealmContext(), polarisContext.getRealmConfig());
+    PolarisEntityManager entityManager = new PolarisEntityManager(metaStoreManager, entityCache);
 
     CallContext.setCurrentContext(polarisContext);
 
@@ -226,15 +223,13 @@ public void before(TestInfo testInfo) {
     PolarisPassthroughResolutionView passthroughView =
         new PolarisPassthroughResolutionView(
             polarisContext, entityManager, securityContext, CATALOG_NAME);
-    RealmEntityManagerFactory realmEntityManagerFactory =
-        new RealmEntityManagerFactory(
-            metaStoreManagerFactory, configurationStore, storageCredentialCache);
     FileIOFactory fileIOFactory =
-        new DefaultFileIOFactory(realmEntityManagerFactory, metaStoreManagerFactory);
+        new DefaultFileIOFactory(storageCredentialCache, metaStoreManagerFactory);
 
     testPolarisEventListener = (TestPolarisEventListener) polarisEventListener;
     this.catalog =
         new IcebergCatalog(
+            storageCredentialCache,
             entityManager,
             metaStoreManager,
             polarisContext,

diff --git a/...va/org/apache/polaris/service/quarkus/catalog/AbstractPolarisGenericTableCatalogTest.java b/...va/org/apache/polaris/service/quarkus/catalog/AbstractPolarisGenericTableCatalogTest.java
@@ -56,21 +56,20 @@
 import org.apache.polaris.core.persistence.MetaStoreManagerFactory;
 import org.apache.polaris.core.persistence.PolarisEntityManager;
 import org.apache.polaris.core.persistence.PolarisMetaStoreManager;
+import org.apache.polaris.core.persistence.cache.EntityCache;
 import org.apache.polaris.core.secrets.UserSecretsManager;
 import org.apache.polaris.core.secrets.UserSecretsManagerFactory;
 import org.apache.polaris.core.storage.PolarisStorageIntegration;
 import org.apache.polaris.core.storage.PolarisStorageIntegrationProvider;
 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;
 import org.apache.polaris.service.catalog.iceberg.IcebergCatalog;
 import org.apache.polaris.service.catalog.io.DefaultFileIOFactory;
 import org.apache.polaris.service.catalog.io.FileIOFactory;
-import org.apache.polaris.service.config.RealmEntityManagerFactory;
 import org.apache.polaris.service.config.ReservedProperties;
 import org.apache.polaris.service.events.NoOpPolarisEventListener;
 import org.apache.polaris.service.storage.PolarisStorageIntegrationProviderImpl;
@@ -102,7 +101,7 @@ public abstract class AbstractPolarisGenericTableCatalogTest {
   @Inject MetaStoreManagerFactory metaStoreManagerFactory;
   @Inject UserSecretsManagerFactory userSecretsManagerFactory;
   @Inject PolarisConfigurationStore configurationStore;
-  @Inject StorageCredentialCacheConfig storageCredentialCacheConfig;
+  @Inject StorageCredentialCache storageCredentialCache;
   @Inject PolarisStorageIntegrationProvider storageIntegrationProvider;
   @Inject PolarisDiagnostics diagServices;
 
@@ -138,6 +137,8 @@ protected void bootstrapRealm(String realmName) {}
   @BeforeEach
   @SuppressWarnings("unchecked")
   public void before(TestInfo testInfo) {
+    storageCredentialCache.invalidateAll();
+
     realmName =
         "realm_%s_%s"
             .formatted(
@@ -155,14 +156,11 @@ public void before(TestInfo testInfo) {
             diagServices,
             configurationStore,
             Clock.systemDefaultZone());
-    StorageCredentialCache storageCredentialCache =
-        new StorageCredentialCache(storageCredentialCacheConfig);
-    entityManager =
-        new PolarisEntityManager(
-            metaStoreManager,
-            storageCredentialCache,
-            metaStoreManagerFactory.getOrCreateEntityCache(
-                realmContext, polarisContext.getRealmConfig()));
+
+    EntityCache entityCache =
+        metaStoreManagerFactory.getOrCreateEntityCache(
+            realmContext, polarisContext.getRealmConfig());
+    entityManager = new PolarisEntityManager(metaStoreManager, entityCache);
 
     PrincipalEntity rootEntity =
         new PrincipalEntity(
@@ -226,11 +224,7 @@ public void before(TestInfo testInfo) {
         new PolarisPassthroughResolutionView(
             polarisContext, entityManager, securityContext, CATALOG_NAME);
     TaskExecutor taskExecutor = Mockito.mock();
-    RealmEntityManagerFactory realmEntityManagerFactory =
-        new RealmEntityManagerFactory(
-            metaStoreManagerFactory, configurationStore, storageCredentialCache);
-    this.fileIOFactory =
-        new DefaultFileIOFactory(realmEntityManagerFactory, metaStoreManagerFactory);
+    this.fileIOFactory = new DefaultFileIOFactory(storageCredentialCache, metaStoreManagerFactory);
 
     StsClient stsClient = Mockito.mock(StsClient.class);
     when(stsClient.assumeRole(isA(AssumeRoleRequest.class)))
@@ -254,6 +248,7 @@ public void before(TestInfo testInfo) {
     this.genericTableCatalog.initialize(CATALOG_NAME, Map.of());
     this.icebergCatalog =
         new IcebergCatalog(
+            storageCredentialCache,
             entityManager,
             metaStoreManager,
             polarisContext,