[flink] branch release-1.15 updated: [hotfix][python][connector/pulsar] Improve PulsarDeserializationSchema.flink_type_info set execution_config default to None

[di...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

dianfu pushed a commit to branch release-1.15
in repository https://gitbox.apache.org/repos/asf/flink.git


The following commit(s) were added to refs/heads/release-1.15 by this push:
     new 73590f432d8 [hotfix][python][connector/pulsar] Improve PulsarDeserializationSchema.flink_type_info set execution_config default to None
73590f432d8 is described below

commit 73590f432d8dc4cd53f0357025e250a823e6b857
Author: Dian Fu <di...@apache.org>
AuthorDate: Sat May 21 17:49:33 2022 +0800

    [hotfix][python][connector/pulsar] Improve PulsarDeserializationSchema.flink_type_info set execution_config default to None
---
 flink-python/pyflink/datastream/connectors.py            | 7 ++++---
 flink-python/pyflink/datastream/tests/test_connectors.py | 2 +-
 2 files changed, 5 insertions(+), 4 deletions(-)

diff --git a/flink-python/pyflink/datastream/connectors.py b/flink-python/pyflink/datastream/connectors.py
index dfac9967e72..f207939753c 100644
--- a/flink-python/pyflink/datastream/connectors.py
+++ b/flink-python/pyflink/datastream/connectors.py
@@ -1108,16 +1108,17 @@ class PulsarDeserializationSchema(object):
         return PulsarDeserializationSchema(_j_pulsar_deserialization_schema)
 
     @staticmethod
-    def flink_type_info(type_information: TypeInformation, execution_config: ExecutionConfig) \
-            -> 'PulsarDeserializationSchema':
+    def flink_type_info(type_information: TypeInformation,
+                        execution_config: ExecutionConfig = None) -> 'PulsarDeserializationSchema':
         """
         Create a PulsarDeserializationSchema by using the given TypeInformation. This method is
         only used for treating message that was written into pulsar by TypeInformation.
         """
         JPulsarDeserializationSchema = get_gateway().jvm.org.apache.flink \
             .connector.pulsar.source.reader.deserializer.PulsarDeserializationSchema
+        JExecutionConfig = get_gateway().jvm.org.apache.flink.api.common.ExecutionConfig
         _j_execution_config = execution_config._j_execution_config \
-            if execution_config is not None else None
+            if execution_config is not None else JExecutionConfig()
         _j_pulsar_deserialization_schema = JPulsarDeserializationSchema.flinkTypeInfo(
             type_information.get_java_type_info(), _j_execution_config)
         return PulsarDeserializationSchema(_j_pulsar_deserialization_schema)
diff --git a/flink-python/pyflink/datastream/tests/test_connectors.py b/flink-python/pyflink/datastream/tests/test_connectors.py
index 407af8e8d04..393801a8d5d 100644
--- a/flink-python/pyflink/datastream/tests/test_connectors.py
+++ b/flink-python/pyflink/datastream/tests/test_connectors.py
@@ -183,7 +183,7 @@ class FlinkPulsarTest(ConnectorTestBase):
             .set_subscription_name('ff') \
             .set_subscription_type(SubscriptionType.Exclusive) \
             .set_deserialization_schema(
-                PulsarDeserializationSchema.flink_type_info(Types.STRING(), None)) \
+                PulsarDeserializationSchema.flink_type_info(Types.STRING())) \
             .set_deserialization_schema(
                 PulsarDeserializationSchema.flink_schema(SimpleStringSchema())) \
             .set_config(test_option, True) \