[FLINK-37505][python] Add pyflink YAML based config support

Author: gaborgsomogyi

flink-python/pyflink/common/configuration.py

@@ -69,18 +69,20 @@ def set_string(self, key: str, value: str) -> 'Configuration':
         jars_key = jvm.org.apache.flink.configuration.PipelineOptions.JARS.key()
         classpaths_key = jvm.org.apache.flink.configuration.PipelineOptions.CLASSPATHS.key()
         if key in [jars_key, classpaths_key]:
-            jar_urls = Configuration.parse_jars_value(value, jvm)
+            jar_urls = Configuration.parse_list_value(value)
             add_jars_to_context_class_loader(jar_urls)
         self._j_configuration.setString(key, value)
         return self
 
     @staticmethod
-    def parse_jars_value(value: str, jvm):
+    def parse_list_value(value: str):
+        if not value:
+            return []
         from ruamel.yaml import YAML
         yaml = YAML(typ='safe')
-        jar_urls_list = yaml.load(value)
-        if isinstance(jar_urls_list, list):
-            return jar_urls_list
+        value_list = yaml.load(value)
+        if isinstance(value_list, list):
+            return value_list
         return value.split(";")
 
     def get_integer(self, key: str, default_value: int) -> int:

flink-python/pyflink/common/tests/test_configuration.py

@@ -18,7 +18,6 @@
 from copy import deepcopy
 
 from pyflink.common import Configuration
-from pyflink.java_gateway import get_gateway
 from pyflink.testing.test_case_utils import PyFlinkTestCase
 
 
@@ -165,16 +164,27 @@ def test_hash_equal_str(self):
 
         self.assertEqual(str(conf), "{k1=v1, k2=1}")
 
-    def test_parse_jars_value(self):
-        jvm = get_gateway().jvm
+    def test_parse_list_value(self):
+        # test None
+        value = None
+        expected_result = []
+        result = Configuration.parse_list_value(value)
+        self.assertEqual(result, expected_result)
+
         # test parse YAML list
+        value = "[jar1, jar2, jar3]"
+        expected_result = ['jar1', 'jar2', 'jar3']
+        result = Configuration.parse_list_value(value)
+        self.assertEqual(result, expected_result)
+
+        # test parse multiline YAML list
         value = "- jar1\n- jar2\n- jar3"
         expected_result = ['jar1', 'jar2', 'jar3']
-        result = Configuration.parse_jars_value(value, jvm)
+        result = Configuration.parse_list_value(value)
         self.assertEqual(result, expected_result)
 
         # test parse legacy pattern
         value = "jar1;jar2;jar3"
         expected_result = ['jar1', 'jar2', 'jar3']
-        result = Configuration.parse_jars_value(value, jvm)
+        result = Configuration.parse_list_value(value)
         self.assertEqual(result, expected_result)

flink-python/pyflink/datastream/stream_execution_environment.py

@@ -566,11 +566,10 @@ def add_jars(self, *jars_path: str):
         jars_key = jvm.org.apache.flink.configuration.PipelineOptions.JARS.key()
         env_config = jvm.org.apache.flink.python.util.PythonConfigUtil \
             .getEnvironmentConfig(self._j_stream_execution_environment)
-        old_jar_paths = env_config.getString(jars_key, None)
-        joined_jars_path = ';'.join(jars_path)
-        if old_jar_paths and old_jar_paths.strip():
-            joined_jars_path = ';'.join([old_jar_paths, joined_jars_path])
-        env_config.setString(jars_key, joined_jars_path)
+        old_jars_path = env_config.getString(jars_key, None)
+        old_jars_list = Configuration.parse_list_value(old_jars_path)
+        joined_jars_list = [*old_jars_list, *jars_path]
+        env_config.setString(jars_key, str(joined_jars_list))
 
     def add_classpaths(self, *classpaths: str):
         """
@@ -585,10 +584,9 @@ def add_classpaths(self, *classpaths: str):
         env_config = jvm.org.apache.flink.python.util.PythonConfigUtil \
             .getEnvironmentConfig(self._j_stream_execution_environment)
         old_classpaths = env_config.getString(classpaths_key, None)
-        joined_classpaths = ';'.join(list(classpaths))
-        if old_classpaths and old_classpaths.strip():
-            joined_classpaths = ';'.join([old_classpaths, joined_classpaths])
-        env_config.setString(classpaths_key, joined_classpaths)
+        old_classpaths_list = Configuration.parse_list_value(old_classpaths)
+        joined_classpaths_list = [*old_classpaths_list, *classpaths]
+        env_config.setString(classpaths_key, str(joined_classpaths_list))
 
     def get_default_local_parallelism(self) -> int:
         """

