[hotfix] Added convert to properties with prefix key utility class. (#74)

Author: zhuangchong

fluss-common/src/main/java/com/alibaba/fluss/utils/PropertiesUtils.java

@@ -0,0 +1,68 @@
+/*
+ * Copyright (c) 2025 Alibaba Group Holding Ltd.
+ *
+ * Licensed 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 com.alibaba.fluss.utils;
+
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/** Utility class for properties related helper functions. */
+public class PropertiesUtils {
+
+    /** Returns the properties as a map copy with a prefix key. */
+    public static <V> Map<String, V> asPrefixedMap(Map<String, V> properties, String prefix) {
+        return properties.entrySet().stream()
+                .collect(Collectors.toMap(e -> prefix + e.getKey(), Map.Entry::getValue));
+    }
+
+    /**
+     * Extracts the properties with the given prefix and removes the prefix from the keys.
+     *
+     * @param originalMap The original map
+     * @param prefix The prefix to filter the keys
+     */
+    public static <V> Map<String, V> extractAndRemovePrefix(
+            Map<String, V> originalMap, String prefix) {
+        return originalMap.entrySet().stream()
+                .filter(entry -> entry.getKey().startsWith(prefix))
+                .collect(
+                        Collectors.toMap(
+                                entry -> entry.getKey().substring(prefix.length()),
+                                Map.Entry::getValue));
+    }
+
+    /**
+     * Extracts the properties with the given prefix.
+     *
+     * @param originalMap The original map
+     * @param prefix The prefix to filter the keys
+     */
+    public static <V> Map<String, V> extractPrefix(Map<String, V> originalMap, String prefix) {
+        return originalMap.entrySet().stream()
+                .filter(entry -> 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()
+                .filter(entry -> !entry.getKey().startsWith(prefix))
+                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+    }
+
+    // Make sure that we cannot instantiate this class
+    private PropertiesUtils() {}
+}

fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/utils/CatalogPropertiesUtils.java

@@ -37,6 +37,7 @@
 import java.util.stream.IntStream;
 import java.util.stream.Stream;
 
+import static com.alibaba.fluss.utils.PropertiesUtils.excludeByPrefix;
 import static org.apache.flink.util.Preconditions.checkNotNull;
 
 /**
@@ -75,13 +76,7 @@ public class CatalogPropertiesUtils {
     private static final Pattern SCHEMA_COLUMN_NAME_SUFFIX = Pattern.compile("\\d+\\.name");
 
     public static Map<String, String> deserializeOptions(Map<String, String> map) {
-        return map.entrySet().stream()
-                .filter(
-                        e -> {
-                            final String key = e.getKey();
-                            return !key.startsWith(SCHEMA + SEPARATOR);
-                        })
-                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+        return excludeByPrefix(map, SCHEMA + SEPARATOR);
     }
 
     public static void deserializeWatermark(Map<String, String> map, Schema.Builder builder) {

fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/utils/DataLakeUtils.java

@@ -20,9 +20,10 @@
 import com.alibaba.fluss.config.Configuration;
 import com.alibaba.fluss.metadata.DataLakeFormat;
 
-import java.util.HashMap;
 import java.util.Map;
 
+import static com.alibaba.fluss.utils.PropertiesUtils.extractAndRemovePrefix;
+
 /** Utility class for accessing data lake related configurations. */
 public class DataLakeUtils {
 
@@ -43,16 +44,7 @@ public static Map<String, String> extractLakeCatalogProperties(Configuration tab
         }
 
         // currently, extract datalake catalog config
-        Map<String, String> datalakeConfig = new HashMap<>();
-        Map<String, String> flussConfig = tableOptions.toMap();
         String dataLakePrefix = "table.datalake." + datalakeFormat + ".";
-        for (Map.Entry<String, String> configEntry : flussConfig.entrySet()) {
-            String configKey = configEntry.getKey();
-            String configValue = configEntry.getValue();
-            if (configKey.startsWith(dataLakePrefix)) {
-                datalakeConfig.put(configKey.substring(dataLakePrefix.length()), configValue);
-            }
-        }
-        return datalakeConfig;
+        return extractAndRemovePrefix(tableOptions.toMap(), dataLakePrefix);
     }
 }

fluss-lakehouse/fluss-lakehouse-cli/src/main/java/com/alibaba/fluss/lakehouse/cli/FlussLakehouseCli.java

@@ -35,6 +35,7 @@
 import java.util.List;
 import java.util.Map;
 
+import static com.alibaba.fluss.utils.PropertiesUtils.extractPrefix;
 import static org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions.CHECKPOINTING_INTERVAL;
 
 /** Cli for Fluss lakehouse integrating. */
@@ -162,13 +163,7 @@ private static Map<String, String> getLakeStorageConfig(Configuration configurat
                             + ConfigOptions.DATALAKE_FORMAT.key());
         }
         String datalakeConfigPrefix = "datalake." + datalakeFormat + ".";
