Add newIcebergCatalog helper (#2134)

creation of `IcebergCatalog` instances was quite redundant as tests
mostly use the same parameters most of the time.

also remove an unused field in 2 other tests.

Author: XN137

runtime/service/src/test/java/org/apache/polaris/service/quarkus/admin/PolarisAuthzTestBase.java

@@ -72,7 +72,6 @@
 import org.apache.polaris.core.persistence.PolarisMetaStoreManager;
 import org.apache.polaris.core.persistence.dao.entity.EntityResult;
 import org.apache.polaris.core.persistence.resolver.PolarisResolutionManifest;
-import org.apache.polaris.core.persistence.transactional.TransactionalPersistence;
 import org.apache.polaris.core.policy.PredefinedPolicyTypes;
 import org.apache.polaris.core.secrets.UserSecretsManager;
 import org.apache.polaris.core.secrets.UserSecretsManagerFactory;
@@ -200,7 +199,6 @@ public Map<String, String> getConfigOverrides() {
   protected PolarisEntityManager entityManager;
   protected PolarisMetaStoreManager metaStoreManager;
   protected UserSecretsManager userSecretsManager;
-  protected TransactionalPersistence metaStoreSession;
   protected PolarisBaseEntity catalogEntity;
   protected PrincipalEntity principalEntity;
   protected CallContext callContext;

runtime/service/src/test/java/org/apache/polaris/service/quarkus/admin/PolarisS3InteroperabilityTest.java

@@ -31,7 +31,6 @@
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.exceptions.ForbiddenException;
 import org.apache.iceberg.inmemory.InMemoryFileIO;
-import org.apache.iceberg.io.FileIO;
 import org.apache.iceberg.rest.requests.CreateNamespaceRequest;
 import org.apache.iceberg.rest.requests.CreateTableRequest;
 import org.apache.iceberg.rest.responses.GetNamespaceResponse;
@@ -61,8 +60,6 @@ public class PolarisS3InteroperabilityTest {
           "true",
           "SUPPORTED_CATALOG_STORAGE_TYPES",
           List.of("FILE", "S3"));
-  private static final FileIO fileIO = new InMemoryFileIO();
-
   private final TestServices services;
 
   private static String makeNamespaceLocation(String catalogName, String namespace, String scheme) {

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

@@ -400,20 +400,7 @@ protected IcebergCatalog catalog() {
   @Override
   protected IcebergCatalog initCatalog(
       String catalogName, Map<String, String> additionalProperties) {
-    PolarisPassthroughResolutionView passthroughView =
-        new PolarisPassthroughResolutionView(
-            polarisContext, entityManager, securityContext, CATALOG_NAME);
-    TaskExecutor taskExecutor = Mockito.mock();
-    IcebergCatalog icebergCatalog =
-        new IcebergCatalog(
-            entityManager,
-            metaStoreManager,
-            polarisContext,
-            passthroughView,
-            securityContext,
-            taskExecutor,
-            fileIOFactory,
-            polarisEventListener);
+    IcebergCatalog icebergCatalog = newIcebergCatalog(CATALOG_NAME);
     fileIO = new InMemoryFileIO();
     icebergCatalog.setCatalogFileIo(fileIO);
     ImmutableMap.Builder<String, String> propertiesBuilder =
@@ -444,6 +431,32 @@ protected boolean supportsNotifications() {
     return true;
   }
 
+  protected IcebergCatalog newIcebergCatalog(String catalogName) {
+    return newIcebergCatalog(catalogName, metaStoreManager);
+  }
+
+  protected IcebergCatalog newIcebergCatalog(
+      String catalogName, PolarisMetaStoreManager metaStoreManager) {
+    return newIcebergCatalog(catalogName, metaStoreManager, fileIOFactory);
+  }
+
+  protected IcebergCatalog newIcebergCatalog(
+      String catalogName, PolarisMetaStoreManager metaStoreManager, FileIOFactory fileIOFactory) {
+    PolarisPassthroughResolutionView passthroughView =
+        new PolarisPassthroughResolutionView(
+            polarisContext, entityManager, securityContext, catalogName);
+    TaskExecutor taskExecutor = Mockito.mock(TaskExecutor.class);
+    return new IcebergCatalog(
+        entityManager,
+        metaStoreManager,
+        polarisContext,
+        passthroughView,
+        securityContext,
+        taskExecutor,
+        fileIOFactory,
+        polarisEventListener);
+  }
+
   @Test
   public void testEmptyNamespace() {
     IcebergCatalog catalog = catalog();
@@ -982,21 +995,9 @@ public void testValidateNotificationFailToCreateFileIO() {
     // filename.
     final String tableLocation = "s3://externally-owned-bucket/validate_table/";
     final String tableMetadataLocation = tableLocation + "metadata/";
-    PolarisPassthroughResolutionView passthroughView =
-        new PolarisPassthroughResolutionView(
-            polarisContext, entityManager, securityContext, catalog().name());
     FileIOFactory fileIOFactory =
         spy(new DefaultFileIOFactory(realmEntityManagerFactory, metaStoreManagerFactory));
-    IcebergCatalog catalog =
-        new IcebergCatalog(
-            entityManager,
-            metaStoreManager,
-            polarisContext,
-            passthroughView,
-            securityContext,
-            Mockito.mock(TaskExecutor.class),
-            fileIOFactory,
-            polarisEventListener);
+    IcebergCatalog catalog = newIcebergCatalog(catalog().name(), metaStoreManager, fileIOFactory);
     catalog.initialize(
         CATALOG_NAME,
         ImmutableMap.of(
@@ -1314,20 +1315,7 @@ public void testUpdateNotificationCreateTableWithLocalFilePrefix() {
                     .build()
                     .asCatalog()));
 
-    PolarisPassthroughResolutionView passthroughView =
-        new PolarisPassthroughResolutionView(
-            polarisContext, entityManager, securityContext, catalogWithoutStorage);
-    TaskExecutor taskExecutor = Mockito.mock();
-    IcebergCatalog catalog =
-        new IcebergCatalog(
-            entityManager,
-            metaStoreManager,
-            polarisContext,
-            passthroughView,
-            securityContext,
-            taskExecutor,
-            fileIOFactory,
-            polarisEventListener);
+    IcebergCatalog catalog = newIcebergCatalog(catalogWithoutStorage);
     catalog.initialize(
         catalogWithoutStorage,
         ImmutableMap.of(
@@ -1378,21 +1366,7 @@ public void testUpdateNotificationCreateTableWithHttpPrefix() {
                 .build()
                 .asCatalog()));
 
-    PolarisPassthroughResolutionView passthroughView =
-        new PolarisPassthroughResolutionView(
-            polarisContext, entityManager, securityContext, catalogName);
-    TaskExecutor taskExecutor = Mockito.mock();
-    InMemoryFileIO localFileIO = new InMemoryFileIO();
-    IcebergCatalog catalog =
-        new IcebergCatalog(
-            entityManager,
-            metaStoreManager,
-            polarisContext,
-            passthroughView,
-            securityContext,
-            taskExecutor,
-            fileIOFactory,
-            polarisEventListener);
+    IcebergCatalog catalog = newIcebergCatalog(catalogName);
     catalog.initialize(
         catalogName,
         ImmutableMap.of(
@@ -1909,19 +1883,8 @@ public void testDropTableWithPurgeDisabled() {
                     polarisContext, noPurgeStorageConfigModel, storageLocation)
                 .build()
                 .asCatalog()));
-    PolarisPassthroughResolutionView passthroughView =
-        new PolarisPassthroughResolutionView(
-            polarisContext, entityManager, securityContext, noPurgeCatalogName);
     IcebergCatalog noPurgeCatalog =
-        new IcebergCatalog(
-            entityManager,
-            metaStoreManager,
-            polarisContext,
-            passthroughView,
-            securityContext,
-            Mockito.mock(),
-            fileIOFactory,
-            polarisEventListener);
+        newIcebergCatalog(noPurgeCatalogName, metaStoreManager, fileIOFactory);
     noPurgeCatalog.initialize(
         noPurgeCatalogName,
         ImmutableMap.of(
@@ -2012,22 +1975,9 @@ static Stream<Arguments> testRetriableException() {
 
   @Test
   public void testFileIOWrapper() {
-    PolarisPassthroughResolutionView passthroughView =
-        new PolarisPassthroughResolutionView(
-            polarisContext, entityManager, securityContext, CATALOG_NAME);
-
     MeasuredFileIOFactory measured =
         new MeasuredFileIOFactory(realmEntityManagerFactory, metaStoreManagerFactory);
-    IcebergCatalog catalog =
-        new IcebergCatalog(
-            entityManager,
-            metaStoreManager,
-            polarisContext,
-            passthroughView,
-            securityContext,
-            Mockito.mock(),
-            measured,
-            polarisEventListener);
+    IcebergCatalog catalog = newIcebergCatalog(CATALOG_NAME, metaStoreManager, measured);
     catalog.initialize(
         CATALOG_NAME,
         ImmutableMap.of(
@@ -2116,19 +2066,7 @@ public void testConcurrencyConflictCreateTableUpdatedDuringFinalTransaction() {
     // Use a spy so that non-transactional pre-requisites succeed normally, but we inject
     // a concurrency failure at final commit.
     PolarisMetaStoreManager spyMetaStore = spy(metaStoreManager);
-    PolarisPassthroughResolutionView passthroughView =
-        new PolarisPassthroughResolutionView(
-            polarisContext, entityManager, securityContext, CATALOG_NAME);
-    final IcebergCatalog catalog =
-        new IcebergCatalog(
-            entityManager,
-            spyMetaStore,
-            polarisContext,
-            passthroughView,
-            securityContext,
-            Mockito.mock(TaskExecutor.class),
-            fileIOFactory,
-            polarisEventListener);
+    final IcebergCatalog catalog = newIcebergCatalog(CATALOG_NAME, spyMetaStore);
     catalog.initialize(
         CATALOG_NAME,
         ImmutableMap.of(
@@ -2165,19 +2103,7 @@ public void testConcurrencyConflictUpdateTableDuringFinalTransaction() {
     // Use a spy so that non-transactional pre-requisites succeed normally, but we inject
     // a concurrency failure at final commit.
     PolarisMetaStoreManager spyMetaStore = spy(metaStoreManager);
-    PolarisPassthroughResolutionView passthroughView =
-        new PolarisPassthroughResolutionView(
-            polarisContext, entityManager, securityContext, CATALOG_NAME);
-    final IcebergCatalog catalog =
-        new IcebergCatalog(
-            entityManager,
-            spyMetaStore,
-            polarisContext,
-            passthroughView,
-            securityContext,
-            Mockito.mock(TaskExecutor.class),
-            fileIOFactory,
-            polarisEventListener);
+    final IcebergCatalog catalog = newIcebergCatalog(CATALOG_NAME, spyMetaStore);
     catalog.initialize(
         CATALOG_NAME,
         ImmutableMap.of(