[BEAM-14036] Read Configuration for Pub/Sub SchemaTransform #17730

damondouglas · 2022-05-22T00:54:22Z

This PR is the first of four planned PRs. It contributes to BEAM-14036 by implementing a configuration for reading from Pub/Sub. There are no dependencies on this PR.

I would like to request the following to review this PR.
R: @angoenka
R: @chamikaramj

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

asf-ci · 2022-05-22T00:54:24Z

Can one of the admins verify this patch?

asf-ci · 2022-05-22T00:54:24Z

Can one of the admins verify this patch?

asf-ci · 2022-05-22T00:54:24Z

Can one of the admins verify this patch?

damondouglas · 2022-05-23T16:09:33Z

Run Java PreCommit

chamikaramj

Thanks.

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

chamikaramj · 2022-05-24T18:55:50Z

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

+   * letter queue topic string.
+   */
+  @Nullable
+  public abstract String getDeadLetterQueue();


Pls change to getDeadLetterTopic to be consistent with the Pub/Sub read transform.

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

Line 865 in d9436c4

public Read<T> withDeadLetterTopic(String deadLetterTopic) {

@chamikaramj (cc: @angoenka ) Thank you again for reviewing. May we consider leaving the name getDeadLetterQueue?

In a previous use of AutoValueSchema with AutoValue in a different project, I observed it needed the getters to be named as get in order for the serialization to work. For example, I had a property called fooName. When I named the getter fooName(), the return value of fooName() was null when invoked in the context of a DoFn. However, when I changed the getter to getFooName() the return value was what I expected. I am not sure if my observation is still valid.

Adding to supporting get instead of with, the design goals of the configuration class are to hold data needed by its corresponding SchemaProvider. The method is not doing any action implied by the with preposition. The getter is simply getting data.

I see, I think the confusion here is that this property does not correspond to that PubSubIO.getDeadLetterTopic I referenced by maps to the dlq property below.

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

Line 81 in 408664b

public abstract boolean useDlq();

Is that correct ?

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

chamikaramj · 2022-05-24T18:58:52Z

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

+   * {@link org.apache.beam.sdk.schemas.io.payloads.PayloadSerializers}.
+   */
+  @Nullable
+  public abstract String getFormat();


I don't see format, protoClass, thriftClass attributes in the original Read config.

@chamikaramj (cc: @angoenka )
I saw https://github.com/apache/beam/blob/master/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubSchemaIOProvider.java#L128. Additionally, the original Provider's https://github.com/apache/beam/blob/master/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubSchemaIOProvider.java#L209 calls the config::serializer method. In said serializer method, the format, protoClass, and thriftClass attributes are referenced.

These again map to PubSubSchemaIO not the original PubSubIO. Is that plan here to make Pub/Sub SchemaTransform just consistent with existing Pub/Sub SchemaIO ? I'm not sure.

@TheNeuralBit @dpcollins-google WDYT ? Should we just be consistent with Pub/Sub SchemaIO implementation here or should SchemaTransform be a different design ?

These again map to PubSubSchemaIO not the original PubSubIO. Is that plan here to make Pub/Sub SchemaTransform just consistent with existing Pub/Sub SchemaIO ? I'm not sure.

@TheNeuralBit @dpcollins-google WDYT ? Should we just be consistent with Pub/Sub SchemaIO implementation here or should SchemaTransform be a different design ?

The plan and replicate like-for-like SchemaIO questions are critical and blocking design decisions that relates to this thread as well. I will hold off on any changes to this PR until we get the feedback needed. Thank you again.

codecov · 2022-05-25T16:38:50Z

Codecov Report

Merging #17730 (408664b) into master (0c9cf43) will increase coverage by 0.01%.
The diff coverage is n/a.

❗ Current head 408664b differs from pull request most recent head 3099d60. Consider uploading reports for the commit 3099d60 to get more accurate results

@@            Coverage Diff             @@
##           master   #17730      +/-   ##
==========================================
+ Coverage   74.00%   74.01%   +0.01%     
==========================================
  Files         696      698       +2     
  Lines       91851    92224     +373     
==========================================
+ Hits        67975    68263     +288     
- Misses      22627    22710      +83     
- Partials     1249     1251       +2

Flag	Coverage Δ
go	`50.93% <ø> (+0.48%)`	⬆️
python	`83.60% <ø> (-0.16%)`	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files	Coverage Δ
...ache_beam/runners/dataflow/dataflow_job_service.py	`50.00% <0.00%> (-12.17%)`	⬇️
sdks/go/pkg/beam/io/textio/textio.go	`55.15% <0.00%> (-10.42%)`	⬇️
...s/interactive/dataproc/dataproc_cluster_manager.py	`77.41% <0.00%> (-6.80%)`	⬇️
.../apache_beam/runners/interactive/dataproc/types.py	`96.55% <0.00%> (-3.45%)`	⬇️
sdks/python/apache_beam/dataframe/io.py	`88.78% <0.00%> (-3.26%)`	⬇️
sdks/go/pkg/beam/pardo.go	`47.41% <0.00%> (-3.03%)`	⬇️
...eam/runners/portability/fn_api_runner/fn_runner.py	`87.51% <0.00%> (-2.50%)`	⬇️
...nners/portability/fn_api_runner/worker_handlers.py	`77.89% <0.00%> (-1.45%)`	⬇️
sdks/python/apache_beam/internal/dill_pickler.py	`84.89% <0.00%> (-1.44%)`	⬇️
...ks/go/pkg/beam/runners/dataflow/dataflowlib/job.go	`21.55% <0.00%> (-1.30%)`	⬇️
... and 45 more

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 0c9cf43...3099d60. Read the comment docs.

chamikaramj

Thanks.

chamikaramj · 2022-06-10T02:23:03Z

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

+   * letter queue topic string.
+   */
+  @Nullable
+  public abstract String getDeadLetterQueue();


I see, I think the confusion here is that this property does not correspond to that PubSubIO.getDeadLetterTopic I referenced by maps to the dlq property below.

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

Line 81 in 408664b

public abstract boolean useDlq();

Is that correct ?

chamikaramj · 2022-06-10T02:43:04Z

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

+   * {@link org.apache.beam.sdk.schemas.io.payloads.PayloadSerializers}.
+   */
+  @Nullable
+  public abstract String getFormat();


These again map to PubSubSchemaIO not the original PubSubIO. Is that plan here to make Pub/Sub SchemaTransform just consistent with existing Pub/Sub SchemaIO ? I'm not sure.

@TheNeuralBit @dpcollins-google WDYT ? Should we just be consistent with Pub/Sub SchemaIO implementation here or should SchemaTransform be a different design ?

damondouglas · 2022-06-14T16:22:43Z

@chamikaramj , @TheNeuralBit , @dpcollins-google , @pabloem If I don't see any activity on this PR in the next couple days, I will go ahead and close it to keep the PR space clean in this project.

chamikaramj · 2022-06-14T16:32:14Z

I think we can just be compatible with existing SchemaIO config for now and update in the future if needed.

chamikaramj

LGTM.

…7730) * Read Configuration for Pub/Sub SchemaTransform * Add idAttribute to Read Configuration * Add Experimental annotation/remove SuppressWarning

