[beam] branch master updated: Fix py3 type error in bundle_processor

[ro...@apache.org]

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

robertwb pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git


The following commit(s) were added to refs/heads/master by this push:
     new a519547  Fix py3 type error in bundle_processor
     new 07c014e  Merge pull request #7521 [BEAM-5953] Fix py3 type error in bundle_processor
a519547 is described below

commit a51954785640140262e097a1bdab8b2aa8ca1545
Author: Mark Liu <ma...@google.com>
AuthorDate: Tue Jan 15 16:03:09 2019 -0800

    Fix py3 type error in bundle_processor
---
 sdks/python/apache_beam/runners/worker/bundle_processor.py | 2 +-
 sdks/python/apache_beam/runners/worker/operation_specs.py  | 3 ++-
 2 files changed, 3 insertions(+), 2 deletions(-)

diff --git a/sdks/python/apache_beam/runners/worker/bundle_processor.py b/sdks/python/apache_beam/runners/worker/bundle_processor.py
index 679b8ea..fc8b128 100644
--- a/sdks/python/apache_beam/runners/worker/bundle_processor.py
+++ b/sdks/python/apache_beam/runners/worker/bundle_processor.py
@@ -592,7 +592,7 @@ class BeamTransformFactory(object):
     else:
       # No URN, assume cloud object encoding json bytes.
       return operation_specs.get_coder_from_spec(
-          json.loads(coder_proto.spec.spec.payload))
+          json.loads(coder_proto.spec.spec.payload.decode('utf-8')))
 
   def get_windowed_coder(self, pcoll_id):
     coder = self.get_coder(self.descriptor.pcollections[pcoll_id].coder_id)
diff --git a/sdks/python/apache_beam/runners/worker/operation_specs.py b/sdks/python/apache_beam/runners/worker/operation_specs.py
index ebae476..464558d 100644
--- a/sdks/python/apache_beam/runners/worker/operation_specs.py
+++ b/sdks/python/apache_beam/runners/worker/operation_specs.py
@@ -354,7 +354,8 @@ def get_coder_from_spec(coder_spec):
 
   # We pass coders in the form "<coder_name>$<pickled_data>" to make the job
   # description JSON more readable.
-  return coders.coders.deserialize_coder(coder_spec['@type'])
+  return coders.coders.deserialize_coder(
+      coder_spec['@type'].encode('ascii'))
 
 
 class MapTask(object):