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

gemelen · 2022-07-11T07:55:54Z

Add orderingKey from proto PubsubMessage to beam Pubsubmessage.
Add a new coder for a PubsubMessage with all fields.
Propagate this new view of a PubsubMessage to PubsubIO as a new API
call.
Add a unit test for a coder.

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).
Mention the appropriate issue in your description (for example: addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment fixes #<ISSUE NUMBER> instead.
Update CHANGES.md with noteworthy changes.
If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI.

…bsubMessage Add orderingKey from proto PubsubMessage to beam Pubsubmessage. Add a new coder for a PubsubMessage with all fields. Propagate this new view of a PubsubMessage to PubsubIO as a new API call. Add a unit test for a coder.

gemelen · 2022-07-11T07:57:03Z

Solves (hopefully)

Add getOrderingKey in org.apache.beam.sdk.io.gcp.pubsub.PubsubMessage #21304

gemelen · 2022-07-11T08:11:53Z

@damccorm Hi, could you please suggest a reviewer for this changeset?

gemelen · 2022-07-11T08:30:00Z

Addresses BEAM-13592

damccorm · 2022-07-11T16:50:45Z

@johnjcasey who is a good reviewer here?

johnjcasey · 2022-07-11T17:26:34Z

I am, I'll take a look

johnjcasey · 2022-07-11T19:02:35Z

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

+   * PubsubMessage#getAttributeMap() attributes}, along with the {@link PubsubMessage#getMessageId()
+   * messageId} and {PubsubMessage#getOrderingKey() orderingKey} from PubSub.
+   */
+  public static Read<PubsubMessage> readMessagesWithAllAttributesAndMessageIdAndOrderingKey() {


Can we remove "all" from the message name to keep it consistent with the above method?
Also, based on the above method, do we need .setNeedsAttributes(true) here?
Finally, should we name the coder PubsubMessageCoder, given that the above method calls it PubsubMessageWithAttributesAndMessageIdCoder?

I think that since this coder grabs all fields of an incoming PubsubMessage then it could be named without specifying all these fields in its name. Accordingly, method would be better named readMessages but this name is already taken by other implementation (that reads only payload of PubsubMessage).

I agree that readMessagesWithAllAttributesAndMessageIdAndOrderingKey then should be named readMessagesWithAttributesAndMessageIdAndOrderingKey, so I'm updating it to this.

Regardint the

Also, based on the above method, do we need .setNeedsAttributes(true) here?

newly added method doesn't include setNeedsAttributes, it's a setNeedsOrderingKey:

public static Read<PubsubMessage> readMessagesWithAttributesAndMessageIdAndOrderingKey() { return Read.newBuilder() .setCoder(PubsubMessageCoder.of()) .setNeedsOrderingKey(true) .build(); }

johnjcasey · 2022-07-11T19:03:36Z

...oogle-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubMessageCoder.java

+  // A message's payload cannot be null
+  private static final Coder<byte[]> PAYLOAD_CODER = ByteArrayCoder.of();
+  // A message's attributes can be null.
+  private static final Coder<Map<String, String>> ATTRIBUTES_CODER =


can the attribute map be null, or can elements within the map be null?

This part is identical to other coders, so I assume that attributes field either null or contains something.
My guess that elements could be null too, but this needs to be changed in all PubsubMessage-related coders.

johnjcasey · 2022-07-11T19:04:08Z

...oogle-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubMessageCoder.java

+    PAYLOAD_CODER.encode(value.getPayload(), outStream);
+    ATTRIBUTES_CODER.encode(value.getAttributeMap(), outStream);
+    MESSAGE_ID_CODER.encode(value.getMessageId(), outStream);
+    // TODO(discuss what to do with publish_time field)


What needs to be discussed here?

I'd like to discuss if my approach here is correct:

This changeset aims to add only ordering key property of a PubsubMessage, but it's not possible to omit publish_time since it precedes ordering_key in protobuf schema PubsubMessage and thus in byte stream.

Changeset doesn't expose publish_time and omits it after byte stream deserialization in decode function. Also, in encode function of the relevant coder I use meaningless default instance of protobuf.Timestamp just to add required amount of bytes into serialized form of a PubsubMessage.

So, is it correct way or you could suggest what needs to be done with this field instead?

gemelen · 2022-07-18T11:47:47Z

@johnjcasey could you please take a look on a changeset once more

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

egalpin · 2022-07-19T13:21:27Z

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

        .build();
  }

+  /**


I believe to support bounded writes, on line 1329 of this file you'll need to add explicit copying of the orderingKey as well (in the output.add(OutgoingMessage.of(... section of processElement

egalpin · 2022-07-19T13:30:40Z

Relates to #21162. I think it may also be worth mentioning that there was a previous email thread[1] about the feature to support subscriptions where ordering is enabled, but Beam itself does not guarantee ordering so this could be a confusing user experience. I may be missing additional context which has developed since that 2-year-old thread though!

Writes with orderingKey should definitely be supported though :-)

[1] https://lists.apache.org/thread/1koqhmhr1tb5572cj3fq5b65dnwbg7fg

egalpin · 2022-07-19T13:43:21Z

...oogle-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubMessageCoder.java

+@SuppressWarnings({
+  "nullness" // TODO(https://github.com/apache/beam/issues/20497)
+})
+public class PubsubMessageCoder extends CustomCoder<PubsubMessage> {


I noticed that the other coders, like PubsubMessagePayloadOnlyCoder have specific names indicating the properties of the messages which they are meant to be the Coder for. IMO the name PubsubMessageCoder implies by name that it's a generic coder for pubsub messages. Following the naming convention in place, this class might be named PubsubMessageWithAttributesAndMessageIdAndOrderingKeyCoder (which is comically long, so I'm not sure which is better haha).

I believe you'll also need to register this in PubsubCoderProviderRegistrar

Which I think actually implies that there will need to be a coder for all possible permutations (payload, messageId, attributes, orderingKey)

+1 for giving this the comically long, specific name to be consistent

egalpin · 2022-07-20T13:01:21Z

I believe IncomingMessage and OutgoingMessage (defined in PubsubClient.java)also need to have their static constructors (of(..) methods, maybe there's a more proper name for these?) updated to also pull in message.getOrderingKey()

egalpin · 2022-07-20T20:41:41Z

At the risk of hijacking... Having support for orderingKey is high-priority for me as well, so I've been playing around with this locally. I can't for the life of me get it to work with dataflow when attempting to publish messages with an ordering key. I did get it to work locally, where I could see (via the GCP UI) published messages had orderingKey assigned. Something must be overriding it in dataflow. Thought it might be worth mentioning.

I also had a lot of trouble with coders and getting the correct coder to be maintained throughout the publish process, specifically the coder would be altered to be PubsubMessageWithAttributesCoder after data went through PubsubMessages.ParsePayloadAsPubsubMessageProto. I'm definitely missing an understanding of how to make proper use of CoderRegistrar

egalpin · 2022-07-20T21:04:36Z

here's a comparison showing some of the changes I was playing with: https://github.com/gemelen/beam/compare/beam-13592-pubsub-java-orderingkey...egalpin:beam:egalpin/gemelen-beam-13592-fork?expand=1

…ub registrar

egalpin · 2022-07-22T15:35:17Z

Ok the publish side does in fact work: https://lists.apache.org/thread/c6929ms0bjxtcw9ho4tdb5y3t8wnwnfy

I'd like to include those changes with the changes already added by @gemelen. Any opposition?

egalpin · 2022-07-22T20:05:02Z

...d-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubCoderProviderRegistrar.java

  @Override
  public List<CoderProvider> getCoderProviders() {
    return ImmutableList.of(
+        CoderProviders.forCoder(TypeDescriptor.of(PubsubMessage.class), PubsubMessageCoder.of()),


I'm not 100% confident in the order that is required here, or if order matters. I couldn't find documentation around how coders are resolved when multiple coders are registered for the same class. I tried tracing the code but wasn't able to come up with a definitive answer

egalpin · 2022-07-25T20:19:35Z

...d-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubCoderProviderRegistrar.java

            TypeDescriptor.of(PubsubMessage.class),
-            PubsubMessageWithAttributesAndMessageIdCoder.of()));
+            PubsubMessageWithAttributesAndMessageIdCoder.of()),
+        CoderProviders.forCoder(TypeDescriptor.of(PubsubMessage.class), PubsubMessageCoder.of()));


moved this code to be the last in the registrar (rather than first), otherwise there are failures in the tests. @johnjcasey Do you have any further info on coder registrars and how a coder is resolved when multiple coders are registered for the same class?

I do not, sorry

generally though, I would be surprised to have multiple coder types for a single type, though I don't know if that is my unfamiliarity with coders or not

@TheNeuralBit I think this is the area where there is lack of clarity. As best as I can tell, here there are multiple coders registered for the same class (PubsubMessage.class). It seems like order matters here, but I'm not sure all the implications. Ex. if I put the new coder first in the registrar, existing pubsub unit tests will fail with coder errors.

Thanks! I dug into the infrastructure here, and it looks like we will always just use the first coder registered for a given type, so I think registering these other three will be a no-op. In theory a coder could raise CannotProvideCoderException and we'd move on to the next one:

beam/sdks/java/core/src/main/java/org/apache/beam/sdk/coders/CoderRegistry.java

Line 678 in 4ffeae4

} catch (CannotProvideCoderException e) {

but that's not going to happen here, PubsubMessageWithAttributesCoder will never do that.

I think the actually important change is just adding readMessagesWithAttributesAndMessageIdAndOrderingKey with the setCoder call.

It's worth noting that what we specify here will be used as the default in any PCollection<PubsubMessage> that's not created by one of the PubsubIO helper methods. So we should consider making it the one that will always work. There are implications for breaking pipeline update though if we change the default. My suggestion:

Leave this as-is for now (don't add the new coder)

We can file an issue to follow-up on changing this

(I gathered some context by looking at #8370, and also spoke to @lukecwik)

Thanks for the info @TheNeuralBit , this is super helpful context. Sounds like what's here then is not going to break anything, which is the most important.

One aspect that I predict will be confusing for users is that if one attempts to write pubsub messages with ordering key, the PubsubMessageWithAttributesCoder will be used by default unless setCoder is called; in that default case, the orderingKey will be "mysteriously" stripped away by the coder. I suppose we might also need a new write helper method like writeMessagesWithOrderingKeys in addition to writeMessages in order help users to set the correct coder?

Filed #23525 for the follow-up

gemelen · 2022-08-18T07:48:05Z

@damccorm Could you please help to find a reviewer?

damccorm · 2022-08-18T12:49:34Z

@johnjcasey mind taking another look?

johnjcasey

LGTM

johnjcasey · 2022-08-18T13:58:10Z

...d-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubCoderProviderRegistrar.java

            TypeDescriptor.of(PubsubMessage.class),
-            PubsubMessageWithAttributesAndMessageIdCoder.of()));
+            PubsubMessageWithAttributesAndMessageIdCoder.of()),
+        CoderProviders.forCoder(TypeDescriptor.of(PubsubMessage.class), PubsubMessageCoder.of()));


I do not, sorry

johnjcasey · 2022-08-18T13:58:55Z

...d-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubCoderProviderRegistrar.java

            TypeDescriptor.of(PubsubMessage.class),
-            PubsubMessageWithAttributesAndMessageIdCoder.of()));
+            PubsubMessageWithAttributesAndMessageIdCoder.of()),
+        CoderProviders.forCoder(TypeDescriptor.of(PubsubMessage.class), PubsubMessageCoder.of()));


generally though, I would be surprised to have multiple coder types for a single type, though I don't know if that is my unfamiliarity with coders or not

johnjcasey · 2022-08-18T14:01:33Z

Though it would be good to get a better understanding of coders first. I think @TheNeuralBit has some more experience there

TheNeuralBit · 2022-08-22T18:34:36Z

Though it would be good to get a better understanding of coders first. I think @TheNeuralBit has some more experience there

Is there a specific question I can help with?

…AndOrderingKeyCoder

egalpin · 2022-09-29T15:08:24Z

Friendly bump, anything outstanding here or is this ready to be merged? (cc @johnjcasey @TheNeuralBit )

TheNeuralBit · 2022-10-06T20:03:05Z

Friendly bump, anything outstanding here or is this ready to be merged? (cc @johnjcasey @TheNeuralBit )

Apologies. I think this is ready to merge once the merge conflict is resolved.

I will file an issue to track the default coder issue

…eam-13592-fork

egalpin · 2022-10-07T20:28:04Z

thanks @TheNeuralBit ! I merged in master + resolved conflicts. For future reference, is merge or rebase preferred here? I opted for merge so that prior commits would not be marked as new for reviewers.

egalpin · 2022-10-11T14:36:33Z

Thank you @gemelen for kicking off this high-value addition to PubsubIO! 🎉

gemelen · 2022-10-11T14:40:50Z

Thank you @gemelen for kicking off this high-value addition to PubsubIO! 🎉

You deserve thank you for that changeset more than anybody else.

gemelen · 2022-10-11T14:43:37Z

Oh, last thing is to update https://issues.apache.org/jira/browse/BEAM-13592

TheNeuralBit · 2022-10-14T00:15:42Z

thanks @TheNeuralBit ! I merged in master + resolved conflicts. For future reference, is merge or rebase preferred here? I opted for merge so that prior commits would not be marked as new for reviewers.

I think this is just up to personal preference :)

github-actions bot added gcp io java labels Jul 11, 2022

Add CHANGES entry

d6f5ae4

gemelen marked this pull request as ready for review July 11, 2022 08:26

johnjcasey requested changes Jul 11, 2022

View reviewed changes

Rename transform name according to review comment

267cd10

egalpin reviewed Jul 18, 2022

View reviewed changes

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

Update to pass ordering key

bc97c6b

egalpin reviewed Jul 19, 2022

View reviewed changes

gemelen and others added 7 commits July 21, 2022 16:35

Add CHANGES entry

7be63b9

Adds ordering key to OutgoingMessage builder, adds new coders to pubs…

bd1050a

…ub registrar

Fixes pubsub bounded writer allowing for orderingKey

e2a7c44

Alters order of pubsub message support in registrar

bc60b74

Removed publishTime and messageId in grpc pubsub client publish

46497d1

Attempts to allow different pubsub root url for PubsubIO.Write

c3ea563

Fixes pubsub tests root url

92e4d39

egalpin reviewed Jul 22, 2022

View reviewed changes

Puts PubsubMessageCoder last in registrar

0d622d9

egalpin reviewed Jul 25, 2022

View reviewed changes

Uses MoreObjects over Objects

8d50229

johnjcasey approved these changes Aug 18, 2022

View reviewed changes

Renames PubsubMessageCoder to PubsubMessageWithAttributesAndMessageId…

fd7dedd

…AndOrderingKeyCoder

TheNeuralBit mentioned this pull request Oct 6, 2022

[Bug]: Default PubsubMessage coder will drop message id and orderingKey #23525

Closed

Merge remote-tracking branch 'upstream/master' into egalpin/gemelen-b…

9ec1429

…eam-13592-fork

egalpin merged commit 1d573e2 into apache:master Oct 11, 2022

gemelen deleted the beam-13592-pubsub-java-orderingkey branch October 11, 2022 14:41

egalpin mentioned this pull request Oct 11, 2022

Add getOrderingKey in org.apache.beam.sdk.io.gcp.pubsub.PubsubMessage #21304

Closed

egalpin mentioned this pull request Jan 3, 2023

Add ability to Write to GCP PubSub with an orderingKey #21162

Closed

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

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

Uh oh!

Conversation

gemelen commented Jul 11, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

GitHub Actions Tests Status (on master branch)

Uh oh!

gemelen commented Jul 11, 2022

Uh oh!

gemelen commented Jul 11, 2022

Uh oh!

gemelen commented Jul 11, 2022

Uh oh!

damccorm commented Jul 11, 2022

Uh oh!

johnjcasey commented Jul 11, 2022

Uh oh!

Choose a reason for hiding this comment

Uh oh!

gemelen Jul 12, 2022 • 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!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

gemelen commented Jul 18, 2022

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

egalpin commented Jul 19, 2022

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

egalpin commented Jul 20, 2022

Uh oh!

egalpin commented Jul 20, 2022

Uh oh!

egalpin commented Jul 20, 2022

Uh oh!

egalpin commented Jul 22, 2022

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

TheNeuralBit Aug 22, 2022 • 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!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

gemelen commented Aug 18, 2022

Uh oh!

damccorm commented Aug 18, 2022

Uh oh!

johnjcasey left a comment

Choose a reason for hiding this comment

Uh oh!

gemelen commented Jul 11, 2022 •

edited

Loading

gemelen Jul 12, 2022 •

edited

Loading

TheNeuralBit Aug 22, 2022 •

edited

Loading