[hotfix] Pass datalake related args run paimon tiering service (#560)

Author: luoyuxia

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

@@ -75,15 +75,8 @@ public static int mainInternal(String[] args) {
                     GlobalConfiguration.loadConfiguration(
                             configDirAndDynamicConfig.f0, configDirAndDynamicConfig.f1);
 
-            // validate datalake format
-            DataLakeFormat datalakeFormat = globalConfiguration.get(ConfigOptions.DATALAKE_FORMAT);
-            if (datalakeFormat == null) {
-                throw new IllegalArgumentException(
-                        "The datalake format is not set,"
-                                + " please set the configuration "
-                                + ConfigOptions.DATALAKE_FORMAT.key());
-            }
-
+            // get config for lake storage
+            Map<String, String> lakeStorageConfig = getLakeStorageConfig(globalConfiguration);
             // get config for fluss source, now only bootstrap server is required
             String flussBootstrapServer = getFlussBootStrapServers(globalConfiguration);
 
@@ -100,7 +93,14 @@ public static int mainInternal(String[] args) {
             flinkConfig.set(PipelineOptions.NAME, DEFAULT_PIPELINE_NAME);
 
             // now, start execute
-            retCode = run(runAction, jarFile, flussBootstrapServer, flinkConfig, dynamicConfigs);
+            retCode =
+                    run(
+                            runAction,
+                            jarFile,
+                            flussBootstrapServer,
+                            lakeStorageConfig,
+                            flinkConfig,
+                            dynamicConfigs);
         } catch (Throwable t) {
             t.printStackTrace(System.err);
         }
@@ -125,6 +125,7 @@ private static int run(
             String runAction,
             String jarFile,
             String flussBootstrapServer,
+            Map<String, String> lakeStorageConfig,
             org.apache.flink.configuration.Configuration flinkConfig,
             Map<String, String> dynamicConfigs) {
         List<CustomCommandLine> customCommandLines = new ArrayList<>();
@@ -139,6 +140,10 @@ private static int run(
                                 jarFile,
                                 "--" + ConfigOptions.BOOTSTRAP_SERVERS.key(),
                                 flussBootstrapServer));
+        for (Map.Entry<String, String> entry : lakeStorageConfig.entrySet()) {
+            arguments.add("--" + entry.getKey());
+            arguments.add(entry.getValue());
+        }
         for (Map.Entry<String, String> entry : dynamicConfigs.entrySet()) {
             arguments.add("--" + entry.getKey());
             arguments.add(entry.getValue());
@@ -147,6 +152,25 @@ private static int run(
         return cliFrontend.parseAndRun(newArgs);
     }
 
+    private static Map<String, String> getLakeStorageConfig(Configuration configuration) {
+        // validate datalake format
+        DataLakeFormat datalakeFormat = configuration.get(ConfigOptions.DATALAKE_FORMAT);
+        if (datalakeFormat == null) {
+            throw new IllegalArgumentException(
+                    "The datalake format is not set,"
+                            + " please set the configuration "
+                            + 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;
+    }
+
     private static String getFlussBootStrapServers(Configuration configuration) {
         Map<String, String> configMap = configuration.toMap();
         String bootstrapServers = configMap.get(ConfigOptions.BOOTSTRAP_SERVERS.key());