From f73253be6622ac7ec19a27620c3121702b55571e Mon Sep 17 00:00:00 2001
From: ouyangwulink <mr.artboy@gmail.com>
Date: Sat, 5 Oct 2024 17:00:20 +0800
Subject: [PATCH] [fixed] fixed envsetting unload flink-conf.yaml,if not like
 catalogstore conf will not work

---
 .../streampark/flink/core/FlinkTableInitializer.scala       | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/scala/org/apache/streampark/flink/core/FlinkTableInitializer.scala b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/scala/org/apache/streampark/flink/core/FlinkTableInitializer.scala
index 7cb463ed75..e4affaab24 100644
--- a/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/scala/org/apache/streampark/flink/core/FlinkTableInitializer.scala
+++ b/streampark-flink/streampark-flink-shims/streampark-flink-shims-base/src/main/scala/org/apache/streampark/flink/core/FlinkTableInitializer.scala
@@ -118,6 +118,12 @@ private[flink] class FlinkTableInitializer(args: Array[String], apiType: ApiType
         }
     }
 
+    parameter.get(KEY_FLINK_CONF(), null) match {
+      case null | "" =>
+        throw new ExceptionInInitializerError(
+          "[StreamPark] Usage:can't find config,please set \"--flink.conf $conf \" in main arguments")
+      case conf => builder.withConfiguration(Configuration.fromMap(PropertiesUtils.fromYamlText(DeflaterUtils.unzipString(conf))) status)
+    }
     val buildWith =
       (parameter.get(KEY_FLINK_TABLE_CATALOG), parameter.get(KEY_FLINK_TABLE_DATABASE))
     buildWith match {