[BEAM-7819] Python - parse PubSub message_id into attributes property

[matt-darwin]

According to the documentation, system generated metadata should be in the attributes property of the PubSub message, however this is not the case for message_id which is a seperate property of the protobuf PubSub message.

Parse this and add to the attributes dictionary of the beam PubSub message.

Amend tests to expect message_id k,v to be in the attributes property.

---

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

• [X ] Choose reviewer(s) and mention them in a comment (R: @username).
• [ X] Format the pull request title like [BEAM-XXX] Fixes bug in ApproximateQuantiles, where you replace BEAM-XXX with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
• [ X] If this contribution is large, please file an Apache Individual Contributor License Agreement.

Post-Commit Tests Status (on master branch)

Lang	SDK	Apex	Dataflow	Flink	Gearpump	Samza	Spark
Go		---	---		---	---
Java
Python		---			---	---

Pre-Commit Tests Status (on master branch)

---	Java	Python	Go	Website
Non-portable
Portable	---		---	---

See .test-infra/jenkins/README for trigger phrase, status and link of all Jenkins jobs.

[matt-darwin]

@aaltay @udim

[aaltay]

When you are ready, let's also run a test on Dataflow. Since Dataflow and Direct runner has different pubsub implementations.

[lukecwik]

FYI, there is #8370 which is attempting to add the message id to the Beam Java SDK.

[udim]

Note that in Dataflow, the Python SDK uses FnAPI to read from Pub/Sub using a Java harness. In other words, the message id might be missing because Java code doesn't provide it somehow. I'll look into it later.

When running locally using DirectRunner, there's a different implementation that replaces ReadFromPubSub entirely, which probably needs modification as well. See:

beam/sdks/python/apache_beam/runners/direct/transform_evaluator.py

Line 380 in 0226799

class _PubSubReadEvaluator(_TransformEvaluator):

[matt-darwin]

FYI, there is #8370 which is attempting to add the message id to the Beam Java SDK.

I'll have a look at building the runner from this change to see if I can then access the message_id from the dataflow runner. I'm not adept at all at Java however, so wouldn't be much help there.

[matt-darwin]

Note that in Dataflow, the Python SDK uses FnAPI to read from Pub/Sub using a Java harness. In other words, the message id might be missing because Java code doesn't provide it somehow. I'll look into it later.

When running locally using DirectRunner, there's a different implementation that replaces ReadFromPubSub entirely, which probably needs modification as well. See:

beam/sdks/python/apache_beam/runners/direct/transform_evaluator.py

Line 380 in 0226799

class _PubSubReadEvaluator(_TransformEvaluator):

The directrunner has been working ok with the above changes; I think the issue is on the dataflow runner side. The directrunner is using the _from_message method, and this is parsing correctly and returning the pubsub message id in my testing so far.

def _read_from_pubsub(self, timestamp_attribute):
   from apache_beam.io.gcp.pubsub import PubsubMessage
   from google.cloud import pubsub

   def _get_element(message):
     parsed_message = PubsubMessage._from_message(message)

[matt-darwin]

Run Python Dataflow ValidatesRunner

[matt-darwin]

Run Python Dataflow ValidatesRunner

[matt-darwin]

Need to fix up some tests, and the PubSubMessage tests need amending to be in line with those. Haven't had chance to test with the dataflowrunner as yet.

[matt-darwin]

Ok, so the PubSub timestamp is actually the google.protobuf.timestamp_pb2.Timestamp type. Have amended everything to expect this for the publish_time attribute

[udim]

FYI, support for message_id and publish_time in Dataflow should be available in a future update of Dataflow. Until then, those fields will appear unset or blank.

[matt-darwin]

FYI, support for message_id and publish_time in Dataflow should be available in a future update of Dataflow. Until then, those fields will appear unset or blank.

Is that external to the Beam project?

[matt-darwin]

Run Python Dataflow ValidatesRunner

[udim]

Is that external to the Beam project?

Yes

[matt-darwin]

Run Python Dataflow ValidatesRunner

[matt-darwin]

Back to the isort error now:

Running isort...
--
ERROR: /home/jenkins/jenkins-slave/workspace/beam_PreCommit_Python_Commit/src/sdks/python/test-suites/tox/py2/build/srcs/sdks/python/apache_beam/io/gcp/pubsub_test.py Imports are incorrectly sorted.
--- /home/jenkins/jenkins-slave/workspace/beam_PreCommit_Python_Commit/src/sdks/python/test-suites/tox/py2/build/srcs/sdks/python/apache_beam/io/gcp/pubsub_test.py:before	2019-08-17 06:59:14.752504
+++ /home/jenkins/jenkins-slave/workspace/beam_PreCommit_Python_Commit/src/sdks/python/test-suites/tox/py2/build/srcs/sdks/python/apache_beam/io/gcp/pubsub_test.py:after	2019-08-17 07:01:17.704126
@@ -26,6 +26,7 @@
 
import hamcrest as hc
import mock
+from google.protobuf.timestamp_pb2 import Timestamp
 
import apache_beam as beam
from apache_beam.io.gcp.pubsub import PubsubMessage
@@ -52,7 +53,6 @@
from apache_beam.transforms.display_test import DisplayDataItemMatcher
from apache_beam.utils import timestamp
 
-from google.protobuf.timestamp_pb2 import Timestamp
# Protect against environments where the PubSub library is not available.
try:
from google.cloud import pubsub
Command exited with non-zero status 1

@aaltay are you happy for the file to be excluded? If I make the isort recommended change, it then errors due to grouping on the google side.

The issue is if I leave the import in the try block, I get None exceptions all over the place for the timestamp object. Or is this simply because the try block for imports is triggered by a requirement to import the first module in the block and I have them in the wrong order?

[matt-darwin]

Run Python Dataflow ValidatesRunner

[aaltay]

@matt-darwin - Re: isort - Sure, it is fine to exclude it from isort. Thank you for the explanation.

[matt-darwin]

@aaltay and @udim was there anything further you wanted changing on this?

[matt-darwin]

Run Python Dataflow ValidatesRunner

[aaltay]

LGTM. Thank you @matt-darwin.

We can merge it once the other reviewers also complete their reviews.

[aaltay]

@udim do you have additional comments? Can we merge this?

[udim]

One comment in run_pylint.sh, otherwise LGTM

[aaltay]

The last comment was also addressed. Merging this.

[udim]

Post-commits are failing, opened: https://issues.apache.org/jira/browse/BEAM-8153

[aaltay]

Thank you @udim. Sent you #9479 to revert this PR. We can revert first than address the test issue.