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/06/08 23:00:25 UTC
[GitHub] [beam] zhoufek opened a new pull request #14971: [BEAM-12297] Add methods to PubsubIO for reading DynamicMessage
zhoufek opened a new pull request #14971:
URL: https://github.com/apache/beam/pull/14971
This adds a couple methods to `PubsubIO` for reading in a `DynamicMessage` given either a `Descriptor` or a `ProtoDomain` and message name. This can be useful if the type can't be known at compile time for whatever reason.
------------------------
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.
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).
`ValidatesRunner` compliance status (on master branch)
--------------------------------------------------------
<table>
<thead>
<tr>
<th>Lang</th>
<th>ULR</th>
<th>Dataflow</th>
<th>Flink</th>
<th>Samza</th>
<th>Spark</th>
<th>Twister2</th>
</tr>
</thead>
<tbody>
<tr>
<td>Go</td>
<td>---</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/badge/icon">
</a>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Flink/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Flink/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Spark/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Spark/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
</tr>
<tr>
<td>Java</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_ULR/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_ULR/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/badge/icon?subject=V1">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Streaming/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Streaming/lastCompletedBuild/badge/icon?subject=V1+Streaming">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Java11/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Java11/lastCompletedBuild/badge/icon?subject=V1+Java+11">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_VR_Dataflow_V2/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_VR_Dataflow_V2/lastCompletedBuild/badge/icon?subject=V2">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_VR_Dataflow_V2_Streaming/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_VR_Dataflow_V2_Streaming/lastCompletedBuild/badge/icon?subject=V2+Streaming">
</a><br>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/badge/icon?subject=Java+8">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink_Java11/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink_Java11/lastCompletedBuild/badge/icon?subject=Java+11">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch/lastCompletedBuild/badge/icon?subject=Portable">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Streaming/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Streaming/lastCompletedBuild/badge/icon?subject=Portable+Streaming">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Samza/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Samza/lastCompletedBuild/badge/icon?subject=Portable">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Spark_Batch/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Spark_Batch/lastCompletedBuild/badge/icon?subject=Portable">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming/lastCompletedBuild/badge/icon?subject=Structured+Streaming">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Twister2/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Twister2/lastCompletedBuild/badge/icon">
</a>
</td>
</tr>
<tr>
<td>Python</td>
<td>---</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/badge/icon?subject=V1">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow_V2/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow_V2/lastCompletedBuild/badge/icon?subject=V2">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/badge/icon?subject=ValCont">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Python_PVR_Flink_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Python_PVR_Flink_Cron/lastCompletedBuild/badge/icon?subject=Portable">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Flink/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Flink/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Spark/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Spark/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
</tr>
<tr>
<td>XLang</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Direct/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Direct/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Dataflow/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Dataflow/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Flink/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Flink/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Spark/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Spark/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
</tr>
</tbody>
</table>
Examples testing status on various runners
--------------------------------------------------------
<table>
<thead>
<tr>
<th>Lang</th>
<th>ULR</th>
<th>Dataflow</th>
<th>Flink</th>
<th>Samza</th>
<th>Spark</th>
<th>Twister2</th>
</tr>
</thead>
<tbody>
<tr>
<td>Go</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
</tr>
<tr>
<td>Java</td>
<td>---</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Java_Examples_Dataflow_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Java_Examples_Dataflow_Cron/lastCompletedBuild/badge/icon?subject=V1">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Java_Examples_Dataflow_Java11_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Java_Examples_Dataflow_Java11_Cron/lastCompletedBuild/badge/icon?subject=V1+Java11">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_Examples_Dataflow_V2/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_Examples_Dataflow_V2/lastCompletedBuild/badge/icon?subject=V2">
</a><br>
</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
</tr>
<tr>
<td>Python</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
</tr>
<tr>
<td>XLang</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
</tr>
</tbody>
</table>
Post-Commit SDK/Transform Integration Tests Status (on master branch)
------------------------------------------------------------------------------------------------
<table>
<thead>
<tr>
<th>Go</th>
<th>Java</th>
<th>Python</th>
</tr>
</thead>
<tbody>
<tr>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python36/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python36/lastCompletedBuild/badge/icon?subject=3.6">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python37/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python37/lastCompletedBuild/badge/icon?subject=3.7">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python38/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python38/lastCompletedBuild/badge/icon?subject=3.8">
</a>
</td>
</tr>
</tbody>
</table>
Pre-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
<table>
<thead>
<tr>
<th>---</th>
<th>Java</th>
<th>Python</th>
<th>Go</th>
<th>Website</th>
<th>Whitespace</th>
<th>Typescript</th>
</tr>
</thead>
<tbody>
<tr>
<td>Non-portable</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Java_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Java_Cron/lastCompletedBuild/badge/icon">
</a><br>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Python_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Python_Cron/lastCompletedBuild/badge/icon?subject=Tests">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_PythonLint_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_PythonLint_Cron/lastCompletedBuild/badge/icon?subject=Lint">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_PythonDocker_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_PythonDocker_Cron/badge/icon?subject=Docker">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_PythonDocs_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_PythonDocs_Cron/badge/icon?subject=Docs">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Go_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Go_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Website_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Website_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Whitespace_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Whitespace_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Typescript_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Typescript_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
</tr>
<tr>
<td>Portable</td>
<td>---</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Portable_Python_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Portable_Python_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_GoPortable_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_GoPortable_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
<td>---</td>
<td>---</td>
</tr>
</tbody>
</table>
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] zhoufek commented on pull request #14971: [BEAM-12297] Add methods to PubsubIO for reading DynamicMessage
Posted by GitBox <gi...@apache.org>.
zhoufek commented on pull request #14971:
URL: https://github.com/apache/beam/pull/14971#issuecomment-858801185
Run Java PreCommit
--
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] TheNeuralBit edited a comment on pull request #14971: [BEAM-12297] Add methods to PubsubIO for reading DynamicMessage
Posted by GitBox <gi...@apache.org>.
TheNeuralBit edited a comment on pull request #14971:
URL: https://github.com/apache/beam/pull/14971#issuecomment-857811062
Cool, thanks @zhoufek!
> CC @TheNeuralBit since we recently discussed the issue of PubsubIO requiring a Java class and how we would improve that requirement for portability.
Note we already have [ProtoDynamicMessageSchema](https://github.com/apache/beam/blob/master/sdks/java/extensions/protobuf/src/main/java/org/apache/beam/sdk/extensions/protobuf/ProtoDynamicMessageSchema.java) which can make a `SchemaCoder<DynamicMesage>` given a proto descriptor. That combined with this PR could be a good avenue for making the pubsub TableProvider configurable via a proto descriptor rather than a proto class name. (CC: @apilloud @robinyqiu)
EDIT: Oh I just looked at the code and this is already using ProtoDynamicMessageSchema, so it will just work with Beam schemas :) I think what we'd need to do to to use this approach in SQL and the PubSub TableProvider is implement the logic in the [ProtoPayloadSerializerProvider](https://github.com/apache/beam/blob/master/sdks/java/extensions/protobuf/src/main/java/org/apache/beam/sdk/extensions/protobuf/ProtoPayloadSerializerProvider.java)
--
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] zhoufek commented on a change in pull request #14971: [BEAM-12297] Add methods to PubsubIO for reading DynamicMessage
Posted by GitBox <gi...@apache.org>.
zhoufek commented on a change in pull request #14971:
URL: https://github.com/apache/beam/pull/14971#discussion_r649473094
##########
File path: sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIOTest.java
##########
@@ -416,6 +423,96 @@ public void after() throws IOException {
}
}
+ @Test
+ public void testProto() {
+ ProtoCoder<Primitive> coder = ProtoCoder.of(Primitive.class);
+ ImmutableList<Primitive> inputs =
+ ImmutableList.of(
+ Primitive.newBuilder().setPrimitiveInt32(42).build(),
+ Primitive.newBuilder().setPrimitiveBool(true).build(),
+ Primitive.newBuilder().setPrimitiveString("Hello, World!").build());
+ setupTestClient(inputs, coder);
+ PCollection<Primitive> read =
+ readPipeline.apply(
+ PubsubIO.readProtos(Primitive.class)
+ .fromSubscription(SUBSCRIPTION.getPath())
+ .withClock(CLOCK)
+ .withClientFactory(clientFactory));
+ PAssert.that(read).containsInAnyOrder(inputs);
+ readPipeline.run();
+ }
+
+ @Test
+ public void testProtoDynamicMessage() {
+ ProtoCoder<Primitive> coder = ProtoCoder.of(Primitive.class);
+ ImmutableList<Primitive> inputs =
+ ImmutableList.of(
+ Primitive.newBuilder().setPrimitiveInt32(42).build(),
+ Primitive.newBuilder().setPrimitiveBool(true).build(),
+ Primitive.newBuilder().setPrimitiveString("Hello, World!").build());
+ setupTestClient(inputs, coder);
+
+ ProtoDomain domain = ProtoDomain.buildFrom(Primitive.getDescriptor());
+ String name = Primitive.getDescriptor().getFullName();
+ PCollection<Primitive> read =
+ readPipeline
+ .apply(
+ PubsubIO.readProtoDynamicMessage(domain, name)
+ .fromSubscription(SUBSCRIPTION.getPath())
+ .withClock(CLOCK)
+ .withClientFactory(clientFactory))
+ // DynamicMessage doesn't work well with PAssert, but if the content can be successfully
+ // converted back into the original Primitive, then that should be good enough to
+ // consider it a successful read.
+ .apply(
+ "Return To Primitive",
+ MapElements.into(TypeDescriptor.of(Primitive.class))
+ .via(
+ (DynamicMessage message) -> {
Review comment:
I think that, since this is just a test, it would be acceptable to not catch the exception. Whether we fail by not catching an except or by asserting the failures are empty when they aren't, the end result is essentially the same.
--
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] TheNeuralBit commented on pull request #14971: [BEAM-12297] Add methods to PubsubIO for reading DynamicMessage
Posted by GitBox <gi...@apache.org>.
TheNeuralBit commented on pull request #14971:
URL: https://github.com/apache/beam/pull/14971#issuecomment-857811062
Cool, thanks @zhoufek!
> CC @TheNeuralBit since we recently discussed the issue of PubsubIO requiring a Java class and how we would improve that requirement for portability.
Note we already have [ProtoDynamicMessageSchema](https://github.com/apache/beam/blob/master/sdks/java/extensions/protobuf/src/main/java/org/apache/beam/sdk/extensions/protobuf/ProtoDynamicMessageSchema.java) which can make a `SchemaCoder<DynamicMesage>` given a proto descriptor. That combined with this PR could be a good avenue for making the pubsub TableProvider configurable via a proto descriptor rather than a proto class name.
--
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] zhoufek commented on a change in pull request #14971: [BEAM-12297] Add methods to PubsubIO for reading DynamicMessage
Posted by GitBox <gi...@apache.org>.
zhoufek commented on a change in pull request #14971:
URL: https://github.com/apache/beam/pull/14971#discussion_r648482674
##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIO.java
##########
@@ -483,6 +488,54 @@ public String toString() {
return Read.newBuilder(parsePayloadUsingCoder(coder)).setCoder(coder).build();
}
+ /**
+ * Returns a {@link PTransform} that continuously reads binary encoded protobuf messages for the
+ * type specified by {@code fullMessageName}.
+ *
+ * <p>This is primarily here for cases where the message type cannot be known at compile time. If
+ * it can be known, prefer {@link PubsubIO#readProtos(Class)}, as {@link DynamicMessage} tends to
+ * perform worse than concrete types.
+ *
+ * <p>Beam will infer a schema for the {@link DynamicMessage} schema. Note that some proto schema
+ * features are not supported by all sinks.
+ *
+ * @param domain The {@link ProtoDomain} that contains the target message and its dependencies.
+ * @param fullMessageName The full name of the message for lookup in {@code domain}.
+ */
+ @Experimental(Kind.SCHEMAS)
+ public static Read<DynamicMessage> readProtoDynamicMessage(
+ ProtoDomain domain, String fullMessageName) {
+ SerializableFunction<PubsubMessage, DynamicMessage> parser =
+ message -> {
+ try {
+ return DynamicMessage.parseFrom(
+ domain.getDescriptor(fullMessageName), message.getPayload());
+ } catch (InvalidProtocolBufferException e) {
+ throw new RuntimeException("Could not parse Pub/Sub message", e);
Review comment:
Currently, it isn't supported, and the other methods that parse binary content for Protos or Avro throw a RuntimeException like here.
I do think it would be valuable to have something like `withDeadLetter` to configure the returned `Read` object, but I think that's for a separate PR/Jira.
--
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] zhoufek commented on pull request #14971: [BEAM-12297] Add methods to PubsubIO for reading DynamicMessage
Posted by GitBox <gi...@apache.org>.
zhoufek commented on pull request #14971:
URL: https://github.com/apache/beam/pull/14971#issuecomment-858632912
Run Java PreCommit
--
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] zhoufek commented on a change in pull request #14971: [BEAM-12297] Add methods to PubsubIO for reading DynamicMessage
Posted by GitBox <gi...@apache.org>.
zhoufek commented on a change in pull request #14971:
URL: https://github.com/apache/beam/pull/14971#discussion_r649143507
##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIO.java
##########
@@ -483,6 +488,54 @@ public String toString() {
return Read.newBuilder(parsePayloadUsingCoder(coder)).setCoder(coder).build();
}
+ /**
+ * Returns a {@link PTransform} that continuously reads binary encoded protobuf messages for the
+ * type specified by {@code fullMessageName}.
+ *
+ * <p>This is primarily here for cases where the message type cannot be known at compile time. If
+ * it can be known, prefer {@link PubsubIO#readProtos(Class)}, as {@link DynamicMessage} tends to
+ * perform worse than concrete types.
+ *
+ * <p>Beam will infer a schema for the {@link DynamicMessage} schema. Note that some proto schema
+ * features are not supported by all sinks.
+ *
+ * @param domain The {@link ProtoDomain} that contains the target message and its dependencies.
+ * @param fullMessageName The full name of the message for lookup in {@code domain}.
+ */
+ @Experimental(Kind.SCHEMAS)
+ public static Read<DynamicMessage> readProtoDynamicMessage(
+ ProtoDomain domain, String fullMessageName) {
+ SerializableFunction<PubsubMessage, DynamicMessage> parser =
+ message -> {
+ try {
+ return DynamicMessage.parseFrom(
+ domain.getDescriptor(fullMessageName), message.getPayload());
+ } catch (InvalidProtocolBufferException e) {
+ throw new RuntimeException("Could not parse Pub/Sub message", e);
Review comment:
Opened BEAM-12474 for tracking effort on adding this. I'll probably get around to working on it myself soon.
--
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] zhoufek commented on pull request #14971: [BEAM-12297] Add methods to PubsubIO for reading DynamicMessage
Posted by GitBox <gi...@apache.org>.
zhoufek commented on pull request #14971:
URL: https://github.com/apache/beam/pull/14971#issuecomment-858626988
Run Java_Examples_Dataflow PreCommit
--
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] zhoufek commented on pull request #14971: [BEAM-12297] Add methods to PubsubIO for reading DynamicMessage
Posted by GitBox <gi...@apache.org>.
zhoufek commented on pull request #14971:
URL: https://github.com/apache/beam/pull/14971#issuecomment-859655065
--
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] zhoufek commented on a change in pull request #14971: [BEAM-12297] Add methods to PubsubIO for reading DynamicMessage
Posted by GitBox <gi...@apache.org>.
zhoufek commented on a change in pull request #14971:
URL: https://github.com/apache/beam/pull/14971#discussion_r649473094
##########
File path: sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIOTest.java
##########
@@ -416,6 +423,96 @@ public void after() throws IOException {
}
}
+ @Test
+ public void testProto() {
+ ProtoCoder<Primitive> coder = ProtoCoder.of(Primitive.class);
+ ImmutableList<Primitive> inputs =
+ ImmutableList.of(
+ Primitive.newBuilder().setPrimitiveInt32(42).build(),
+ Primitive.newBuilder().setPrimitiveBool(true).build(),
+ Primitive.newBuilder().setPrimitiveString("Hello, World!").build());
+ setupTestClient(inputs, coder);
+ PCollection<Primitive> read =
+ readPipeline.apply(
+ PubsubIO.readProtos(Primitive.class)
+ .fromSubscription(SUBSCRIPTION.getPath())
+ .withClock(CLOCK)
+ .withClientFactory(clientFactory));
+ PAssert.that(read).containsInAnyOrder(inputs);
+ readPipeline.run();
+ }
+
+ @Test
+ public void testProtoDynamicMessage() {
+ ProtoCoder<Primitive> coder = ProtoCoder.of(Primitive.class);
+ ImmutableList<Primitive> inputs =
+ ImmutableList.of(
+ Primitive.newBuilder().setPrimitiveInt32(42).build(),
+ Primitive.newBuilder().setPrimitiveBool(true).build(),
+ Primitive.newBuilder().setPrimitiveString("Hello, World!").build());
+ setupTestClient(inputs, coder);
+
+ ProtoDomain domain = ProtoDomain.buildFrom(Primitive.getDescriptor());
+ String name = Primitive.getDescriptor().getFullName();
+ PCollection<Primitive> read =
+ readPipeline
+ .apply(
+ PubsubIO.readProtoDynamicMessage(domain, name)
+ .fromSubscription(SUBSCRIPTION.getPath())
+ .withClock(CLOCK)
+ .withClientFactory(clientFactory))
+ // DynamicMessage doesn't work well with PAssert, but if the content can be successfully
+ // converted back into the original Primitive, then that should be good enough to
+ // consider it a successful read.
+ .apply(
+ "Return To Primitive",
+ MapElements.into(TypeDescriptor.of(Primitive.class))
+ .via(
+ (DynamicMessage message) -> {
Review comment:
I think that, since this is just a test, it would be acceptable to not catch the exception. Whether we fail by not catching an except or by asserting the failures are empty when they aren't, the end result is essentially the same.
Edit: But thanks for pointing out those exception methods for `MapElements`! They should be helpful.
--
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] zhoufek commented on pull request #14971: [BEAM-12297] Add methods to PubsubIO for reading DynamicMessage
Posted by GitBox <gi...@apache.org>.
zhoufek commented on pull request #14971:
URL: https://github.com/apache/beam/pull/14971#issuecomment-857855111
Squashed commits
--
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] zhoufek commented on pull request #14971: [BEAM-12297] Add methods to PubsubIO for reading DynamicMessage
Posted by GitBox <gi...@apache.org>.
zhoufek commented on pull request #14971:
URL: https://github.com/apache/beam/pull/14971#issuecomment-857875812
Run Java_Examples_Dataflow PreCommit
--
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] zhoufek commented on pull request #14971: [BEAM-12297] Add methods to PubsubIO for reading DynamicMessage
Posted by GitBox <gi...@apache.org>.
zhoufek commented on pull request #14971:
URL: https://github.com/apache/beam/pull/14971#issuecomment-858837651
Run Java PreCommit
--
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] zhoufek commented on pull request #14971: [BEAM-12297] Add methods to PubsubIO for reading DynamicMessage
Posted by GitBox <gi...@apache.org>.
zhoufek commented on pull request #14971:
URL: https://github.com/apache/beam/pull/14971#issuecomment-858800466
Java PreCommit seems flaky due to BEAM-5172. I guess we could just keep trying to run it until it succeeds.
I don't have permission to re-run the Direct Runner on MacOS, but it looks like something happened with Gradle that isn't related to this CL.
--
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] TheNeuralBit commented on a change in pull request #14971: [BEAM-12297] Add methods to PubsubIO for reading DynamicMessage
Posted by GitBox <gi...@apache.org>.
TheNeuralBit commented on a change in pull request #14971:
URL: https://github.com/apache/beam/pull/14971#discussion_r648441321
##########
File path: sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIOTest.java
##########
@@ -416,6 +423,96 @@ public void after() throws IOException {
}
}
+ @Test
+ public void testProto() {
+ ProtoCoder<Primitive> coder = ProtoCoder.of(Primitive.class);
+ ImmutableList<Primitive> inputs =
+ ImmutableList.of(
+ Primitive.newBuilder().setPrimitiveInt32(42).build(),
+ Primitive.newBuilder().setPrimitiveBool(true).build(),
+ Primitive.newBuilder().setPrimitiveString("Hello, World!").build());
+ setupTestClient(inputs, coder);
+ PCollection<Primitive> read =
+ readPipeline.apply(
+ PubsubIO.readProtos(Primitive.class)
+ .fromSubscription(SUBSCRIPTION.getPath())
+ .withClock(CLOCK)
+ .withClientFactory(clientFactory));
+ PAssert.that(read).containsInAnyOrder(inputs);
+ readPipeline.run();
+ }
+
+ @Test
+ public void testProtoDynamicMessage() {
+ ProtoCoder<Primitive> coder = ProtoCoder.of(Primitive.class);
+ ImmutableList<Primitive> inputs =
+ ImmutableList.of(
+ Primitive.newBuilder().setPrimitiveInt32(42).build(),
+ Primitive.newBuilder().setPrimitiveBool(true).build(),
+ Primitive.newBuilder().setPrimitiveString("Hello, World!").build());
+ setupTestClient(inputs, coder);
+
+ ProtoDomain domain = ProtoDomain.buildFrom(Primitive.getDescriptor());
+ String name = Primitive.getDescriptor().getFullName();
+ PCollection<Primitive> read =
+ readPipeline
+ .apply(
+ PubsubIO.readProtoDynamicMessage(domain, name)
+ .fromSubscription(SUBSCRIPTION.getPath())
+ .withClock(CLOCK)
+ .withClientFactory(clientFactory))
+ // DynamicMessage doesn't work well with PAssert, but if the content can be successfully
+ // converted back into the original Primitive, then that should be good enough to
+ // consider it a successful read.
+ .apply(
+ "Return To Primitive",
+ MapElements.into(TypeDescriptor.of(Primitive.class))
+ .via(
+ (DynamicMessage message) -> {
Review comment:
`MapElements` does have `exceptionsInto` and [`exceptionsVia`](https://beam.apache.org/releases/javadoc/2.29.0/org/apache/beam/sdk/transforms/MapElements.html#exceptionsVia-org.apache.beam.sdk.transforms.InferableFunction-) - would that work 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
[GitHub] [beam] zhoufek commented on pull request #14971: [BEAM-12297] Add methods to PubsubIO for reading DynamicMessage
Posted by GitBox <gi...@apache.org>.
zhoufek commented on pull request #14971:
URL: https://github.com/apache/beam/pull/14971#issuecomment-857723078
R: @kennknowles
--
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] zhoufek commented on pull request #14971: [BEAM-12297] Add methods to PubsubIO for reading DynamicMessage
Posted by GitBox <gi...@apache.org>.
zhoufek commented on pull request #14971:
URL: https://github.com/apache/beam/pull/14971#issuecomment-858945863
Looking over the test failures:
* FlinkRunner is a known flake: BEAM-10955
* ElasticsearchIO is also a known flake: BEAM-5172
I'm pretty sure the StreamingDataflowWorkerTest.testHotKeyLogging is also just a flake, since it hasn't happened before and is unrelated to the changes 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
[GitHub] [beam] TheNeuralBit edited a comment on pull request #14971: [BEAM-12297] Add methods to PubsubIO for reading DynamicMessage
Posted by GitBox <gi...@apache.org>.
TheNeuralBit edited a comment on pull request #14971:
URL: https://github.com/apache/beam/pull/14971#issuecomment-857811062
Cool, thanks @zhoufek!
> CC @TheNeuralBit since we recently discussed the issue of PubsubIO requiring a Java class and how we would improve that requirement for portability.
Note we already have [ProtoDynamicMessageSchema](https://github.com/apache/beam/blob/master/sdks/java/extensions/protobuf/src/main/java/org/apache/beam/sdk/extensions/protobuf/ProtoDynamicMessageSchema.java) which can make a `SchemaCoder<DynamicMesage>` given a proto descriptor. That combined with this PR could be a good avenue for making the pubsub TableProvider configurable via a proto descriptor rather than a proto class name. (CC: @apilloud @robinyqiu)
--
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] kennknowles commented on a change in pull request #14971: [BEAM-12297] Add methods to PubsubIO for reading DynamicMessage
Posted by GitBox <gi...@apache.org>.
kennknowles commented on a change in pull request #14971:
URL: https://github.com/apache/beam/pull/14971#discussion_r648422185
##########
File path: sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIOTest.java
##########
@@ -416,6 +423,96 @@ public void after() throws IOException {
}
}
+ @Test
+ public void testProto() {
+ ProtoCoder<Primitive> coder = ProtoCoder.of(Primitive.class);
+ ImmutableList<Primitive> inputs =
+ ImmutableList.of(
+ Primitive.newBuilder().setPrimitiveInt32(42).build(),
+ Primitive.newBuilder().setPrimitiveBool(true).build(),
+ Primitive.newBuilder().setPrimitiveString("Hello, World!").build());
+ setupTestClient(inputs, coder);
+ PCollection<Primitive> read =
+ readPipeline.apply(
+ PubsubIO.readProtos(Primitive.class)
+ .fromSubscription(SUBSCRIPTION.getPath())
+ .withClock(CLOCK)
+ .withClientFactory(clientFactory));
+ PAssert.that(read).containsInAnyOrder(inputs);
+ readPipeline.run();
+ }
+
+ @Test
+ public void testProtoDynamicMessage() {
+ ProtoCoder<Primitive> coder = ProtoCoder.of(Primitive.class);
+ ImmutableList<Primitive> inputs =
+ ImmutableList.of(
+ Primitive.newBuilder().setPrimitiveInt32(42).build(),
+ Primitive.newBuilder().setPrimitiveBool(true).build(),
+ Primitive.newBuilder().setPrimitiveString("Hello, World!").build());
+ setupTestClient(inputs, coder);
+
+ ProtoDomain domain = ProtoDomain.buildFrom(Primitive.getDescriptor());
+ String name = Primitive.getDescriptor().getFullName();
+ PCollection<Primitive> read =
+ readPipeline
+ .apply(
+ PubsubIO.readProtoDynamicMessage(domain, name)
+ .fromSubscription(SUBSCRIPTION.getPath())
+ .withClock(CLOCK)
+ .withClientFactory(clientFactory))
+ // DynamicMessage doesn't work well with PAssert, but if the content can be successfully
+ // converted back into the original Primitive, then that should be good enough to
+ // consider it a successful read.
+ .apply(
+ "Return To Primitive",
+ MapElements.into(TypeDescriptor.of(Primitive.class))
+ .via(
+ (DynamicMessage message) -> {
Review comment:
This boilerplate makes me think that `MapElements` could also benefit from a variant with an output for failed messages. (not actionable for your PR, just a comment)
##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubIO.java
##########
@@ -483,6 +488,54 @@ public String toString() {
return Read.newBuilder(parsePayloadUsingCoder(coder)).setCoder(coder).build();
}
+ /**
+ * Returns a {@link PTransform} that continuously reads binary encoded protobuf messages for the
+ * type specified by {@code fullMessageName}.
+ *
+ * <p>This is primarily here for cases where the message type cannot be known at compile time. If
+ * it can be known, prefer {@link PubsubIO#readProtos(Class)}, as {@link DynamicMessage} tends to
+ * perform worse than concrete types.
+ *
+ * <p>Beam will infer a schema for the {@link DynamicMessage} schema. Note that some proto schema
+ * features are not supported by all sinks.
+ *
+ * @param domain The {@link ProtoDomain} that contains the target message and its dependencies.
+ * @param fullMessageName The full name of the message for lookup in {@code domain}.
+ */
+ @Experimental(Kind.SCHEMAS)
+ public static Read<DynamicMessage> readProtoDynamicMessage(
+ ProtoDomain domain, String fullMessageName) {
+ SerializableFunction<PubsubMessage, DynamicMessage> parser =
+ message -> {
+ try {
+ return DynamicMessage.parseFrom(
+ domain.getDescriptor(fullMessageName), message.getPayload());
+ } catch (InvalidProtocolBufferException e) {
+ throw new RuntimeException("Could not parse Pub/Sub message", e);
Review comment:
Does PubsubIO have a dead-letter capability that will catch this error and output unparseable messages to some secondary output? Can users turn that capability on after the `Read` transform is returned by this method?
(I'm not opening up the code and reading, because I just want to get your review back to you and I'm sure you know the answer and will do the right thing)
--
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] zhoufek commented on pull request #14971: [BEAM-12297] Add methods to PubsubIO for reading DynamicMessage
Posted by GitBox <gi...@apache.org>.
zhoufek commented on pull request #14971:
URL: https://github.com/apache/beam/pull/14971#issuecomment-857897200
Run Java PreCommit
--
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