Read Configuration for Pub/Sub SchemaTransform

0c0ab4b

github-actions bot added gcp io java labels May 22, 2022

chamikaramj reviewed May 24, 2022

View reviewed changes

damondouglas added 2 commits June 2, 2022 16:53

Add idAttribute to Read Configuration

6c2798d

Add Experimental annotation/remove SuppressWarning

3099d60

damondouglas mentioned this pull request Jun 7, 2022

Convert Pub/Sub Schema IO to SchemaTransform #21412

Closed

damondouglas requested a review from chamikaramj June 8, 2022 15:31

chamikaramj reviewed Jun 10, 2022

View reviewed changes

chamikaramj approved these changes Jun 14, 2022

View reviewed changes

chamikaramj merged commit 6cfcc9d into apache:master Jun 14, 2022

damondouglas deleted the BEAM-14036-PubSubSchemaIOProviderReadConfig branch June 30, 2022 23:18

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[BEAM-14036] Read Configuration for Pub/Sub SchemaTransform #17730

[BEAM-14036] Read Configuration for Pub/Sub SchemaTransform #17730

damondouglas commented May 22, 2022 •

edited

asf-ci commented May 22, 2022

asf-ci commented May 22, 2022

asf-ci commented May 22, 2022

damondouglas commented May 23, 2022

chamikaramj left a comment

chamikaramj May 24, 2022

damondouglas Jun 2, 2022 •

edited

chamikaramj Jun 10, 2022

chamikaramj May 24, 2022

damondouglas Jun 2, 2022 •

edited

chamikaramj Jun 10, 2022

damondouglas Jun 10, 2022 •

edited

codecov bot commented May 25, 2022 •

edited

chamikaramj left a comment

chamikaramj Jun 10, 2022

chamikaramj Jun 10, 2022

damondouglas commented Jun 14, 2022

chamikaramj commented Jun 14, 2022

chamikaramj left a comment

[BEAM-14036] Read Configuration for Pub/Sub SchemaTransform #17730

[BEAM-14036] Read Configuration for Pub/Sub SchemaTransform #17730

Conversation

damondouglas commented May 22, 2022 • edited

GitHub Actions Tests Status (on master branch)

asf-ci commented May 22, 2022

asf-ci commented May 22, 2022

asf-ci commented May 22, 2022

damondouglas commented May 23, 2022

chamikaramj left a comment

Choose a reason for hiding this comment

chamikaramj May 24, 2022

Choose a reason for hiding this comment

damondouglas Jun 2, 2022 • edited

Choose a reason for hiding this comment

chamikaramj Jun 10, 2022

Choose a reason for hiding this comment

chamikaramj May 24, 2022

Choose a reason for hiding this comment

damondouglas Jun 2, 2022 • edited

Choose a reason for hiding this comment

chamikaramj Jun 10, 2022

Choose a reason for hiding this comment

damondouglas Jun 10, 2022 • edited

Choose a reason for hiding this comment

codecov bot commented May 25, 2022 • edited

Codecov Report

chamikaramj left a comment

Choose a reason for hiding this comment

chamikaramj Jun 10, 2022

Choose a reason for hiding this comment

chamikaramj Jun 10, 2022

Choose a reason for hiding this comment

damondouglas commented Jun 14, 2022

chamikaramj commented Jun 14, 2022

chamikaramj left a comment

Choose a reason for hiding this comment

damondouglas commented May 22, 2022 •

edited

damondouglas Jun 2, 2022 •

edited

damondouglas Jun 2, 2022 •

edited

damondouglas Jun 10, 2022 •

edited

codecov bot commented May 25, 2022 •

edited