Address review comments

Author: poojanilangekar

extensions/federation/hadoop/src/main/java/org/apache/polaris/extensions/federation/hadoop/HadoopFederatedCatalogFactory.java

@@ -23,7 +23,7 @@
 import org.apache.hadoop.conf.Configuration;
 import org.apache.iceberg.catalog.Catalog;
 import org.apache.iceberg.hadoop.HadoopCatalog;
-import org.apache.polaris.core.catalog.NonRESTCatalogFactory;
+import org.apache.polaris.core.catalog.ExternalCatalogFactory;
 import org.apache.polaris.core.connection.AuthenticationParametersDpo;
 import org.apache.polaris.core.connection.AuthenticationType;
 import org.apache.polaris.core.connection.ConnectionConfigInfoDpo;
@@ -35,7 +35,7 @@
 /** Factory class for creating a Hadoop catalog handle based on connection configuration. */
 @ApplicationScoped
 @Identifier("hadoop")
-public class HadoopFederatedCatalogFactory implements NonRESTCatalogFactory {
+public class HadoopFederatedCatalogFactory implements ExternalCatalogFactory {
   private static final Logger LOGGER = LoggerFactory.getLogger(HadoopFederatedCatalogFactory.class);
 
   @Override

polaris-core/src/main/java/org/apache/polaris/core/catalog/NonRESTCatalogFactory.java renamed to polaris-core/src/main/java/org/apache/polaris/core/catalog/ExternalCatalogFactory.java

@@ -23,15 +23,15 @@
 import org.apache.polaris.core.secrets.UserSecretsManager;
 
 /**
- * Factory interface for creating non-REST catalog handles based on connection configuration.
+ * Factory interface for creating external catalog handles based on connection configuration.
  *
  * <p>Implementations should be annotated with CDI annotations and use the @Identifier annotation to
  * specify which connection type they support.
  */
-public interface NonRESTCatalogFactory {
+public interface ExternalCatalogFactory {
 
   /**
-   * Creates a catalog hadnle for the given non-REST connection configuration.
+   * Creates a catalog handle for the given connection configuration.
    *
    * @param connectionConfig the connection configuration
    * @param userSecretsManager the user secrets manager for handling credentials

runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalogAdapter.java

@@ -63,7 +63,7 @@
 import org.apache.iceberg.rest.responses.LoadTableResponse;
 import org.apache.polaris.core.auth.AuthenticatedPolarisPrincipal;
 import org.apache.polaris.core.auth.PolarisAuthorizer;
-import org.apache.polaris.core.catalog.NonRESTCatalogFactory;
+import org.apache.polaris.core.catalog.ExternalCatalogFactory;
 import org.apache.polaris.core.context.CallContext;
 import org.apache.polaris.core.context.RealmContext;
 import org.apache.polaris.core.entity.PolarisEntity;
@@ -147,7 +147,7 @@ public class IcebergCatalogAdapter
   private final CatalogPrefixParser prefixParser;
   private final ReservedProperties reservedProperties;
   private final CatalogHandlerUtils catalogHandlerUtils;
-  private final Instance<NonRESTCatalogFactory> nonRESTCatalogFactories;
+  private final Instance<ExternalCatalogFactory> externalCatalogFactories;
 
   @Inject
   public IcebergCatalogAdapter(
@@ -162,7 +162,7 @@ public IcebergCatalogAdapter(
       CatalogPrefixParser prefixParser,
       ReservedProperties reservedProperties,
       CatalogHandlerUtils catalogHandlerUtils,
-      @Any Instance<NonRESTCatalogFactory> nonRESTCatalogFactories) {
+      @Any Instance<ExternalCatalogFactory> externalCatalogFactories) {
     this.realmContext = realmContext;
     this.callContext = callContext;
     this.catalogFactory = catalogFactory;
@@ -174,7 +174,7 @@ public IcebergCatalogAdapter(
     this.prefixParser = prefixParser;
     this.reservedProperties = reservedProperties;
     this.catalogHandlerUtils = catalogHandlerUtils;
-    this.nonRESTCatalogFactories = nonRESTCatalogFactories;
+    this.externalCatalogFactories = externalCatalogFactories;
   }
 
   /**
@@ -212,7 +212,7 @@ IcebergCatalogHandler newHandlerWrapper(SecurityContext securityContext, String
         polarisAuthorizer,
         reservedProperties,
         catalogHandlerUtils,
-        nonRESTCatalogFactories);
+        externalCatalogFactories);
   }
 
   @Override

runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalogHandler.java

@@ -47,7 +47,6 @@
 import org.apache.iceberg.UpdateRequirement;
 import org.apache.iceberg.catalog.Catalog;
 import org.apache.iceberg.catalog.Namespace;
-import org.apache.iceberg.catalog.SessionCatalog;
 import org.apache.iceberg.catalog.SupportsNamespaces;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.catalog.ViewCatalog;
@@ -56,8 +55,6 @@
 import org.apache.iceberg.exceptions.CommitFailedException;
 import org.apache.iceberg.exceptions.ForbiddenException;
 import org.apache.iceberg.exceptions.NoSuchTableException;
-import org.apache.iceberg.rest.HTTPClient;
-import org.apache.iceberg.rest.RESTCatalog;
 import org.apache.iceberg.rest.credentials.ImmutableCredential;
 import org.apache.iceberg.rest.requests.CommitTransactionRequest;
 import org.apache.iceberg.rest.requests.CreateNamespaceRequest;
@@ -76,11 +73,10 @@
 import org.apache.iceberg.rest.responses.UpdateNamespacePropertiesResponse;
 import org.apache.polaris.core.auth.PolarisAuthorizableOperation;
 import org.apache.polaris.core.auth.PolarisAuthorizer;
-import org.apache.polaris.core.catalog.NonRESTCatalogFactory;
+import org.apache.polaris.core.catalog.ExternalCatalogFactory;
 import org.apache.polaris.core.config.FeatureConfiguration;
 import org.apache.polaris.core.connection.ConnectionConfigInfoDpo;
 import org.apache.polaris.core.connection.ConnectionType;
-import org.apache.polaris.core.connection.iceberg.IcebergRestConnectionConfigInfoDpo;
 import org.apache.polaris.core.context.CallContext;
 import org.apache.polaris.core.entity.CatalogEntity;
 import org.apache.polaris.core.entity.PolarisEntitySubType;
@@ -130,7 +126,7 @@ public class IcebergCatalogHandler extends CatalogHandler implements AutoCloseab
   private final ReservedProperties reservedProperties;
   private final CatalogHandlerUtils catalogHandlerUtils;
 
-  private final Instance<NonRESTCatalogFactory> nonRESTCatalogFactories;
+  private final Instance<ExternalCatalogFactory> externalCatalogFactories;
 
   // Catalog instance will be initialized after authorizing resolver successfully resolves
   // the catalog entity.
@@ -152,14 +148,14 @@ public IcebergCatalogHandler(
       PolarisAuthorizer authorizer,
       ReservedProperties reservedProperties,
       CatalogHandlerUtils catalogHandlerUtils,
-      Instance<NonRESTCatalogFactory> nonRESTCatalogFactories) {
+      Instance<ExternalCatalogFactory> externalCatalogFactories) {
     super(callContext, resolutionManifestFactory, securityContext, catalogName, authorizer);
     this.metaStoreManager = metaStoreManager;
     this.userSecretsManager = userSecretsManager;
     this.catalogFactory = catalogFactory;
     this.reservedProperties = reservedProperties;
     this.catalogHandlerUtils = catalogHandlerUtils;
-    this.nonRESTCatalogFactories = nonRESTCatalogFactories;
+    this.externalCatalogFactories = externalCatalogFactories;
   }
 
   /**
@@ -221,34 +217,31 @@ protected void initializeCatalog() {
       Catalog federatedCatalog;
       ConnectionType connectionType =
           ConnectionType.fromCode(connectionConfigInfoDpo.getConnectionTypeCode());
+
+      // Use the unified factory pattern for all external catalog types
+      String factoryIdentifier;
       switch (connectionType) {
         case ICEBERG_REST:
-          SessionCatalog.SessionContext context = SessionCatalog.SessionContext.createEmpty();
-          federatedCatalog =
-              new RESTCatalog(
-                  context,
-                  (config) ->
-                      HTTPClient.builder(config)
-                          .uri(config.get(org.apache.iceberg.CatalogProperties.URI))
-                          .build());
-          federatedCatalog.initialize(
-              ((IcebergRestConnectionConfigInfoDpo) connectionConfigInfoDpo).getRemoteCatalogName(),
-              connectionConfigInfoDpo.asIcebergCatalogProperties(getUserSecretsManager()));
+          factoryIdentifier = "iceberg-rest";
           break;
         case HADOOP:
-          // Use CDI to select the Hadoop federation factory at runtime
-          Instance<NonRESTCatalogFactory> hadoopFactory =
-              nonRESTCatalogFactories.select(Identifier.Literal.of("hadoop"));
-          if (!hadoopFactory.isUnsatisfied()) {
-            federatedCatalog =
-                hadoopFactory.get().createCatalog(connectionConfigInfoDpo, getUserSecretsManager());
-          } else {
-            throw new UnsupportedOperationException("Hadoop federation factory unavailable.");
-          }
+          factoryIdentifier = "hadoop";
           break;
         default:
           throw new UnsupportedOperationException("Unsupported connection type: " + connectionType);
       }
+
+      Instance<ExternalCatalogFactory> externalCatalogFactory =
+          externalCatalogFactories.select(Identifier.Literal.of(factoryIdentifier));
+      if (!externalCatalogFactory.isUnsatisfied()) {
+        federatedCatalog =
+            externalCatalogFactory
+                .get()
+                .createCatalog(connectionConfigInfoDpo, getUserSecretsManager());
+      } else {
+        throw new UnsupportedOperationException(
+            "External catalog factory for type '" + connectionType + "' is unavailable.");
+      }
       this.baseCatalog = federatedCatalog;
     } else {
       LOGGER.atInfo().log("Initializing non-federated catalog");

runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergRESTExternalCatalogFactory.java

@@ -0,0 +1,64 @@
+/*
+ * 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.catalog.iceberg;
+
+import io.smallrye.common.annotation.Identifier;
+import jakarta.enterprise.context.ApplicationScoped;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.SessionCatalog;
+import org.apache.iceberg.rest.HTTPClient;
+import org.apache.iceberg.rest.RESTCatalog;
+import org.apache.polaris.core.catalog.ExternalCatalogFactory;
+import org.apache.polaris.core.connection.ConnectionConfigInfoDpo;
+import org.apache.polaris.core.connection.iceberg.IcebergRestConnectionConfigInfoDpo;
+import org.apache.polaris.core.secrets.UserSecretsManager;
+
+/** Factory class for creating an Iceberg REST catalog handle based on connection configuration. */
+@ApplicationScoped
+@Identifier("iceberg-rest")
+public class IcebergRESTExternalCatalogFactory implements ExternalCatalogFactory {
+
+  @Override
+  public Catalog createCatalog(
+      ConnectionConfigInfoDpo connectionConfig, UserSecretsManager userSecretsManager) {
+    if (!(connectionConfig instanceof IcebergRestConnectionConfigInfoDpo)) {
+      throw new IllegalArgumentException(
+          "Expected IcebergRestConnectionConfigInfoDpo but got: "
+              + connectionConfig.getClass().getSimpleName());
+    }
+
+    IcebergRestConnectionConfigInfoDpo icebergConfig =
+        (IcebergRestConnectionConfigInfoDpo) connectionConfig;
+
+    SessionCatalog.SessionContext context = SessionCatalog.SessionContext.createEmpty();
+    RESTCatalog federatedCatalog =
+        new RESTCatalog(
+            context,
+            (config) ->
+                HTTPClient.builder(config)
+                    .uri(config.get(org.apache.iceberg.CatalogProperties.URI))
+                    .build());
+
+    federatedCatalog.initialize(
+        icebergConfig.getRemoteCatalogName(),
+        connectionConfig.asIcebergCatalogProperties(userSecretsManager));
+
+    return federatedCatalog;
+  }
+}

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

@@ -56,7 +56,7 @@
 import org.apache.polaris.core.admin.model.PrincipalWithCredentialsCredentials;
 import org.apache.polaris.core.admin.model.StorageConfigInfo;
 import org.apache.polaris.core.auth.AuthenticatedPolarisPrincipal;
-import org.apache.polaris.core.catalog.NonRESTCatalogFactory;
+import org.apache.polaris.core.catalog.ExternalCatalogFactory;
 import org.apache.polaris.core.context.CallContext;
 import org.apache.polaris.core.entity.CatalogEntity;
 import org.apache.polaris.core.entity.CatalogRoleEntity;
@@ -82,8 +82,8 @@
 public class IcebergCatalogHandlerAuthzTest extends PolarisAuthzTestBase {
 
   @SuppressWarnings("unchecked")
-  private static Instance<NonRESTCatalogFactory> emptyNonRESTCatalogFactory() {
-    Instance<NonRESTCatalogFactory> mock = Mockito.mock(Instance.class);
+  private static Instance<ExternalCatalogFactory> emptyExternalCatalogFactory() {
+    Instance<ExternalCatalogFactory> mock = Mockito.mock(Instance.class);
     Mockito.when(mock.select(Mockito.any())).thenReturn(mock);
     Mockito.when(mock.isUnsatisfied()).thenReturn(true);
     return mock;
@@ -112,7 +112,7 @@ private IcebergCatalogHandler newWrapper(
         polarisAuthorizer,
         reservedProperties,
         catalogHandlerUtils,
-        emptyNonRESTCatalogFactory());
+        emptyExternalCatalogFactory());
   }
 
   /**
@@ -254,7 +254,7 @@ public void testInsufficientPermissionsPriorToSecretRotation() {
             polarisAuthorizer,
             reservedProperties,
             catalogHandlerUtils,
-            emptyNonRESTCatalogFactory());
+            emptyExternalCatalogFactory());
 
     // a variety of actions are all disallowed because the principal's credentials must be rotated
     doTestInsufficientPrivileges(
@@ -290,7 +290,7 @@ public void testInsufficientPermissionsPriorToSecretRotation() {
             polarisAuthorizer,
             reservedProperties,
             catalogHandlerUtils,
-            emptyNonRESTCatalogFactory());
+            emptyExternalCatalogFactory());
 
     doTestSufficientPrivilegeSets(
         List.of(Set.of(PolarisPrivilege.NAMESPACE_LIST)),

runtime/service/src/testFixtures/java/org/apache/polaris/service/TestServices.java

@@ -37,7 +37,7 @@
 import org.apache.polaris.core.PolarisDiagnostics;
 import org.apache.polaris.core.auth.AuthenticatedPolarisPrincipal;
 import org.apache.polaris.core.auth.PolarisAuthorizer;
-import org.apache.polaris.core.catalog.NonRESTCatalogFactory;
+import org.apache.polaris.core.catalog.ExternalCatalogFactory;
 import org.apache.polaris.core.config.PolarisConfigurationStore;
 import org.apache.polaris.core.context.CallContext;
 import org.apache.polaris.core.context.RealmContext;
@@ -217,9 +217,10 @@ public TestServices build() {
       CatalogHandlerUtils catalogHandlerUtils =
           new CatalogHandlerUtils(callContext.getRealmConfig());
 
-      Instance<NonRESTCatalogFactory> nonRESTCatalogFactory = Mockito.mock(Instance.class);
-      Mockito.when(nonRESTCatalogFactory.select(Mockito.any())).thenReturn(nonRESTCatalogFactory);
-      Mockito.when(nonRESTCatalogFactory.isUnsatisfied()).thenReturn(true);
+      @SuppressWarnings("unchecked")
+      Instance<ExternalCatalogFactory> externalCatalogFactory = Mockito.mock(Instance.class);
+      Mockito.when(externalCatalogFactory.select(Mockito.any())).thenReturn(externalCatalogFactory);
+      Mockito.when(externalCatalogFactory.isUnsatisfied()).thenReturn(true);
 
       IcebergCatalogAdapter catalogService =
           new IcebergCatalogAdapter(
@@ -234,7 +235,7 @@ public TestServices build() {
               new DefaultCatalogPrefixParser(),
               reservedProperties,
               catalogHandlerUtils,
-              nonRESTCatalogFactory);
+              externalCatalogFactory);
 
       IcebergRestCatalogApi restApi = new IcebergRestCatalogApi(catalogService);
       IcebergRestConfigurationApi restConfigurationApi =