-        Map<String, String> lakeStorageConfig = new HashMap<>();
-        for (Map.Entry<String, String> entry : configuration.toMap().entrySet()) {
-            if (entry.getKey().startsWith(datalakeConfigPrefix)) {
-                lakeStorageConfig.put(entry.getKey(), entry.getValue());
-            }
-        }
-        return lakeStorageConfig;
+        return extractPrefix(configuration.toMap(), datalakeConfigPrefix);
     }
 
     private static String getFlussBootStrapServers(Configuration configuration) {

fluss-lakehouse/fluss-lakehouse-paimon/src/main/java/com/alibaba/fluss/lakehouse/paimon/FlussLakehousePaimon.java

@@ -30,10 +30,11 @@
 import org.apache.flink.streaming.api.datastream.DataStreamSource;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 
-import java.util.HashMap;
 import java.util.Map;
 import java.util.regex.Pattern;
 
+import static com.alibaba.fluss.utils.PropertiesUtils.extractAndRemovePrefix;
+
 /** The entrypoint for fluss tier data to Paimon. */
 public class FlussLakehousePaimon {
 
@@ -52,7 +53,7 @@ public static void main(String[] args) throws Exception {
         String database = paramsMap.get(DATABASE);
 
         // extract fluss config
-        Map<String, String> flussConfigMap = extractConfigStartWith(paramsMap, FLUSS_CONF_PREFIX);
+        Map<String, String> flussConfigMap = extractAndRemovePrefix(paramsMap, FLUSS_CONF_PREFIX);
         // we need to get bootstrap.servers
         String bootstrapServers = paramsMap.get(ConfigOptions.BOOTSTRAP_SERVERS.key());
         if (bootstrapServers == null) {
@@ -61,7 +62,7 @@ public static void main(String[] args) throws Exception {
         flussConfigMap.put(ConfigOptions.BOOTSTRAP_SERVERS.key(), bootstrapServers);
 
         // extract paimon config
-        Map<String, String> paimonConfig = extractConfigStartWith(paramsMap, PAIMON_CONF_PREFIX);
+        Map<String, String> paimonConfig = extractAndRemovePrefix(paramsMap, PAIMON_CONF_PREFIX);
 
         // then build the fluss to paimon job
         final StreamExecutionEnvironment execEnv =
@@ -110,17 +111,4 @@ public boolean test(String database) {
             return databasePattern.matcher(database).matches();
         }
     }
-
-    private static Map<String, String> extractConfigStartWith(
-            Map<String, String> configParams, String prefix) {
-        Map<String, String> extractedConfig = new HashMap<>();
-        for (Map.Entry<String, String> configEntry : configParams.entrySet()) {
-            String configKey = configEntry.getKey();
-            String configValue = configEntry.getValue();
-            if (configKey.startsWith(prefix)) {
-                extractedConfig.put(configKey.substring(prefix.length()), configValue);
-            }
-        }
-        return extractedConfig;
-    }
 }

fluss-server/src/main/java/com/alibaba/fluss/server/utils/LakeStorageUtils.java

@@ -22,10 +22,12 @@
 
 import javax.annotation.Nullable;
 
-import java.util.HashMap;
 import java.util.Map;
 import java.util.Optional;
 
+import static com.alibaba.fluss.utils.PropertiesUtils.asPrefixedMap;
+import static com.alibaba.fluss.utils.PropertiesUtils.extractPrefix;
+
 /** Utils for Fluss lake storage. */
 public class LakeStorageUtils {
 
@@ -36,14 +38,8 @@ public static Map<String, String> generateDefaultTableLakeOptions(Configuration
         if (!optDataLakeFormat.isPresent()) {
             return null;
         }
-        Map<String, String> datalakeProperties = new HashMap<>();
+
         String dataLakePrefix = "datalake." + optDataLakeFormat.get() + ".";
-        for (Map.Entry<String, String> configurationEntry : clusterConf.toMap().entrySet()) {
-            if (configurationEntry.getKey().startsWith(dataLakePrefix)) {
-                datalakeProperties.put(
-                        "table." + configurationEntry.getKey(), configurationEntry.getValue());
-            }
-        }
-        return datalakeProperties;
+        return asPrefixedMap(extractPrefix(clusterConf.toMap(), dataLakePrefix), "table.");
     }
 }