You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by fa...@apache.org on 2024/09/24 10:42:43 UTC
(flink) 02/02: [FLINK-34082][config] Adapt the python configuration
This is an automated email from the ASF dual-hosted git repository.
fanrui pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit c9213b3bf7513114b3f676cc6a3ae49e4f72789e
Author: Rui Fan <19...@gmail.com>
AuthorDate: Sat Sep 14 13:58:49 2024 +0800
[FLINK-34082][config] Adapt the python configuration
---
flink-python/pyflink/common/configuration.py | 30 ++++++++++++++++-----
.../datastream/connectors/tests/test_pulsar.py | 31 +++++++++++++---------
flink-python/pyflink/datastream/data_stream.py | 6 ++---
3 files changed, 45 insertions(+), 22 deletions(-)
diff --git a/flink-python/pyflink/common/configuration.py b/flink-python/pyflink/common/configuration.py
index d833f83ba02..d09a2eb0899 100644
--- a/flink-python/pyflink/common/configuration.py
+++ b/flink-python/pyflink/common/configuration.py
@@ -92,7 +92,10 @@ class Configuration:
associated with the given key.
:return: The (default) value associated with the given key.
"""
- return self._j_configuration.getLong(key, default_value)
+ gateway = get_gateway()
+ JConfigOptions = gateway.jvm.org.apache.flink.configuration.ConfigOptions
+ config_option = JConfigOptions.key(key).longType().noDefaultValue()
+ return self._j_configuration.get(config_option, default_value)
def set_integer(self, key: str, value: int) -> 'Configuration':
"""
@@ -101,7 +104,10 @@ class Configuration:
:param key: The key of the key/value pair to be added.
:param value: The value of the key/value pair to be added.
"""
- self._j_configuration.setLong(key, value)
+ gateway = get_gateway()
+ JConfigOptions = gateway.jvm.org.apache.flink.configuration.ConfigOptions
+ config_option = JConfigOptions.key(key).longType().noDefaultValue()
+ self._j_configuration.set(config_option, value)
return self
def get_boolean(self, key: str, default_value: bool) -> bool:
@@ -113,7 +119,10 @@ class Configuration:
associated with the given key.
:return: The (default) value associated with the given key.
"""
- return self._j_configuration.getBoolean(key, default_value)
+ gateway = get_gateway()
+ JConfigOptions = gateway.jvm.org.apache.flink.configuration.ConfigOptions
+ config_option = JConfigOptions.key(key).booleanType().noDefaultValue()
+ return self._j_configuration.get(config_option, default_value)
def set_boolean(self, key: str, value: bool) -> 'Configuration':
"""
@@ -122,7 +131,10 @@ class Configuration:
:param key: The key of the key/value pair to be added.
:param value: The value of the key/value pair to be added.
"""
- self._j_configuration.setBoolean(key, value)
+ gateway = get_gateway()
+ JConfigOptions = gateway.jvm.org.apache.flink.configuration.ConfigOptions
+ config_option = JConfigOptions.key(key).booleanType().noDefaultValue()
+ self._j_configuration.set(config_option, value)
return self
def get_float(self, key: str, default_value: float) -> float:
@@ -134,7 +146,10 @@ class Configuration:
associated with the given key.
:return: The (default) value associated with the given key.
"""
- return self._j_configuration.getDouble(key, float(default_value))
+ gateway = get_gateway()
+ JConfigOptions = gateway.jvm.org.apache.flink.configuration.ConfigOptions
+ config_option = JConfigOptions.key(key).doubleType().noDefaultValue()
+ return self._j_configuration.get(config_option, float(default_value))
def set_float(self, key: str, value: float) -> 'Configuration':
"""
@@ -143,7 +158,10 @@ class Configuration:
:param key: The key of the key/value pair to be added.
:param value: The value of the key/value pair to be added.
"""
- self._j_configuration.setDouble(key, float(value))
+ gateway = get_gateway()
+ JConfigOptions = gateway.jvm.org.apache.flink.configuration.ConfigOptions
+ config_option = JConfigOptions.key(key).doubleType().noDefaultValue()
+ self._j_configuration.set(config_option, float(value))
return self
def get_bytearray(self, key: str, default_value: bytearray) -> bytearray:
diff --git a/flink-python/pyflink/datastream/connectors/tests/test_pulsar.py b/flink-python/pyflink/datastream/connectors/tests/test_pulsar.py
index 55f4659c28f..7e547cc3da1 100644
--- a/flink-python/pyflink/datastream/connectors/tests/test_pulsar.py
+++ b/flink-python/pyflink/datastream/connectors/tests/test_pulsar.py
@@ -15,6 +15,8 @@
# See the License for the specific language governing permissions and
# limitations under the License.
################################################################################
+import unittest
+
from pyflink.common import WatermarkStrategy, SimpleStringSchema, Types, ConfigOptions, Duration
from pyflink.datastream.connectors import DeliveryGuarantee
from pyflink.datastream.connectors.pulsar import TopicRoutingMode, MessageDelayer, PulsarSink, \
@@ -23,6 +25,9 @@ from pyflink.testing.test_case_utils import PyFlinkUTTestCase
from pyflink.util.java_utils import get_field_value, is_instance_of
+@unittest.skip("Disabled due to pulsar is still using the deprecated api, and it should be removed "
+ "after flink-sql-connector-pulsar 4.2.0 is released. "
+ "Get more details from https://github.com/apache/flink-connector-pulsar/pull/96")
class FlinkPulsarTest(PyFlinkUTTestCase):
def test_pulsar_source(self):
@@ -50,26 +55,26 @@ class FlinkPulsarTest(PyFlinkUTTestCase):
configuration = get_field_value(pulsar_source.get_java_function(), "sourceConfiguration")
self.assertEqual(
- configuration.getString(
+ configuration.get(
ConfigOptions.key('pulsar.client.serviceUrl')
.string_type()
.no_default_value()._j_config_option), 'pulsar://localhost:6650')
self.assertEqual(
- configuration.getString(
+ configuration.get(
ConfigOptions.key('pulsar.admin.adminUrl')
.string_type()
.no_default_value()._j_config_option), 'http://localhost:8080')
self.assertEqual(
- configuration.getString(
+ configuration.get(
ConfigOptions.key('pulsar.consumer.subscriptionName')
.string_type()
.no_default_value()._j_config_option), 'ff')
test_option = ConfigOptions.key(TEST_OPTION_NAME).boolean_type().no_default_value()
self.assertEqual(
- configuration.getBoolean(
+ configuration.get(
test_option._j_config_option), True)
self.assertEqual(
- configuration.getLong(
+ configuration.get(
ConfigOptions.key('pulsar.source.autoCommitCursorInterval')
.long_type()
.no_default_value()._j_config_option), 1000)
@@ -107,10 +112,10 @@ class FlinkPulsarTest(PyFlinkUTTestCase):
.build()
configuration = get_field_value(pulsar_source.get_java_function(), "sourceConfiguration")
self.assertEqual(
- configuration.getBoolean(
+ configuration.get(
test_option._j_config_option), True)
self.assertEqual(
- configuration.getLong(
+ configuration.get(
ConfigOptions.key('pulsar.source.autoCommitCursorInterval')
.long_type()
.no_default_value()._j_config_option), 1000)
@@ -200,17 +205,17 @@ class FlinkPulsarTest(PyFlinkUTTestCase):
self.assertEqual('pulsar sink: Writer', plan['nodes'][1]['type'])
configuration = get_field_value(pulsar_sink.get_java_function(), "sinkConfiguration")
self.assertEqual(
- configuration.getString(
+ configuration.get(
ConfigOptions.key('pulsar.client.serviceUrl')
.string_type()
.no_default_value()._j_config_option), 'pulsar://localhost:6650')
self.assertEqual(
- configuration.getString(
+ configuration.get(
ConfigOptions.key('pulsar.admin.adminUrl')
.string_type()
.no_default_value()._j_config_option), 'http://localhost:8080')
self.assertEqual(
- configuration.getString(
+ configuration.get(
ConfigOptions.key('pulsar.producer.producerName')
.string_type()
.no_default_value()._j_config_option), 'fo - %s')
@@ -225,7 +230,7 @@ class FlinkPulsarTest(PyFlinkUTTestCase):
'org.apache.flink.api.common.serialization.SimpleStringSchema'))
self.assertEqual(
- configuration.getString(
+ configuration.get(
ConfigOptions.key('pulsar.sink.deliveryGuarantee')
.string_type()
.no_default_value()._j_config_option), 'at-least-once')
@@ -242,10 +247,10 @@ class FlinkPulsarTest(PyFlinkUTTestCase):
test_option = ConfigOptions.key(TEST_OPTION_NAME).boolean_type().no_default_value()
self.assertEqual(
- configuration.getBoolean(
+ configuration.get(
test_option._j_config_option), True)
self.assertEqual(
- configuration.getLong(
+ configuration.get(
ConfigOptions.key('pulsar.producer.batchingMaxMessages')
.long_type()
.no_default_value()._j_config_option), 100)
diff --git a/flink-python/pyflink/datastream/data_stream.py b/flink-python/pyflink/datastream/data_stream.py
index 78e53d367a3..b410c78fdcc 100644
--- a/flink-python/pyflink/datastream/data_stream.py
+++ b/flink-python/pyflink/datastream/data_stream.py
@@ -1218,7 +1218,7 @@ class KeyedStream(DataStream):
gateway = get_gateway()
j_conf = get_j_env_configuration(self._j_data_stream.getExecutionEnvironment())
python_execution_mode = (
- j_conf.getString(
+ j_conf.get(
gateway.jvm.org.apache.flink.python.PythonOptions.PYTHON_EXECUTION_MODE))
class ReduceProcessKeyedProcessFunctionAdapter(KeyedProcessFunction):
@@ -2783,7 +2783,7 @@ def _get_one_input_stream_operator(data_stream: DataStream,
j_output_type_info = output_type_info.get_java_type_info()
j_conf = get_j_env_configuration(data_stream._j_data_stream.getExecutionEnvironment())
python_execution_mode = (
- j_conf.getString(gateway.jvm.org.apache.flink.python.PythonOptions.PYTHON_EXECUTION_MODE))
+ j_conf.get(gateway.jvm.org.apache.flink.python.PythonOptions.PYTHON_EXECUTION_MODE))
from pyflink.fn_execution.flink_fn_execution_pb2 import UserDefinedDataStreamFunction
if func_type == UserDefinedDataStreamFunction.PROCESS: # type: ignore
@@ -2864,7 +2864,7 @@ def _get_two_input_stream_operator(connected_streams: ConnectedStreams,
j_conf = get_j_env_configuration(
connected_streams.stream1._j_data_stream.getExecutionEnvironment())
python_execution_mode = (
- j_conf.getString(gateway.jvm.org.apache.flink.python.PythonOptions.PYTHON_EXECUTION_MODE))
+ j_conf.get(gateway.jvm.org.apache.flink.python.PythonOptions.PYTHON_EXECUTION_MODE))
from pyflink.fn_execution.flink_fn_execution_pb2 import UserDefinedDataStreamFunction
if func_type == UserDefinedDataStreamFunction.CO_PROCESS: # type: ignore