[hotfix] Added convert to properties With prefix key utility class.

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

@@ -0,0 +1,31 @@
+/*
+ * Copyright (c) 2024 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 Options related helper functions. */
+public class OptionsUtils {
+
+    public static Map<String, String> convertToPropertiesWithPrefixKey(
+            Map<String, String> configParams, final String prefixKey) {
+        return configParams.keySet().stream()
+                .filter(k -> k.startsWith(prefixKey))
+                .collect(Collectors.toMap(k -> k.substring(prefixKey.length()), configParams::get));
+    }
+}

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.OptionsUtils.convertToPropertiesWithPrefixKey;
+
 /** The entrypoint for fluss tier data to Paimon. */
 public class FlussLakehousePaimon {
 
@@ -52,7 +53,8 @@ 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 =
+                convertToPropertiesWithPrefixKey(paramsMap, FLUSS_CONF_PREFIX);
         // we need to get bootstrap.servers
         String bootstrapServers = paramsMap.get(ConfigOptions.BOOTSTRAP_SERVERS.key());
         if (bootstrapServers == null) {
@@ -62,7 +64,7 @@ public static void main(String[] args) throws Exception {
 
         // extract paimon config
         Map<String, String> paimonConfig =
-                extractConfigStartWith(paramsMap, PAIMON_CATALOG_CONF_PREFIX);
+                convertToPropertiesWithPrefixKey(paramsMap, PAIMON_CATALOG_CONF_PREFIX);
 
         // then build the fluss to paimon job
         final StreamExecutionEnvironment execEnv =
@@ -111,17 +113,4 @@ public boolean test(String database) {
             return databasePattern.matcher(database).matches();
         }
     }
-
-    public 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

@@ -20,9 +20,10 @@
 import com.alibaba.fluss.config.Configuration;
 import com.alibaba.fluss.lakehouse.LakeStorageInfo;
 
-import java.util.HashMap;
 import java.util.Map;
 
+import static com.alibaba.fluss.utils.OptionsUtils.convertToPropertiesWithPrefixKey;
+
 /** Utils for Fluss lake storage. */
 public class LakeStorageUtils {
 
@@ -49,15 +50,8 @@ public static LakeStorageInfo getLakeStorageInfo(Configuration configuration) {
 
         // currently, extract catalog config
         String catalogPrefix = datalakeStorage + "." + CATALOG_PREFIX;
-        Map<String, String> catalogConfig = new HashMap<>();
-        Map<String, String> flussConfig = configuration.toMap();
-        for (Map.Entry<String, String> configEntry : flussConfig.entrySet()) {
-            String configKey = configEntry.getKey();
-            String configValue = configEntry.getValue();
-            if (configKey.startsWith(catalogPrefix)) {
-                catalogConfig.put(configKey.substring(catalogPrefix.length()), configValue);
-            }
-        }
+        Map<String, String> catalogConfig =
+                convertToPropertiesWithPrefixKey(configuration.toMap(), catalogPrefix);
         return new LakeStorageInfo(datalakeStorage, catalogConfig);
     }
 }