flink-python/pyflink/datastream/tests/test_stream_execution_environment.py

@@ -469,6 +469,40 @@ def check_python_exec(i):
         expected.sort()
         self.assertEqual(expected, result)
 
+    def test_add_jars_basic(self):
+        jvm = get_gateway().jvm
+        jars_key = jvm.org.apache.flink.configuration.PipelineOptions.JARS.key()
+        env_config = jvm.org.apache.flink.python.util.PythonConfigUtil \
+            .getEnvironmentConfig(self.env._j_stream_execution_environment)
+
+        old_jars = env_config.getString(jars_key, None)
+        self.assertIsNone(old_jars)
+
+        self.env.add_jars('file://1.jar')
+        new_jars = env_config.getString(jars_key, None)
+        self.assertEqual(new_jars, '[\'file://1.jar\']')
+
+        self.env.add_jars('file://2.jar', 'file://3.jar')
+        new_jars = env_config.getString(jars_key, None)
+        self.assertEqual(new_jars, '[\'file://1.jar\', \'file://2.jar\', \'file://3.jar\']')
+
+    def test_add_classpaths_basic(self):
+        jvm = get_gateway().jvm
+        classpaths_key = jvm.org.apache.flink.configuration.PipelineOptions.CLASSPATHS.key()
+        env_config = jvm.org.apache.flink.python.util.PythonConfigUtil \
+            .getEnvironmentConfig(self.env._j_stream_execution_environment)
+
+        old_classpaths = env_config.getString(classpaths_key, None)
+        self.assertIsNone(old_classpaths)
+
+        self.env.add_classpaths('file://1.jar')
+        new_classpaths = env_config.getString(classpaths_key, None)
+        self.assertEqual(new_classpaths, '[\'file://1.jar\']')
+
+        self.env.add_classpaths('file://2.jar', 'file://3.jar')
+        new_classpaths = env_config.getString(classpaths_key, None)
+        self.assertEqual(new_classpaths, '[\'file://1.jar\', \'file://2.jar\', \'file://3.jar\']')
+
     @unittest.skip("Disable due to Kafka connector need to release a new version 2.0")
     def test_add_jars(self):
         # find kafka connector jars

flink-python/pyflink/table/table_config.py

@@ -106,7 +106,7 @@ def set(self, key: str, value: str) -> 'TableConfig':
         jars_key = jvm.org.apache.flink.configuration.PipelineOptions.JARS.key()
         classpaths_key = jvm.org.apache.flink.configuration.PipelineOptions.CLASSPATHS.key()
         if key in [jars_key, classpaths_key]:
-            jar_urls = Configuration.parse_jars_value(value, jvm)
+            jar_urls = Configuration.parse_list_value(value)
             add_jars_to_context_class_loader(jar_urls)
         return self
 

flink-python/pyflink/table/table_environment.py

@@ -1486,17 +1486,16 @@ def _add_jars_to_j_env_config(self, config_key):
         if jar_urls:
             jvm = get_gateway().jvm
             jar_urls_list = []
-            parsed_jar_urls = Configuration.parse_jars_value(jar_urls, jvm)
+            parsed_jar_urls = Configuration.parse_list_value(jar_urls)
             url_strings = [
                 jvm.java.net.URL(url).toString() if url else ""
                 for url in parsed_jar_urls
             ]
             self._parse_urls(url_strings, jar_urls_list)
 
             j_configuration = get_j_env_configuration(self._get_j_env())
-            parsed_jar_urls = Configuration.parse_jars_value(
-                j_configuration.getString(config_key, ""),
-                jvm
+            parsed_jar_urls = Configuration.parse_list_value(
+                j_configuration.getString(config_key, "")
             )
             self._parse_urls(parsed_jar_urls, jar_urls_list)