[Improve] load flink-config minor improvements

apache · Oct 7, 2024 · bbe7731 · bbe7731
1 parent dcef164
commit bbe7731
Show file tree

Hide file tree

Showing 3 changed files with 3 additions and 23 deletions.
diff --git a/...lient-core/src/main/scala/org/apache/streampark/flink/client/trait/FlinkClientTrait.scala b/...lient-core/src/main/scala/org/apache/streampark/flink/client/trait/FlinkClientTrait.scala
@@ -77,7 +77,7 @@ trait FlinkClientTrait extends Logger {
          |""".stripMargin)
 
     val flinkConfig = prepareConfig(submitRequest)
-
+    flinkConfig.toMap.foreach(c => logInfo(s"flinkConfig:  ${c._1}: ${c._2}"))
     setConfig(submitRequest, flinkConfig)
 
     Try(doSubmit(submitRequest, flinkConfig)) match {

diff --git a/...hims-base/src/main/scala/org/apache/streampark/flink/core/FlinkStreamingInitializer.scala b/...hims-base/src/main/scala/org/apache/streampark/flink/core/FlinkStreamingInitializer.scala
@@ -95,17 +95,7 @@ private[flink] class FlinkStreamingInitializer(args: Array[String], apiType: Api
       .mergeWith(ParameterTool.fromMap(appConf))
       .mergeWith(argsMap)
 
-    val flinkConf: Map[String, String] = {
-      parameter.get(KEY_FLINK_CONF(), null) match {
-        case flinkConf if flinkConf != null =>
-          PropertiesUtils
-            .loadFlinkConfYaml(DeflaterUtils.unzipString(flinkConf))
-            .filter(_._2.nonEmpty)
-        case _ => Map.empty
-      }
-    }
-
-    val envConfig = Configuration.fromMap(flinkConf ++ appFlinkConf)
+    val envConfig = Configuration.fromMap(appFlinkConf)
     FlinkConfiguration(parameter, envConfig, null)
   }
 

diff --git a/...nk-shims-base/src/main/scala/org/apache/streampark/flink/core/FlinkTableInitializer.scala b/...nk-shims-base/src/main/scala/org/apache/streampark/flink/core/FlinkTableInitializer.scala
@@ -187,17 +187,7 @@ private[flink] class FlinkTableInitializer(args: Array[String], apiType: ApiType
         val tableConf = extractConfigByPrefix(configMap, KEY_FLINK_TABLE_PREFIX)
         val sqlConf = extractConfigByPrefix(configMap, KEY_SQL_PREFIX)
 
-        val flinkConf: Map[String, String] = {
-          parameter.get(KEY_FLINK_CONF(), null) match {
-            case flinkConf if flinkConf != null =>
-              PropertiesUtils
-                .loadFlinkConfYaml(DeflaterUtils.unzipString(flinkConf))
-                .filter(_._2.nonEmpty)
-            case _ => Map.empty
-          }
-        }
-
-        val envConfig = Configuration.fromMap(flinkConf ++ appFlinkConf)
+        val envConfig = Configuration.fromMap(appFlinkConf)
         val tableConfig = Configuration.fromMap(tableConf)
 
         val parameterTool = ParameterTool