You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by GitBox <gi...@apache.org> on 2021/03/22 07:56:03 UTC
[GitHub] [beam] victuos opened a new pull request #14294: [BEAM-12007] Set message ordering key for Google Pubsub
victuos opened a new pull request #14294:
URL: https://github.com/apache/beam/pull/14294
Implemented ordering key for Google Pubsub
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] [**Choose reviewer(s)**](https://beam.apache.org/contribute/#make-your-change) 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.
- [ ] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://beam.apache.org/contribute/#make-reviewers-job-easier).
Post-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
Lang | SDK | Dataflow | Flink | Samza | Spark | Twister2
--- | --- | --- | --- | --- | --- | ---
Go | [](https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Flink/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Spark/lastCompletedBuild/) | ---
Java | [](https://ci-beam.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_VR_Dataflow_V2/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Java11/lastCompletedBuild/) | [](https://ci-beam
.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink_Java11/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Streaming/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Spark_Batch/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Twister2/lastCompletedBuild/)
Python | [](https://ci-beam.apache.org/job/beam_PostCommit_Python36/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python37/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python38/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow_V2/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Python_PVR_Flink_Cron/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Flink/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Spark/lastCompletedBuild/) | ---
XLang | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Direct/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Dataflow/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Flink/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Spark/lastCompletedBuild/) | ---
Pre-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
--- |Java | Python | Go | Website | Whitespace | Typescript
--- | --- | --- | --- | --- | --- | ---
Non-portable | [](https://ci-beam.apache.org/job/beam_PreCommit_Java_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Python_Cron/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PreCommit_PythonLint_Cron/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PreCommit_PythonDocker_Cron/lastCompletedBuild/) <br>[](https://ci-beam.apache.org/job/beam_PreCommit_PythonDocs_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Go_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Website_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Whitespace_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Typescript_Cron/lastCompletedBuild/)
Portable | --- | [](https://ci-beam.apache.org/job/beam_PreCommit_Portable_Python_Cron/lastCompletedBuild/) | --- | --- | --- | ---
See [.test-infra/jenkins/README](https://github.com/apache/beam/blob/master/.test-infra/jenkins/README.md) for trigger phrase, status and link of all Jenkins jobs.
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] aaltay commented on pull request #14294: [BEAM-12007] Set message ordering key for Google Pubsub
Posted by GitBox <gi...@apache.org>.
aaltay commented on pull request #14294:
URL: https://github.com/apache/beam/pull/14294#issuecomment-812073152
@victuos @dpcollins-google - What would be the next step on this PR?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] haraldrombaut commented on a change in pull request #14294: [BEAM-12007] Set message ordering key for Google Pubsub
Posted by GitBox <gi...@apache.org>.
haraldrombaut commented on a change in pull request #14294:
URL: https://github.com/apache/beam/pull/14294#discussion_r599528698
##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIO.java
##########
@@ -1093,14 +1090,7 @@ public void processElement(ProcessContext c) throws IOException, SizeLimitExceed
}
// NOTE: The record id is always null.
- output.add(
- OutgoingMessage.of(
- com.google.pubsub.v1.PubsubMessage.newBuilder()
- .setData(ByteString.copyFrom(payload))
- .putAllAttributes(attributes)
- .build(),
- c.timestamp().getMillis(),
- null));
+ output.add(OutgoingMessage.of(message, c.timestamp().getMillis(), null));
Review comment:
The unbounded sink already uses this static method, where the com.google.pubsub.v1.PubsubMessage is being created from the PubsubIO PubsubMessage
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] dpcollins-google commented on pull request #14294: [BEAM-12007] Set message ordering key for Google Pubsub
Posted by GitBox <gi...@apache.org>.
dpcollins-google commented on pull request #14294:
URL: https://github.com/apache/beam/pull/14294#issuecomment-804889499
> We would expect that when a consumer of a subscription, with ordering enabled, would receive the outputted messages (with the same ordering key) in the order they were published.
This will likely not be achieved by this PR. The beam model does not preserve ordering within a PCollection, allowing elements to be reordered arbitrarily.
> The order in which these messages are published is as far as I know determined by the timestamp of each element.
I do not believe a Beam runner is required to process windows in timestamp order: if processing of a given window stalls, another window can begin to proceed concurrently. In this case, it would result in disordered messages between windows.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] haraldrombaut edited a comment on pull request #14294: [BEAM-12007] Set message ordering key for Google Pubsub
Posted by GitBox <gi...@apache.org>.
haraldrombaut edited a comment on pull request #14294:
URL: https://github.com/apache/beam/pull/14294#issuecomment-804884719
> > Beam PCollections do not guarantee ordering of elements
>
> I think what would be useful to know from the author, is what they would expect the result to be if they set this key in beam.
>
> @victuos, were you expecting the result of this to be message ordering preserved end to end through your pipeline? If so, that may not be the result here.
We would expect that when a consumer of a subscription, with ordering enabled, would receive the outputted messages (with the same ordering key) in the order they were published. The order in which these messages are published is as far as I know determined by the timestamp of each element.
Being able to set the ordering key will not ensure end to end in order message processing. But if we group elements in a window by the ordering key, and output those elements with a specific timestamp, we can ensure they're published in order (for that window) and ordering can be enabled on subscriptions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] dpcollins-google commented on pull request #14294: [BEAM-12007] Set message ordering key for Google Pubsub
Posted by GitBox <gi...@apache.org>.
dpcollins-google commented on pull request #14294:
URL: https://github.com/apache/beam/pull/14294#issuecomment-812093808
Thank you @victuos for doing this. Unfortunately, I don't believe this will work as written since Beam will not preserve any form of order on the publish side through to Pub/Sub for the ordering key to be meaningful. The Beam model is complex and not all aspects are well documented, this might be one of those areas to improve. I would suggest closing this for now and keeping this jira open for a future complete implementation. @victuos, what do you think?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] victuos commented on pull request #14294: [BEAM-12007] Set message ordering key for Google Pubsub
Posted by GitBox <gi...@apache.org>.
victuos commented on pull request #14294:
URL: https://github.com/apache/beam/pull/14294#issuecomment-803849471
R: @chamikaramj @charlesccychen
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] haraldrombaut commented on pull request #14294: [BEAM-12007] Set message ordering key for Google Pubsub
Posted by GitBox <gi...@apache.org>.
haraldrombaut commented on pull request #14294:
URL: https://github.com/apache/beam/pull/14294#issuecomment-804884719
> I think what would be useful to know from the author, is what they would expect the result to be if they set this key in beam.
We would expect that when a consumer of a subscription, with ordering enabled, would receive the outputted messages (with the same ordering key) in the order they were published. The order in which these messages are published is as far as I know determined by the timestamp of each element.
Being able to set the ordering key will not ensure end to end in order message processing. But if we group elements in a window by the ordering key, and output those elements with a specific timestamp, we can ensure they're published in order (for that window) and ordering can be enabled on subscriptions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] chamikaramj commented on a change in pull request #14294: [BEAM-12007] Set message ordering key for Google Pubsub
Posted by GitBox <gi...@apache.org>.
chamikaramj commented on a change in pull request #14294:
URL: https://github.com/apache/beam/pull/14294#discussion_r598910775
##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIO.java
##########
@@ -1093,14 +1090,7 @@ public void processElement(ProcessContext c) throws IOException, SizeLimitExceed
}
// NOTE: The record id is always null.
- output.add(
- OutgoingMessage.of(
- com.google.pubsub.v1.PubsubMessage.newBuilder()
- .setData(ByteString.copyFrom(payload))
- .putAllAttributes(attributes)
- .build(),
- c.timestamp().getMillis(),
- null));
+ output.add(OutgoingMessage.of(message, c.timestamp().getMillis(), null));
Review comment:
Note that this only affects the bounded case (usually batch pipelines).
##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubCoderProviderRegistrar.java
##########
@@ -31,6 +31,9 @@
@Override
public List<CoderProvider> getCoderProviders() {
return ImmutableList.of(
+ CoderProviders.forCoder(
+ TypeDescriptor.of(PubsubMessage.class),
Review comment:
I think this change unfortunately will break update feature for runners that support it (for example Dataflow).
cc: @reuvenlax
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] victuos closed pull request #14294: [BEAM-12007] Set message ordering key for Google Pubsub
Posted by GitBox <gi...@apache.org>.
victuos closed pull request #14294:
URL: https://github.com/apache/beam/pull/14294
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] victuos commented on pull request #14294: [BEAM-12007] Set message ordering key for Google Pubsub
Posted by GitBox <gi...@apache.org>.
victuos commented on pull request #14294:
URL: https://github.com/apache/beam/pull/14294#issuecomment-812105322
> @victuos @dpcollins-google - What would be the next step on this PR?
> Thank you @victuos for doing this. Unfortunately, I don't believe this will work as written since Beam will not preserve any form of order on the publish side through to Pub/Sub for the ordering key to be meaningful. The Beam model is complex and not all aspects are well documented, this might be one of those areas to improve. I would suggest closing this for now and keeping this jira open for a future complete implementation. @victuos, what do you think?
Unfortunately I agree, so I'll close the PR.
We have set up an internal meeting with Google devs on how we should continue. I will update the Jira ticket once I have more information.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] dpcollins-google commented on pull request #14294: [BEAM-12007] Set message ordering key for Google Pubsub
Posted by GitBox <gi...@apache.org>.
dpcollins-google commented on pull request #14294:
URL: https://github.com/apache/beam/pull/14294#issuecomment-804253112
> Beam PCollections do not guarantee ordering of elements
I think what would be useful to know from the author, is what they would expect the result to be if they set this key in beam.
@victuos, were you expecting the result of this to be message ordering preserved end to end through your pipeline? If so, that may not be the result here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org