[#6005]fix(iceberg-rest-server): Support to get custom property for d…

…ynamic config provider (#6026)

### What changes were proposed in this pull request?

Users can pass custom parameters to catalog-backend via
gravitino.bypass.xxx


### Why are the changes needed?

Fix: #6005

### Does this PR introduce _any_ user-facing change?

no

### How was this patch tested?

add a new test case in TestDynamicIcebergConfigProvider

iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/provider/DynamicIcebergConfigProvider.java

@@ -18,8 +18,11 @@
  */
 package org.apache.gravitino.iceberg.service.provider;
 
+import static org.apache.gravitino.connector.BaseCatalog.CATALOG_BYPASS_PREFIX;
+
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
+import java.util.HashMap;
 import java.util.Map;
 import java.util.Optional;
 import org.apache.commons.lang3.StringUtils;
@@ -29,6 +32,7 @@
 import org.apache.gravitino.client.GravitinoAdminClient;
 import org.apache.gravitino.exceptions.NoSuchCatalogException;
 import org.apache.gravitino.iceberg.common.IcebergConfig;
+import org.apache.gravitino.utils.MapUtils;
 
 /**
  * This provider proxy Gravitino lakehouse-iceberg catalogs.
@@ -75,8 +79,11 @@ public Optional<IcebergConfig> getIcebergCatalogConfig(String catalogName) {
         "lakehouse-iceberg".equals(catalog.provider()),
         String.format("%s.%s is not iceberg catalog", gravitinoMetalake, catalogName));
 
-    Map<String, String> properties =
-        IcebergPropertiesUtils.toIcebergCatalogProperties(catalog.properties());
+    Map<String, String> catalogProperties = catalog.properties();
+    Map<String, String> properties = new HashMap<>();
+    properties.putAll(IcebergPropertiesUtils.toIcebergCatalogProperties(catalogProperties));
+    properties.putAll(MapUtils.getPrefixMap(catalogProperties, CATALOG_BYPASS_PREFIX));
+
     return Optional.of(new IcebergConfig(properties));
   }
 

iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/provider/TestDynamicIcebergConfigProvider.java

@@ -19,10 +19,13 @@
 package org.apache.gravitino.iceberg.service.provider;
 
 import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
 import org.apache.gravitino.Catalog;
 import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants;
 import org.apache.gravitino.client.GravitinoAdminClient;
 import org.apache.gravitino.client.GravitinoMetalake;
+import org.apache.gravitino.iceberg.common.IcebergConfig;
 import org.apache.gravitino.iceberg.common.ops.IcebergCatalogWrapper;
 import org.apache.iceberg.hive.HiveCatalog;
 import org.apache.iceberg.jdbc.JdbcCatalog;
@@ -113,4 +116,44 @@ public void testInvalidIcebergTableOps() {
         IllegalArgumentException.class,
         () -> provider.getIcebergCatalogConfig(IcebergConstants.ICEBERG_REST_DEFAULT_CATALOG));
   }
+
+  @Test
+  public void testCustomProperties() {
+    String customCatalogName = "custom_backend";
+    String customKey1 = "custom-k1";
+    String customValue1 = "custom-v1";
+    String customKey2 = "custom-k2";
+    String customValue2 = "custom-v2";
+
+    Catalog customMockCatalog = Mockito.mock(Catalog.class);
+
+    GravitinoMetalake gravitinoMetalake = Mockito.mock(GravitinoMetalake.class);
+    Mockito.when(gravitinoMetalake.loadCatalog(customCatalogName)).thenReturn(customMockCatalog);
+
+    Mockito.when(customMockCatalog.provider()).thenReturn("lakehouse-iceberg");
+
+    Mockito.when(customMockCatalog.properties())
+        .thenReturn(
+            new HashMap<String, String>() {
+              {
+                put(IcebergConstants.CATALOG_BACKEND, "custom");
+                put(IcebergConstants.CATALOG_BACKEND_NAME, customCatalogName);
+                put("gravitino.bypass." + customKey1, customValue1);
+                put(customKey2, customValue2);
+              }
+            });
+    GravitinoAdminClient client = Mockito.mock(GravitinoAdminClient.class);
+    Mockito.when(client.loadMetalake(Mockito.any())).thenReturn(gravitinoMetalake);
+    DynamicIcebergConfigProvider provider = new DynamicIcebergConfigProvider();
+    provider.setClient(client);
+    Optional<IcebergConfig> icebergCatalogConfig =
+        provider.getIcebergCatalogConfig(customCatalogName);
+    Assertions.assertTrue(icebergCatalogConfig.isPresent());
+    Map<String, String> icebergCatalogProperties =
+        icebergCatalogConfig.get().getIcebergCatalogProperties();
+    Assertions.assertEquals(icebergCatalogProperties.get(customKey1), customValue1);
+    Assertions.assertFalse(icebergCatalogProperties.containsKey(customKey2));
+    Assertions.assertEquals(
+        icebergCatalogProperties.get(IcebergConstants.CATALOG_BACKEND_NAME), customCatalogName);
+  }
 }