[BEAM-3489] add PubSub messageId in PubsubMessage #8370

thinhha · 2019-04-21T15:11:02Z

Make the recordId field from PubsubClient.IncomingMessage available as a new field called messageId in PubsubMessage.
Updated the coder for PubsubMessage to encode and decode the messageId.

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

Choose reviewer(s) and mention them in a comment (R: @username).
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.
If this contribution is large, please file an Apache Individual Contributor License Agreement.

Post-Commit Tests Status (on master branch)

Lang	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.

thinhha · 2019-04-21T15:16:25Z

R: @lukecwik

thinhha · 2019-04-21T20:18:52Z

retest this please

lukecwik

Changing the encoding of the coder will break pipeline update since the encoding used before is not compatible with the new encoding. You'll want to:

add a new coder that is like PubsubMessageWithAttributesCoder
add a method like readMessagesWithAttributes that uses the coders that you created in 1.

Repeat steps 1 and 2 for readMessages if you also want to create a readMessagesWithMessageId

You'll also want to update the implementation in the Dataflow specific PubsubReader here when constructing the PubsubMessage to pass in the message id as well:

beam/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/PubsubReader.java

Line 124 in fff46e1

new PubsubMessage(

thinhha · 2019-05-05T10:24:15Z

Thanks - I've added a few changes based on your comment. It's my first PR so sorry for all the questions!

For PubsubMesageWithMessageIdCoder, do I need to explicitly skip the attributes map which is second in pubsub.proto before getting to the message_id (which is third)?
Where would be a good place to add some tests for this PR?

lukecwik · 2019-05-06T21:36:10Z

.../io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubMessage.java

  public PubsubMessage(byte[] payload, Map<String, String> attributes) {
    this.message = payload;
    this.attributes = attributes;
+    this.messageId = "";


I think you should set this to null.

lukecwik

Glad to help with the questions.

Your current implementation for PubsubMesageWithMessageIdCoder is correct. You don't need to encode/decode an empty map as there is no point.

Tests would typically go here:
https://github.com/apache/beam/tree/master/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/pubsub

You could add an integration test that shows that reading messages with a message id works here:
https://github.com/apache/beam/blob/master/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubReadIT.java
You could also add unit tests for your coders similar to
https://github.com/apache/beam/blob/master/sdks/java/core/src/test/java/org/apache/beam/sdk/coders/MapCoderTest.java
Since your coder is not saying its deterministic or consistent with equals, you can ignore testing for those. Feel free to add additional unit tests for the other pubsub coders that are missing if you want.

lukecwik · 2019-05-06T21:37:00Z

...ain/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubMessageWithAttributesAndMessageIdCoder.java

+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.values.TypeDescriptor;
+
+/** A coder for PubsubMessage including attributes. */


please update comment to state that you also include the message id.

lukecwik · 2019-05-06T21:37:45Z

...latform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubMessageWithMessageIdCoder.java

+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.vendor.guava.v20_0.com.google.common.collect.ImmutableMap;
+
+/** A coder for PubsubMessage treating the raw bytes being decoded as the message's payload. */


Please update this comment as well that it contains the message id.

lukecwik · 2019-05-06T21:38:15Z

.../io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubMessage.java

- * Class representing a Pub/Sub message. Each message contains a single message payload and a map of
- * attached attributes.
+ * Class representing a Pub/Sub message. Each message contains a single message payload, a map of
+ * attached attributes, and an optional messageId.


Suggested change

* attached attributes, and an optional messageId.

* attached attributes, and a message id.

lukecwik · 2019-05-06T21:54:06Z

Looking good so far, minor comment updates and some tests and this will be merged.

lukecwik · 2019-05-21T22:21:34Z

@thinhha any updates?

thinhha · 2019-05-26T22:50:07Z

Sorry for the delay! I've added some changes based on your comment.

I've added an additional integration test, hopefully it will be OK.

lukecwik · 2019-05-31T18:38:33Z

Run Java PostCommit

lukecwik · 2019-05-31T18:38:57Z

I kicked off the postcommit to validate that the IT you added.

lukecwik · 2019-05-31T21:09:38Z

...a/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubReadIT.java

+        signal.signalSuccessWhen(
+            messages.getCoder(),
+            pubsubMessages ->
+                pubsubMessages.stream().noneMatch(m -> Strings.isNullOrEmpty(m.getMessageId()))));


It looks like the new IT doesn't pass:
https://builds.apache.org/job/beam_PostCommit_Java_PR/169/testReport/junit/org.apache.beam.sdk.io.gcp.pubsub/PubsubReadIT/testReadPubsubMessageId/

Looks like the IT passed this time. Could you take another look?

Thanks!

thinhha · 2019-06-12T08:41:52Z

Run Java PostCommit

thinhha · 2019-06-23T22:36:47Z

Run Java PostCommit

thinhha · 2019-06-23T23:18:07Z

Run Java PreCommit

thinhha · 2019-06-26T11:08:53Z

Run Java PostCommit

lukecwik · 2019-07-09T18:17:08Z

Run Java PostCommit

thinhha · 2019-07-28T17:35:21Z

Run Java PostCommit

thinhha · 2019-07-28T18:43:06Z

Run Java PostCommit

thinhha · 2019-07-28T19:17:19Z

Run Java PostCommit

thinhha · 2019-07-28T20:06:52Z

Run Java PostCommit

thinhha · 2019-07-28T20:33:57Z

Run Java PostCommit

thinhha · 2019-07-28T20:38:43Z

Run Java PostCommit

thinhha · 2019-07-28T21:01:16Z

Run Java PreCommit

thinhha · 2019-07-28T22:12:02Z

Run Java PostCommit

thinhha · 2019-07-28T23:23:32Z

Run Java PostCommit

thinhha · 2019-07-28T23:34:44Z

Run Java PostCommit

thinhha · 2019-08-04T23:42:55Z

Hi @lukecwik, looks like the non-messageId coders are hard-coded in PubsubUnboundedSource:

Line 1088 of beam/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubUnboundedSource.java:

    @Override
    public Coder<PubsubMessage> getOutputCoder() {
      return outer.getNeedsAttributes()
          ? PubsubMessageWithAttributesCoder.of()
          : PubsubMessagePayloadOnlyCoder.of();
    }

This means that while the PubsubMessage is created with the messageId field populated when using PubsubIO.readMessagesWithMessageId()/PubsubIO.readMessagesWithAttributesAndMessageId() , this field does not show up in the subsequent step.

Looks like setNeedsAttributes is a required field. I'm not sure how to change getOutputCoder()
without breaking something.

How do you think this case can be handled?

lukecwik · 2019-08-05T11:44:29Z

Add another constructor to PubsubUnboundedSource where you pass in a boolean as to whether you need the message id and update the getOuptutCoder to look at this property selecting between the 4 pubsub message coder variants. You'll also want to update

beam/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java

Line 1146 in 667bf85

new PubsubMessageWithAttributesCoder());

