[beam] branch release-2.34.0 updated: [BEAM-12769] Few fixes related to Java Class Lookup based cross-language expansion (#15677)

[ibzib]

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

ibzib pushed a commit to branch release-2.34.0
in repository https://gitbox.apache.org/repos/asf/beam.git


The following commit(s) were added to refs/heads/release-2.34.0 by this push:
     new 71ffc66  [BEAM-12769] Few fixes related to Java Class Lookup based 
cross-language expansion (#15677)
     new a91cc8a  Merge pull request #15687 from chamikaramj/cherry_pick_15677
71ffc66 is described below

commit 71ffc6614500fcd95a30e2df2e8ac7a081b7a5e9
Author: Chamikara Jayalath <chamik...@apache.org>
AuthorDate: Thu Oct 7 16:00:20 2021 -0700

    [BEAM-12769] Few fixes related to Java Class Lookup based cross-language 
expansion (#15677)
    
    * Few fixes related to Java Class Lookup based cross-language expansion
    
    * Fix for JavaExternalTransform
---
 sdks/python/apache_beam/portability/common_urns.py |  4 ++--
 sdks/python/apache_beam/transforms/external.py     | 13 +++++++++----
 2 files changed, 11 insertions(+), 6 deletions(-)

diff --git a/sdks/python/apache_beam/portability/common_urns.py 
b/sdks/python/apache_beam/portability/common_urns.py
index 178f7d0..4e23c4f 100644
--- a/sdks/python/apache_beam/portability/common_urns.py
+++ b/sdks/python/apache_beam/portability/common_urns.py
@@ -29,7 +29,7 @@ from apache_beam.portability.api.beam_runner_api_pb2_urns 
import StandardPTransf
 from apache_beam.portability.api.beam_runner_api_pb2_urns import 
StandardRequirements
 from apache_beam.portability.api.beam_runner_api_pb2_urns import 
StandardResourceHints
 from apache_beam.portability.api.beam_runner_api_pb2_urns import 
StandardSideInputTypes
-from apache_beam.portability.api.external_transforms_pb2 import 
ExpansionMethods
+from apache_beam.portability.api.external_transforms_pb2_urns import 
ExpansionMethods
 from apache_beam.portability.api.metrics_pb2_urns import MonitoringInfo
 from apache_beam.portability.api.metrics_pb2_urns import MonitoringInfoSpecs
 from apache_beam.portability.api.metrics_pb2_urns import MonitoringInfoTypeUrns
@@ -68,4 +68,4 @@ requirements = StandardRequirements.Enum
 
 displayData = StandardDisplayData.DisplayData
 
-java_class_lookup = ExpansionMethods.JAVA_CLASS_LOOKUP
+java_class_lookup = ExpansionMethods.Enum.JAVA_CLASS_LOOKUP
diff --git a/sdks/python/apache_beam/transforms/external.py 
b/sdks/python/apache_beam/transforms/external.py
index 005cb18..ab02ae6 100644
--- a/sdks/python/apache_beam/transforms/external.py
+++ b/sdks/python/apache_beam/transforms/external.py
@@ -287,7 +287,12 @@ class JavaExternalTransform(ptransform.PTransform):
   """
   def __init__(self, class_name, expansion_service=None):
     self._payload_builder = JavaClassLookupPayloadBuilder(class_name)
-    self._expansion_service = None
+    self._expansion_service = expansion_service
+
+    # Beam explicitly looks for following attributes. Hence adding
+    # 'None' values here to prevent '__getattr__' from being called.
+    self.inputs = None
+    self._fn_api_payload = None
 
   def __call__(self, *args, **kwargs):
     self._payload_builder.with_constructor(*args, **kwargs)
@@ -310,8 +315,8 @@ class JavaExternalTransform(ptransform.PTransform):
 
   def expand(self, pcolls):
     return pcolls | ExternalTransform(
-        common_urns.java_class_lookup,
-        self._payload_builder.build(),
+        common_urns.java_class_lookup.urn,
+        self._payload_builder,
         self._expansion_service)
 
 
@@ -384,7 +389,7 @@ class ExternalTransform(ptransform.PTransform):
     """
     expansion_service = expansion_service or DEFAULT_EXPANSION_SERVICE
     if not urn and isinstance(payload, JavaClassLookupPayloadBuilder):
-      urn = common_urns.java_class_lookup
+      urn = common_urns.java_class_lookup.urn
     self._urn = urn
     self._payload = (
         payload.payload() if isinstance(payload, PayloadBuilder) else payload)