remove reduant code

Author: xx789633

fluss-common/src/main/java/org/apache/fluss/utils/PropertiesUtils.java

@@ -17,9 +17,7 @@
 
 package org.apache.fluss.utils;
 
-import java.util.Arrays;
 import java.util.Map;
-import java.util.Set;
 import java.util.stream.Collectors;
 
 /** Utility class for properties related helper functions. */
@@ -59,21 +57,6 @@ public static <V> Map<String, V> extractPrefix(Map<String, V> originalMap, Strin
                 .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
     }
 
-    public static <V> Map<String, V> extractPrefix(
-            Map<String, V> originalMap, Class<? extends Enum<?>> enumClass) {
-        Set<String> prefixes =
-                Arrays.stream(enumClass.getEnumConstants())
-                        .map(Enum::toString)
-                        .collect(Collectors.toSet());
-
-        return originalMap.entrySet().stream()
-                .filter(
-                        entry ->
-                                prefixes.stream()
-                                        .anyMatch(prefix -> entry.getKey().startsWith(prefix)))
-                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
-    }
-
     /** Filter out keys that start with the given prefix from the original map. */
     public static <T> Map<String, T> excludeByPrefix(Map<String, T> originalMap, String prefix) {
         return originalMap.entrySet().stream()

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkCatalogFactory.java

@@ -26,6 +26,7 @@
 
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -69,7 +70,11 @@ public FlinkCatalog createCatalog(Context context) {
         helper.validateExcept(PREFIXES_TO_SKIP_VALIDATE.toArray(new String[0]));
         Map<String, String> options = context.getOptions();
         Map<String, String> securityConfigs = extractPrefix(options, CLIENT_SECURITY_PREFIX);
-        Map<String, String> lakeCatalogProperties = extractPrefix(options, DataLakeFormat.class);
+
+        Map<String, String> lakeCatalogProperties = new HashMap<>();
+        for (DataLakeFormat lakeFormat : DataLakeFormat.values()) {
+            lakeCatalogProperties.putAll(extractPrefix(options, lakeFormat.toString()));
+        }
 
         return new FlinkCatalog(
                 context.getName(),

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/lake/LakeFlinkCatalog.java

@@ -21,6 +21,7 @@
 import org.apache.fluss.config.Configuration;
 import org.apache.fluss.flink.utils.DataLakeUtils;
 import org.apache.fluss.metadata.DataLakeFormat;
+import org.apache.fluss.utils.PropertiesUtils;
 
 import org.apache.flink.table.catalog.Catalog;
 import org.apache.paimon.catalog.CatalogContext;
@@ -30,7 +31,6 @@
 
 import java.lang.reflect.Method;
 import java.util.Map;
-import java.util.stream.Collectors;
 
 import static org.apache.fluss.metadata.DataLakeFormat.ICEBERG;
 import static org.apache.fluss.metadata.DataLakeFormat.PAIMON;
@@ -71,18 +71,9 @@ public Catalog getLakeCatalog(
                                         + ConfigOptions.TABLE_DATALAKE_FORMAT.key()
                                         + "' is set.");
                     }
-                    String dataLakePrefix = lakeFormat.toString() + ".";
                     Map<String, String> catalogProperties =
-                            lakeCatalogProperties.entrySet().stream()
-                                    .filter(entry -> entry.getKey().startsWith(dataLakePrefix))
-                                    .collect(
-                                            Collectors.toMap(
-                                                    entry ->
-                                                            entry.getKey()
-                                                                    .substring(
-                                                                            dataLakePrefix
-                                                                                    .length()),
-                                                    Map.Entry::getValue));
+                            PropertiesUtils.extractAndRemovePrefix(
+                                    lakeCatalogProperties, lakeFormat + ".");
 
                     catalogProperties.putAll(
                             DataLakeUtils.extractLakeCatalogProperties(tableOptions));