to look at this property and select between the coder with the message id and attributes or just the attributes coder based upon the new boolean that you added.

...le-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubUnboundedSource.java

lukecwik · 2019-08-09T15:41:48Z

It turns out that one of the Dataflow services doesn't output the message id which makes this change not work with Dataflow until the service is updated. @udim is working on getting that resolved.

thinhha · 2019-08-11T17:27:57Z

Thanks for the update @lukecwik. Good thing the IT caught this!

I've added the changes we discussed above. We can rerun the postCommit tests once the dataflow backend change is complete.

…der()

thinhha · 2019-08-17T17:59:15Z

Run Java PostCommit

thinhha · 2019-08-18T22:07:22Z

Run Java PostCommit

thinhha · 2019-08-28T13:49:53Z

All tests are passed. Friendly ping for review @lukecwik.

lukecwik · 2019-08-30T21:56:40Z

Thanks a lot for contributing this and dealing with the additional complexity that Dataflow added.

lukecwik requested changes May 1, 2019

View reviewed changes

thinhha force-pushed the BEAM-3489 branch 2 times, most recently from a2d1aff to 6f781b2 Compare May 4, 2019 23:07

lukecwik reviewed May 6, 2019

View reviewed changes

lukecwik requested changes May 6, 2019

View reviewed changes

thinhha force-pushed the BEAM-3489 branch from 6f781b2 to 03df21c Compare May 26, 2019 22:44

lukecwik requested changes May 31, 2019

View reviewed changes

thinhha force-pushed the BEAM-3489 branch from 539a39f to 4a06264 Compare June 23, 2019 22:36

thinhha force-pushed the BEAM-3489 branch from b0e1445 to d9edb8e Compare June 26, 2019 07:20

thinhha force-pushed the BEAM-3489 branch from 12d44fb to 4b70620 Compare July 28, 2019 18:41

