Eliminate usage of CatalogHandlers; introduce CatalogHandlerUtils

LICENSE

@@ -219,6 +219,7 @@ This product includes code from Apache Iceberg.
 * spec/polaris-catalog-apis/oauth-tokens-api.yaml
 * integration-tests/src/main/java/org/apache/polaris/service/it/test/PolarisRestCatalogIntegrationTest.java
 * service/common/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalog.java
+* service/common/src/main/java/org/apache/polaris/service/catalog/iceberg/CatalogHandlerUtils.java
 * plugins/spark/v3.5/spark/src/main/java/org/apache/polaris/spark/PolarisSparkCatalog.java
 * plugins/spark/v3.5/spark/src/test/java/org/apache/polaris/spark/PolarisInMemoryCatalog.java
 * plugins/spark/v3.5/spark/src/main/java/org/apache/polaris/spark/PolarisRESTCatalog.java

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

@@ -261,4 +261,12 @@ public static void enforceFeatureEnabledOrThrow(
           .description("The list of supported catalog connection types for federation")
           .defaultValue(List.of(ConnectionType.ICEBERG_REST.name()))
           .buildFeatureConfiguration();
+
+  public static final FeatureConfiguration<Integer> ICEBERG_COMMIT_MAX_RETRIES =
+      PolarisConfiguration.<Integer>builder()
+          .key("ICEBERG_COMMIT_MAX_RETRIES")
+          .catalogConfig("polaris.config.iceberg-commit-max-retries")
+          .description("The max number of times to try committing to an Iceberg table")
+          .defaultValue(4)
+          .buildFeatureConfiguration();
 }

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

@@ -81,6 +81,7 @@
 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.CatalogHandlerUtils;
 import org.apache.polaris.service.catalog.iceberg.IcebergCatalog;
 import org.apache.polaris.service.catalog.io.FileIOFactory;
 import org.apache.polaris.service.catalog.policy.PolicyCatalog;
@@ -192,6 +193,7 @@ public Map<String, String> getConfigOverrides() {
   @Inject protected Clock clock;
   @Inject protected FileIOFactory fileIOFactory;
   @Inject protected PolarisEventListener polarisEventListener;
+  @Inject protected CatalogHandlerUtils catalogHandlerUtils;
 
   protected IcebergCatalog baseCatalog;
   protected PolarisGenericTableCatalog genericTableCatalog;

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

@@ -116,7 +116,8 @@ private IcebergCatalogHandler newWrapper(
         factory,
         catalogName,
         polarisAuthorizer,
-        reservedProperties);
+        reservedProperties,
+        catalogHandlerUtils);
   }
 
   /**
@@ -256,7 +257,8 @@ public void testInsufficientPermissionsPriorToSecretRotation() {
             callContextCatalogFactory,
             CATALOG_NAME,
             polarisAuthorizer,
-            reservedProperties);
+            reservedProperties,
+            catalogHandlerUtils);
 
     // a variety of actions are all disallowed because the principal's credentials must be rotated
     doTestInsufficientPrivileges(
@@ -290,7 +292,8 @@ public void testInsufficientPermissionsPriorToSecretRotation() {
             callContextCatalogFactory,
             CATALOG_NAME,
             polarisAuthorizer,
-            reservedProperties);
+            reservedProperties,
+            catalogHandlerUtils);
 
     doTestSufficientPrivilegeSets(
         List.of(Set.of(PolarisPrivilege.NAMESPACE_LIST)),