[lake/Iceberg] Support use type to load iceberg catalog (#1536)

Author: jackylee-ch

fluss-lake/fluss-lake-iceberg/src/main/java/com/alibaba/fluss/lake/iceberg/IcebergLakeCatalog.java

@@ -48,7 +48,7 @@
 import static com.alibaba.fluss.metadata.TableDescriptor.BUCKET_COLUMN_NAME;
 import static com.alibaba.fluss.metadata.TableDescriptor.OFFSET_COLUMN_NAME;
 import static com.alibaba.fluss.metadata.TableDescriptor.TIMESTAMP_COLUMN_NAME;
-import static org.apache.iceberg.CatalogUtil.loadCatalog;
+import static org.apache.iceberg.CatalogUtil.buildIcebergCatalog;
 
 /** An Iceberg implementation of {@link LakeCatalog}. */
 public class IcebergLakeCatalog implements LakeCatalog {
@@ -82,19 +82,10 @@ protected Catalog getIcebergCatalog() {
     private Catalog createIcebergCatalog(Configuration configuration) {
         Map<String, String> icebergProps = configuration.toMap();
 
-        String catalogType = icebergProps.get("type");
-        if (catalogType == null) {
-            throw new IllegalArgumentException(
-                    "Missing required Iceberg catalog type. Set 'iceberg.catalog.type' in your configuration (e.g., 'hive', 'hadoop', or 'rest').");
-        }
-
         String catalogName = icebergProps.getOrDefault("name", "fluss-iceberg-catalog");
 
-        return loadCatalog(
-                catalogType,
-                catalogName,
-                icebergProps,
-                null // Optional: pass Hadoop configuration if available
+        return buildIcebergCatalog(
+                catalogName, icebergProps, null // Optional: pass Hadoop configuration if available
                 );
     }
 

fluss-lake/fluss-lake-iceberg/src/test/java/com/alibaba/fluss/lake/iceberg/IcebergLakeCatalogTest.java

@@ -55,7 +55,7 @@ class IcebergLakeCatalogTest {
     void setupCatalog() {
         Configuration configuration = new Configuration();
         configuration.setString("warehouse", tempWarehouseDir.toURI().toString());
-        configuration.setString("type", "org.apache.iceberg.inmemory.InMemoryCatalog");
+        configuration.setString("catalog-impl", "org.apache.iceberg.inmemory.InMemoryCatalog");
         configuration.setString("name", "fluss_test_catalog");
 
         this.flussIcebergCatalog = new IcebergLakeCatalog(configuration);