thinhha force-pushed the BEAM-3489 branch from 4b70620 to 778970c Compare July 28, 2019 19:16

thinhha force-pushed the BEAM-3489 branch from 778970c to 20aca08 Compare July 28, 2019 20:06

thinhha force-pushed the BEAM-3489 branch from e4a56c7 to 65556b2 Compare July 28, 2019 22:10

thinhha force-pushed the BEAM-3489 branch 2 times, most recently from e669585 to 8af31b1 Compare July 28, 2019 23:34

lukecwik mentioned this pull request Aug 5, 2019

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

Merged

udim reviewed Aug 9, 2019

View reviewed changes

...le-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubUnboundedSource.java Show resolved Hide resolved

thinhha added 2 commits August 11, 2019 17:58

[BEAM-3489] add PubSub messageId to PubsubMessage

f1d02e9

[BEAM-3489] add Coder test

71f6ddf

thinhha force-pushed the BEAM-3489 branch 3 times, most recently from a2d935c to c02cdbd Compare August 11, 2019 17:25

thinhha force-pushed the BEAM-3489 branch from c02cdbd to 2dcb862 Compare August 11, 2019 17:33

[BEAM-3489] add messageId coders to PubsubUnboundedSource.getOutputCo…

d0b9cac

…der()

thinhha force-pushed the BEAM-3489 branch from 2dcb862 to d0b9cac Compare August 17, 2019 17:21

[BEAM-3489] update docs

b8313b0

lukecwik approved these changes Aug 30, 2019

View reviewed changes

lukecwik merged commit a9e14ff into apache:master Aug 30, 2019

TheNeuralBit mentioned this pull request Aug 22, 2022

BEAM-13592 Add getOrderingKey in o.a.b.sdk.io.gcp.pubsub.PubsubMessage #22216

Merged

4 tasks

	* attached attributes, and an optional messageId.
	* attached attributes, and a message id.

[BEAM-3489] add PubSub messageId in PubsubMessage #8370

[BEAM-3489] add PubSub messageId in PubsubMessage #8370

Uh oh!

Conversation

thinhha commented Apr 21, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Post-Commit Tests Status (on master branch)

Pre-Commit Tests Status (on master branch)

Uh oh!

thinhha commented Apr 21, 2019

Uh oh!

thinhha commented Apr 21, 2019

Uh oh!

lukecwik left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

thinhha commented May 5, 2019

Uh oh!

lukecwik May 6, 2019

Choose a reason for hiding this comment

Uh oh!

lukecwik left a comment

Choose a reason for hiding this comment

Uh oh!

lukecwik May 6, 2019

Choose a reason for hiding this comment

Uh oh!

lukecwik May 6, 2019

Choose a reason for hiding this comment

Uh oh!

lukecwik May 6, 2019

Choose a reason for hiding this comment

Uh oh!

lukecwik commented May 6, 2019

Uh oh!

lukecwik commented May 21, 2019

Uh oh!

thinhha commented May 26, 2019

Uh oh!

lukecwik commented May 31, 2019

Uh oh!

lukecwik commented May 31, 2019

Uh oh!

lukecwik May 31, 2019

Choose a reason for hiding this comment

Uh oh!

thinhha Aug 17, 2019

Choose a reason for hiding this comment

Uh oh!

thinhha commented Jun 12, 2019

Uh oh!

thinhha commented Jun 23, 2019

Uh oh!

thinhha commented Jun 23, 2019

Uh oh!

thinhha commented Jun 26, 2019

Uh oh!

lukecwik commented Jul 9, 2019

Uh oh!

thinhha commented Jul 28, 2019

Uh oh!

thinhha commented Jul 28, 2019

Uh oh!

thinhha commented Jul 28, 2019

Uh oh!

thinhha commented Jul 28, 2019

Uh oh!

thinhha commented Jul 28, 2019

Uh oh!

thinhha commented Jul 28, 2019

Uh oh!

thinhha commented Jul 28, 2019

Uh oh!

thinhha commented Jul 28, 2019

Uh oh!

thinhha commented Jul 28, 2019

Uh oh!

thinhha commented Jul 28, 2019

Uh oh!

thinhha commented Apr 21, 2019 •

edited

Loading

lukecwik left a comment •

edited

Loading

thinhha commented Aug 4, 2019 •

edited

Loading

thinhha commented Aug 11, 2019 •

edited

Loading