You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "Dippatel98 (via GitHub)" <gi...@apache.org> on 2023/07/05 19:27:29 UTC
[GitHub] [beam] Dippatel98 opened a new pull request, #27366: Moving Pubsub Transforms to beam
Dippatel98 opened a new pull request, #27366:
URL: https://github.com/apache/beam/pull/27366
Moving the PubsubReadTransformProvider and PubsubWriteTransformProvider to beam. Removing PubsubSchemaTransformMessageToRowFactory and the corresponding test as that is no longer required by the new transforms. Removing a few other tests which were testing the old transforms and are no longer usable.
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] 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](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/get-started-contributing/#make-the-reviewers-job-easier).
To check the build health, please visit [https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md](https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md)
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)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Go+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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] Dippatel98 commented on a diff in pull request #27366: Moving Pubsub Transforms to beam
Posted by "Dippatel98 (via GitHub)" <gi...@apache.org>.
Dippatel98 commented on code in PR #27366:
URL: https://github.com/apache/beam/pull/27366#discussion_r1253600659
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubWriteSchemaTransformConfiguration.java:
##########
@@ -32,179 +32,29 @@
@DefaultSchema(AutoValueSchema.class)
@AutoValue
public abstract class PubsubWriteSchemaTransformConfiguration {
+ @SchemaFieldDescription(
+ "The encoding format for the data stored in Pubsub. Valid options are: "
+ + PubsubWriteSchemaTransformProvider.VALID_FORMATS_STR)
+ public abstract String getFormat();
- public static final String DEFAULT_TIMESTAMP_ATTRIBUTE = "event_timestamp";
+ @SchemaFieldDescription(
+ "The name of the topic to write data to. " + "Format: projects/${PROJECT}/topics/${TOPIC}")
+ public abstract String getTopic();
public static Builder builder() {
return new AutoValue_PubsubWriteSchemaTransformConfiguration.Builder();
}
- public static TargetConfiguration.Builder targetConfigurationBuilder() {
- return new AutoValue_PubsubWriteSchemaTransformConfiguration_TargetConfiguration.Builder()
- .setTimestampAttributeKey(DEFAULT_TIMESTAMP_ATTRIBUTE);
- }
-
- public static SourceConfiguration.Builder sourceConfigurationBuilder() {
- return new AutoValue_PubsubWriteSchemaTransformConfiguration_SourceConfiguration.Builder();
- }
-
- /**
- * Configuration details of the source {@link org.apache.beam.sdk.values.Row} {@link
- * org.apache.beam.sdk.schemas.Schema}.
- */
- @Nullable
- public abstract SourceConfiguration getSource();
-
- /** Configuration details of the target {@link PubsubMessage}. */
- public abstract TargetConfiguration getTarget();
-
- /**
- * The topic to which to write Pub/Sub messages.
- *
- * <p>See {@link PubsubIO.PubsubTopic#fromPath(String)} for more details on the format of the
- * topic string.
- */
- public abstract String getTopic();
-
- /**
- * The expected format of the Pub/Sub message.
- *
- * <p>Used to retrieve the {@link org.apache.beam.sdk.schemas.io.payloads.PayloadSerializer} from
- * {@link org.apache.beam.sdk.schemas.io.payloads.PayloadSerializers}. See list of supported
- * values by invoking {@link org.apache.beam.sdk.schemas.io.Providers#loadProviders(Class)}.
- *
- * <pre>{@code Providers.loadProviders(PayloadSerializer.class).keySet()}</pre>
- */
- @Nullable
- public abstract String getFormat();
-
- /**
- * When writing to Cloud Pub/Sub where unique record identifiers are provided as Pub/Sub message
- * attributes, specifies the name of the attribute containing the unique identifier.
- */
- @Nullable
- public abstract String getIdAttribute();
-
- /** Builder for {@link PubsubWriteSchemaTransformConfiguration}. */
@AutoValue.Builder
public abstract static class Builder {
+ public abstract Builder setFormat(String format);
- /**
- * Configuration details of the source {@link org.apache.beam.sdk.values.Row} {@link
- * org.apache.beam.sdk.schemas.Schema}.
- */
- public abstract Builder setSource(SourceConfiguration value);
-
- /** Configuration details of the target {@link PubsubMessage}. */
- public abstract Builder setTarget(TargetConfiguration value);
-
- /**
- * The topic to which to write Pub/Sub messages.
- *
- * <p>See {@link PubsubIO.PubsubTopic#fromPath(String)} for more details on the format of the
- * topic string.
- */
- public abstract Builder setTopic(String value);
-
- /**
- * The expected format of the Pub/Sub message.
- *
- * <p>Used to retrieve the {@link org.apache.beam.sdk.schemas.io.payloads.PayloadSerializer}
- * from {@link org.apache.beam.sdk.schemas.io.payloads.PayloadSerializers}. See list of
- * supported values by invoking {@link
- * org.apache.beam.sdk.schemas.io.Providers#loadProviders(Class)}.
- *
- * <pre>{@code Providers.loadProviders(PayloadSerializer.class).keySet()}</pre>
- */
- public abstract Builder setFormat(String value);
-
- /**
- * When reading from Cloud Pub/Sub where unique record identifiers are provided as Pub/Sub
- * message attributes, specifies the name of the attribute containing the unique identifier.
- */
- public abstract Builder setIdAttribute(String value);
+ public abstract Builder setTopic(String topic);
public abstract PubsubWriteSchemaTransformConfiguration build();
}
- @DefaultSchema(AutoValueSchema.class)
- @AutoValue
- public abstract static class SourceConfiguration {
- /**
- * The attributes field name of the source {@link org.apache.beam.sdk.values.Row}. {@link
- * org.apache.beam.sdk.schemas.Schema.FieldType} must be a <code>Map<String, String>
- * </code>
- */
- @Nullable
- public abstract String getAttributesFieldName();
-
- /**
- * The timestamp field name of the source {@link org.apache.beam.sdk.values.Row}. {@link
- * org.apache.beam.sdk.schemas.Schema.FieldType} must be a {@link
- * org.apache.beam.sdk.schemas.Schema.FieldType#DATETIME}.
- */
- @Nullable
- public abstract String getTimestampFieldName();
-
- /**
- * The payload field name of the source {@link org.apache.beam.sdk.values.Row}. {@link
- * org.apache.beam.sdk.schemas.Schema.FieldType} must be either {@link
- * org.apache.beam.sdk.schemas.Schema.FieldType#BYTES} or a {@link
- * org.apache.beam.sdk.values.Row}. If null, payload serialized from user fields other than
- * attributes. Not compatible with other payload intended fields.
- */
- @Nullable
- public abstract String getPayloadFieldName();
-
- @AutoValue.Builder
- public abstract static class Builder {
- /**
- * The attributes field name of the source {@link org.apache.beam.sdk.values.Row}. {@link
- * org.apache.beam.sdk.schemas.Schema.FieldType} must be a <code>Map<String, String>
- * </code>
- */
- public abstract Builder setAttributesFieldName(String value);
-
- /**
- * The timestamp field name of the source {@link org.apache.beam.sdk.values.Row}. {@link
- * org.apache.beam.sdk.schemas.Schema.FieldType} must be a {@link
- * org.apache.beam.sdk.schemas.Schema.FieldType#DATETIME}.
- */
- public abstract Builder setTimestampFieldName(String value);
-
- /**
- * The payload field name of the source {@link org.apache.beam.sdk.values.Row}. {@link
- * org.apache.beam.sdk.schemas.Schema.FieldType} must be either {@link
- * org.apache.beam.sdk.schemas.Schema.FieldType#BYTES} or a {@link
- * org.apache.beam.sdk.values.Row}. If null, payload serialized from user fields other than
- * attributes. Not compatible with other payload intended fields.
- */
- public abstract Builder setPayloadFieldName(String value);
-
- public abstract SourceConfiguration build();
- }
- }
-
- @DefaultSchema(AutoValueSchema.class)
- @AutoValue
- public abstract static class TargetConfiguration {
-
- /**
- * The attribute key to assign the {@link PubsubMessage} stringified timestamp value. {@link
- * #builder()} method defaults value to {@link #DEFAULT_TIMESTAMP_ATTRIBUTE}.
- */
- public abstract String getTimestampAttributeKey();
-
- @AutoValue.Builder
- public abstract static class Builder {
-
- /**
- * The attribute key to assign the {@link PubsubMessage} stringified timestamp value. Defaults
- * to {@link #DEFAULT_TIMESTAMP_ATTRIBUTE}.
- */
- public abstract Builder setTimestampAttributeKey(String value);
-
- public abstract TargetConfiguration build();
- }
- }
+ // public static PubsubWriteSchemaTransformConfiguration create(String format, String topic) {
Review Comment:
Ah, I missed those
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] Dippatel98 commented on a diff in pull request #27366: Moving Pubsub Transforms to beam
Posted by "Dippatel98 (via GitHub)" <gi...@apache.org>.
Dippatel98 commented on code in PR #27366:
URL: https://github.com/apache/beam/pull/27366#discussion_r1253614620
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubWriteSchemaTransformConfiguration.java:
##########
@@ -32,179 +32,29 @@
@DefaultSchema(AutoValueSchema.class)
@AutoValue
public abstract class PubsubWriteSchemaTransformConfiguration {
+ @SchemaFieldDescription(
+ "The encoding format for the data stored in Pubsub. Valid options are: "
+ + PubsubWriteSchemaTransformProvider.VALID_FORMATS_STR)
+ public abstract String getFormat();
- public static final String DEFAULT_TIMESTAMP_ATTRIBUTE = "event_timestamp";
+ @SchemaFieldDescription(
+ "The name of the topic to write data to. " + "Format: projects/${PROJECT}/topics/${TOPIC}")
+ public abstract String getTopic();
public static Builder builder() {
return new AutoValue_PubsubWriteSchemaTransformConfiguration.Builder();
}
- public static TargetConfiguration.Builder targetConfigurationBuilder() {
- return new AutoValue_PubsubWriteSchemaTransformConfiguration_TargetConfiguration.Builder()
- .setTimestampAttributeKey(DEFAULT_TIMESTAMP_ATTRIBUTE);
- }
-
- public static SourceConfiguration.Builder sourceConfigurationBuilder() {
- return new AutoValue_PubsubWriteSchemaTransformConfiguration_SourceConfiguration.Builder();
- }
-
- /**
- * Configuration details of the source {@link org.apache.beam.sdk.values.Row} {@link
- * org.apache.beam.sdk.schemas.Schema}.
- */
- @Nullable
- public abstract SourceConfiguration getSource();
-
- /** Configuration details of the target {@link PubsubMessage}. */
- public abstract TargetConfiguration getTarget();
-
- /**
- * The topic to which to write Pub/Sub messages.
- *
- * <p>See {@link PubsubIO.PubsubTopic#fromPath(String)} for more details on the format of the
- * topic string.
- */
- public abstract String getTopic();
-
- /**
- * The expected format of the Pub/Sub message.
- *
- * <p>Used to retrieve the {@link org.apache.beam.sdk.schemas.io.payloads.PayloadSerializer} from
- * {@link org.apache.beam.sdk.schemas.io.payloads.PayloadSerializers}. See list of supported
- * values by invoking {@link org.apache.beam.sdk.schemas.io.Providers#loadProviders(Class)}.
- *
- * <pre>{@code Providers.loadProviders(PayloadSerializer.class).keySet()}</pre>
- */
- @Nullable
- public abstract String getFormat();
-
- /**
- * When writing to Cloud Pub/Sub where unique record identifiers are provided as Pub/Sub message
- * attributes, specifies the name of the attribute containing the unique identifier.
- */
- @Nullable
- public abstract String getIdAttribute();
-
- /** Builder for {@link PubsubWriteSchemaTransformConfiguration}. */
@AutoValue.Builder
public abstract static class Builder {
+ public abstract Builder setFormat(String format);
- /**
- * Configuration details of the source {@link org.apache.beam.sdk.values.Row} {@link
- * org.apache.beam.sdk.schemas.Schema}.
- */
- public abstract Builder setSource(SourceConfiguration value);
-
- /** Configuration details of the target {@link PubsubMessage}. */
- public abstract Builder setTarget(TargetConfiguration value);
-
- /**
- * The topic to which to write Pub/Sub messages.
- *
- * <p>See {@link PubsubIO.PubsubTopic#fromPath(String)} for more details on the format of the
- * topic string.
- */
- public abstract Builder setTopic(String value);
-
- /**
- * The expected format of the Pub/Sub message.
- *
- * <p>Used to retrieve the {@link org.apache.beam.sdk.schemas.io.payloads.PayloadSerializer}
- * from {@link org.apache.beam.sdk.schemas.io.payloads.PayloadSerializers}. See list of
- * supported values by invoking {@link
- * org.apache.beam.sdk.schemas.io.Providers#loadProviders(Class)}.
- *
- * <pre>{@code Providers.loadProviders(PayloadSerializer.class).keySet()}</pre>
- */
- public abstract Builder setFormat(String value);
-
- /**
- * When reading from Cloud Pub/Sub where unique record identifiers are provided as Pub/Sub
- * message attributes, specifies the name of the attribute containing the unique identifier.
- */
- public abstract Builder setIdAttribute(String value);
+ public abstract Builder setTopic(String topic);
public abstract PubsubWriteSchemaTransformConfiguration build();
}
- @DefaultSchema(AutoValueSchema.class)
- @AutoValue
- public abstract static class SourceConfiguration {
- /**
- * The attributes field name of the source {@link org.apache.beam.sdk.values.Row}. {@link
- * org.apache.beam.sdk.schemas.Schema.FieldType} must be a <code>Map<String, String>
- * </code>
- */
- @Nullable
- public abstract String getAttributesFieldName();
-
- /**
- * The timestamp field name of the source {@link org.apache.beam.sdk.values.Row}. {@link
- * org.apache.beam.sdk.schemas.Schema.FieldType} must be a {@link
- * org.apache.beam.sdk.schemas.Schema.FieldType#DATETIME}.
- */
- @Nullable
- public abstract String getTimestampFieldName();
-
- /**
- * The payload field name of the source {@link org.apache.beam.sdk.values.Row}. {@link
- * org.apache.beam.sdk.schemas.Schema.FieldType} must be either {@link
- * org.apache.beam.sdk.schemas.Schema.FieldType#BYTES} or a {@link
- * org.apache.beam.sdk.values.Row}. If null, payload serialized from user fields other than
- * attributes. Not compatible with other payload intended fields.
- */
- @Nullable
- public abstract String getPayloadFieldName();
-
- @AutoValue.Builder
- public abstract static class Builder {
- /**
- * The attributes field name of the source {@link org.apache.beam.sdk.values.Row}. {@link
- * org.apache.beam.sdk.schemas.Schema.FieldType} must be a <code>Map<String, String>
- * </code>
- */
- public abstract Builder setAttributesFieldName(String value);
-
- /**
- * The timestamp field name of the source {@link org.apache.beam.sdk.values.Row}. {@link
- * org.apache.beam.sdk.schemas.Schema.FieldType} must be a {@link
- * org.apache.beam.sdk.schemas.Schema.FieldType#DATETIME}.
- */
- public abstract Builder setTimestampFieldName(String value);
-
- /**
- * The payload field name of the source {@link org.apache.beam.sdk.values.Row}. {@link
- * org.apache.beam.sdk.schemas.Schema.FieldType} must be either {@link
- * org.apache.beam.sdk.schemas.Schema.FieldType#BYTES} or a {@link
- * org.apache.beam.sdk.values.Row}. If null, payload serialized from user fields other than
- * attributes. Not compatible with other payload intended fields.
- */
- public abstract Builder setPayloadFieldName(String value);
-
- public abstract SourceConfiguration build();
- }
- }
-
- @DefaultSchema(AutoValueSchema.class)
- @AutoValue
- public abstract static class TargetConfiguration {
-
- /**
- * The attribute key to assign the {@link PubsubMessage} stringified timestamp value. {@link
- * #builder()} method defaults value to {@link #DEFAULT_TIMESTAMP_ATTRIBUTE}.
- */
- public abstract String getTimestampAttributeKey();
-
- @AutoValue.Builder
- public abstract static class Builder {
-
- /**
- * The attribute key to assign the {@link PubsubMessage} stringified timestamp value. Defaults
- * to {@link #DEFAULT_TIMESTAMP_ATTRIBUTE}.
- */
- public abstract Builder setTimestampAttributeKey(String value);
-
- public abstract TargetConfiguration build();
- }
- }
+ // public static PubsubWriteSchemaTransformConfiguration create(String format, String topic) {
Review Comment:
Done
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] codecov[bot] commented on pull request #27366: Moving Pubsub Transforms to beam
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #27366:
URL: https://github.com/apache/beam/pull/27366#issuecomment-1622400146
## [Codecov](https://app.codecov.io/gh/apache/beam/pull/27366?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
> Merging [#27366](https://app.codecov.io/gh/apache/beam/pull/27366?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) (ce031b1) into [master](https://app.codecov.io/gh/apache/beam/commit/9a11e28ce79e3b243a13fbf148f2ba26b8c14107?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) (9a11e28) will **increase** coverage by `0.00%`.
> The diff coverage is `n/a`.
```diff
@@ Coverage Diff @@
## master #27366 +/- ##
=======================================
Coverage 71.24% 71.24%
=======================================
Files 854 854
Lines 103736 103736
=======================================
+ Hits 73907 73911 +4
+ Misses 28281 28277 -4
Partials 1548 1548
```
| Flag | Coverage Δ | |
|---|---|---|
| python | `80.59% <ø> (+<0.01%)` | :arrow_up: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
[see 8 files with indirect coverage changes](https://app.codecov.io/gh/apache/beam/pull/27366/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
:mega: We’re building smart automated test selection to slash your CI/CD build times. [Learn more](https://about.codecov.io/iterative-testing/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] Dippatel98 commented on pull request #27366: Moving Pubsub Transforms to beam
Posted by "Dippatel98 (via GitHub)" <gi...@apache.org>.
Dippatel98 commented on PR #27366:
URL: https://github.com/apache/beam/pull/27366#issuecomment-1638748774
@johnjcasey If there are no more concerns here, can we merge 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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ahmedabu98 commented on pull request #27366: Moving Pubsub Transforms to beam
Posted by "ahmedabu98 (via GitHub)" <gi...@apache.org>.
ahmedabu98 commented on PR #27366:
URL: https://github.com/apache/beam/pull/27366#issuecomment-1626047949
Also tagging @damondouglas for visibility as he worked on the existing pubsub schematransforms
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] manavgarg commented on a diff in pull request #27366: Moving Pubsub Transforms to beam
Posted by "manavgarg (via GitHub)" <gi...@apache.org>.
manavgarg commented on code in PR #27366:
URL: https://github.com/apache/beam/pull/27366#discussion_r1253589655
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubWriteSchemaTransformConfiguration.java:
##########
@@ -32,179 +32,29 @@
@DefaultSchema(AutoValueSchema.class)
@AutoValue
public abstract class PubsubWriteSchemaTransformConfiguration {
+ @SchemaFieldDescription(
+ "The encoding format for the data stored in Pubsub. Valid options are: "
+ + PubsubWriteSchemaTransformProvider.VALID_FORMATS_STR)
+ public abstract String getFormat();
- public static final String DEFAULT_TIMESTAMP_ATTRIBUTE = "event_timestamp";
+ @SchemaFieldDescription(
+ "The name of the topic to write data to. " + "Format: projects/${PROJECT}/topics/${TOPIC}")
+ public abstract String getTopic();
public static Builder builder() {
return new AutoValue_PubsubWriteSchemaTransformConfiguration.Builder();
}
- public static TargetConfiguration.Builder targetConfigurationBuilder() {
- return new AutoValue_PubsubWriteSchemaTransformConfiguration_TargetConfiguration.Builder()
- .setTimestampAttributeKey(DEFAULT_TIMESTAMP_ATTRIBUTE);
- }
-
- public static SourceConfiguration.Builder sourceConfigurationBuilder() {
- return new AutoValue_PubsubWriteSchemaTransformConfiguration_SourceConfiguration.Builder();
- }
-
- /**
- * Configuration details of the source {@link org.apache.beam.sdk.values.Row} {@link
- * org.apache.beam.sdk.schemas.Schema}.
- */
- @Nullable
- public abstract SourceConfiguration getSource();
-
- /** Configuration details of the target {@link PubsubMessage}. */
- public abstract TargetConfiguration getTarget();
-
- /**
- * The topic to which to write Pub/Sub messages.
- *
- * <p>See {@link PubsubIO.PubsubTopic#fromPath(String)} for more details on the format of the
- * topic string.
- */
- public abstract String getTopic();
-
- /**
- * The expected format of the Pub/Sub message.
- *
- * <p>Used to retrieve the {@link org.apache.beam.sdk.schemas.io.payloads.PayloadSerializer} from
- * {@link org.apache.beam.sdk.schemas.io.payloads.PayloadSerializers}. See list of supported
- * values by invoking {@link org.apache.beam.sdk.schemas.io.Providers#loadProviders(Class)}.
- *
- * <pre>{@code Providers.loadProviders(PayloadSerializer.class).keySet()}</pre>
- */
- @Nullable
- public abstract String getFormat();
-
- /**
- * When writing to Cloud Pub/Sub where unique record identifiers are provided as Pub/Sub message
- * attributes, specifies the name of the attribute containing the unique identifier.
- */
- @Nullable
- public abstract String getIdAttribute();
-
- /** Builder for {@link PubsubWriteSchemaTransformConfiguration}. */
@AutoValue.Builder
public abstract static class Builder {
+ public abstract Builder setFormat(String format);
- /**
- * Configuration details of the source {@link org.apache.beam.sdk.values.Row} {@link
- * org.apache.beam.sdk.schemas.Schema}.
- */
- public abstract Builder setSource(SourceConfiguration value);
-
- /** Configuration details of the target {@link PubsubMessage}. */
- public abstract Builder setTarget(TargetConfiguration value);
-
- /**
- * The topic to which to write Pub/Sub messages.
- *
- * <p>See {@link PubsubIO.PubsubTopic#fromPath(String)} for more details on the format of the
- * topic string.
- */
- public abstract Builder setTopic(String value);
-
- /**
- * The expected format of the Pub/Sub message.
- *
- * <p>Used to retrieve the {@link org.apache.beam.sdk.schemas.io.payloads.PayloadSerializer}
- * from {@link org.apache.beam.sdk.schemas.io.payloads.PayloadSerializers}. See list of
- * supported values by invoking {@link
- * org.apache.beam.sdk.schemas.io.Providers#loadProviders(Class)}.
- *
- * <pre>{@code Providers.loadProviders(PayloadSerializer.class).keySet()}</pre>
- */
- public abstract Builder setFormat(String value);
-
- /**
- * When reading from Cloud Pub/Sub where unique record identifiers are provided as Pub/Sub
- * message attributes, specifies the name of the attribute containing the unique identifier.
- */
- public abstract Builder setIdAttribute(String value);
+ public abstract Builder setTopic(String topic);
public abstract PubsubWriteSchemaTransformConfiguration build();
}
- @DefaultSchema(AutoValueSchema.class)
- @AutoValue
- public abstract static class SourceConfiguration {
- /**
- * The attributes field name of the source {@link org.apache.beam.sdk.values.Row}. {@link
- * org.apache.beam.sdk.schemas.Schema.FieldType} must be a <code>Map<String, String>
- * </code>
- */
- @Nullable
- public abstract String getAttributesFieldName();
-
- /**
- * The timestamp field name of the source {@link org.apache.beam.sdk.values.Row}. {@link
- * org.apache.beam.sdk.schemas.Schema.FieldType} must be a {@link
- * org.apache.beam.sdk.schemas.Schema.FieldType#DATETIME}.
- */
- @Nullable
- public abstract String getTimestampFieldName();
-
- /**
- * The payload field name of the source {@link org.apache.beam.sdk.values.Row}. {@link
- * org.apache.beam.sdk.schemas.Schema.FieldType} must be either {@link
- * org.apache.beam.sdk.schemas.Schema.FieldType#BYTES} or a {@link
- * org.apache.beam.sdk.values.Row}. If null, payload serialized from user fields other than
- * attributes. Not compatible with other payload intended fields.
- */
- @Nullable
- public abstract String getPayloadFieldName();
-
- @AutoValue.Builder
- public abstract static class Builder {
- /**
- * The attributes field name of the source {@link org.apache.beam.sdk.values.Row}. {@link
- * org.apache.beam.sdk.schemas.Schema.FieldType} must be a <code>Map<String, String>
- * </code>
- */
- public abstract Builder setAttributesFieldName(String value);
-
- /**
- * The timestamp field name of the source {@link org.apache.beam.sdk.values.Row}. {@link
- * org.apache.beam.sdk.schemas.Schema.FieldType} must be a {@link
- * org.apache.beam.sdk.schemas.Schema.FieldType#DATETIME}.
- */
- public abstract Builder setTimestampFieldName(String value);
-
- /**
- * The payload field name of the source {@link org.apache.beam.sdk.values.Row}. {@link
- * org.apache.beam.sdk.schemas.Schema.FieldType} must be either {@link
- * org.apache.beam.sdk.schemas.Schema.FieldType#BYTES} or a {@link
- * org.apache.beam.sdk.values.Row}. If null, payload serialized from user fields other than
- * attributes. Not compatible with other payload intended fields.
- */
- public abstract Builder setPayloadFieldName(String value);
-
- public abstract SourceConfiguration build();
- }
- }
-
- @DefaultSchema(AutoValueSchema.class)
- @AutoValue
- public abstract static class TargetConfiguration {
-
- /**
- * The attribute key to assign the {@link PubsubMessage} stringified timestamp value. {@link
- * #builder()} method defaults value to {@link #DEFAULT_TIMESTAMP_ATTRIBUTE}.
- */
- public abstract String getTimestampAttributeKey();
-
- @AutoValue.Builder
- public abstract static class Builder {
-
- /**
- * The attribute key to assign the {@link PubsubMessage} stringified timestamp value. Defaults
- * to {@link #DEFAULT_TIMESTAMP_ATTRIBUTE}.
- */
- public abstract Builder setTimestampAttributeKey(String value);
-
- public abstract TargetConfiguration build();
- }
- }
+ // public static PubsubWriteSchemaTransformConfiguration create(String format, String topic) {
Review Comment:
can we remove these comments?
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] Dippatel98 commented on a diff in pull request #27366: Moving Pubsub Transforms to beam
Posted by "Dippatel98 (via GitHub)" <gi...@apache.org>.
Dippatel98 commented on code in PR #27366:
URL: https://github.com/apache/beam/pull/27366#discussion_r1260294605
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubUtils.java:
##########
@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.pubsub;
+
+import com.google.cloud.pubsub.v1.SchemaServiceClient;
+import com.google.cloud.pubsub.v1.SubscriptionAdminClient;
+import com.google.cloud.pubsub.v1.TopicAdminClient;
+import com.google.pubsub.v1.Encoding;
+import com.google.pubsub.v1.Schema;
+import com.google.pubsub.v1.Topic;
+import java.io.IOException;
+import org.apache.beam.sdk.io.gcp.common.ProtoSchemaUtils;
+import org.apache.beam.sdk.schemas.utils.AvroUtils;
+import org.apache.beam.sdk.schemas.utils.JsonUtils;
+import org.apache.beam.sdk.transforms.SerializableFunction;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.Row;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+/** A utility class for working with Google Cloud Pub/Sub topics and subscriptions. */
+public class PubsubUtils {
+
+ /**
+ * Retrieves the schema and data conversion function for a Pub/Sub topic.
+ *
+ * @param topicName the name of the Pub/Sub topic
+ * @param subscriptionName the name of the subscription to the topic (optional)
+ * @return a {@link KV} object containing the schema and data conversion function
+ * @throws IOException if an error occurs while retrieving the schema from Pub/Sub
+ * @throws IllegalArgumentException if the schema or encoding is not supported
+ */
+ public static KV<org.apache.beam.sdk.schemas.Schema, SerializableFunction<byte[], Row>>
Review Comment:
As discussed I have added ITs, for the cases.
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] Dippatel98 commented on a diff in pull request #27366: Moving Pubsub Transforms to beam
Posted by "Dippatel98 (via GitHub)" <gi...@apache.org>.
Dippatel98 commented on code in PR #27366:
URL: https://github.com/apache/beam/pull/27366#discussion_r1258437648
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubUtils.java:
##########
@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.pubsub;
+
+import com.google.cloud.pubsub.v1.SchemaServiceClient;
+import com.google.cloud.pubsub.v1.SubscriptionAdminClient;
+import com.google.cloud.pubsub.v1.TopicAdminClient;
+import com.google.pubsub.v1.Encoding;
+import com.google.pubsub.v1.Schema;
+import com.google.pubsub.v1.Topic;
+import java.io.IOException;
+import org.apache.beam.sdk.io.gcp.common.ProtoSchemaUtils;
+import org.apache.beam.sdk.schemas.utils.AvroUtils;
+import org.apache.beam.sdk.schemas.utils.JsonUtils;
+import org.apache.beam.sdk.transforms.SerializableFunction;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.Row;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+/** A utility class for working with Google Cloud Pub/Sub topics and subscriptions. */
+public class PubsubUtils {
+
+ /**
+ * Retrieves the schema and data conversion function for a Pub/Sub topic.
+ *
+ * @param topicName the name of the Pub/Sub topic
+ * @param subscriptionName the name of the subscription to the topic (optional)
+ * @return a {@link KV} object containing the schema and data conversion function
+ * @throws IOException if an error occurs while retrieving the schema from Pub/Sub
+ * @throws IllegalArgumentException if the schema or encoding is not supported
+ */
+ public static KV<org.apache.beam.sdk.schemas.Schema, SerializableFunction<byte[], Row>>
Review Comment:
The method in question uses `com.google.cloud.pubsub.v1.TopicAdminClient` to query the pubsub service for Topic metadata. I don't think we have a test client for that, the only other way would be to create actual topics like an integration test. Would that be considered acceptable?
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ahmedabu98 commented on a diff in pull request #27366: Moving Pubsub Transforms to beam
Posted by "ahmedabu98 (via GitHub)" <gi...@apache.org>.
ahmedabu98 commented on code in PR #27366:
URL: https://github.com/apache/beam/pull/27366#discussion_r1256375873
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubReadSchemaTransformProvider.java:
##########
@@ -43,196 +62,211 @@
* provide no backwards compatibility guarantees, and it should not be implemented outside the Beam
* repository.
*/
-@SuppressWarnings({
- "nullness" // TODO(https://github.com/apache/beam/issues/20497)
-})
-@Internal
@AutoService(SchemaTransformProvider.class)
public class PubsubReadSchemaTransformProvider
extends TypedSchemaTransformProvider<PubsubReadSchemaTransformConfiguration> {
- static final String OUTPUT_TAG = "OUTPUT";
- /** Returns the expected class of the configuration. */
- @Override
- protected Class<PubsubReadSchemaTransformConfiguration> configurationClass() {
- return PubsubReadSchemaTransformConfiguration.class;
- }
+ public static final String VALID_FORMATS_STR = "AVRO,JSON";
+ public static final Set<String> VALID_DATA_FORMATS =
+ Sets.newHashSet(VALID_FORMATS_STR.split(","));
- /** Returns the expected {@link SchemaTransform} of the configuration. */
- @Override
- protected SchemaTransform from(PubsubReadSchemaTransformConfiguration configuration) {
- PubsubMessageToRow toRowTransform =
- PubsubSchemaTransformMessageToRowFactory.from(configuration).buildMessageToRow();
- return new PubsubReadSchemaTransform(configuration, toRowTransform);
- }
-
- /** Implementation of the {@link TypedSchemaTransformProvider} identifier method. */
- @Override
- public String identifier() {
- return "beam:schematransform:org.apache.beam:pubsub_read:v1";
- }
+ public static final TupleTag<Row> OUTPUT_TAG = new TupleTag<Row>() {};
+ public static final TupleTag<Row> ERROR_TAG = new TupleTag<Row>() {};
+ public static final Schema ERROR_SCHEMA =
+ Schema.builder().addStringField("error").addNullableByteArrayField("row").build();
- /**
- * Implementation of the {@link TypedSchemaTransformProvider} inputCollectionNames method. Since
- * no input is expected, this returns an empty list.
- */
@Override
- public List<String> inputCollectionNames() {
- return Collections.emptyList();
+ public Class<PubsubReadSchemaTransformConfiguration> configurationClass() {
+ return PubsubReadSchemaTransformConfiguration.class;
}
- /**
- * Implementation of the {@link TypedSchemaTransformProvider} outputCollectionNames method. Since
- * a single output is expected, this returns a list with a single name.
- */
@Override
- public List<String> outputCollectionNames() {
- return Collections.singletonList(OUTPUT_TAG);
- }
-
- /**
- * An implementation of {@link SchemaTransform} for Pub/Sub reads configured using {@link
- * PubsubReadSchemaTransformConfiguration}.
- */
- static class PubsubReadSchemaTransform extends SchemaTransform {
-
- private final PubsubReadSchemaTransformConfiguration configuration;
- private final PubsubMessageToRow pubsubMessageToRow;
-
- private PubsubClient.PubsubClientFactory clientFactory;
-
- private Clock clock;
-
- private PubsubReadSchemaTransform(
- PubsubReadSchemaTransformConfiguration configuration,
- PubsubMessageToRow pubsubMessageToRow) {
- this.configuration = configuration;
- this.pubsubMessageToRow = pubsubMessageToRow;
+ public SchemaTransform from(PubsubReadSchemaTransformConfiguration configuration) {
+ if (configuration.getSubscription() == null && configuration.getTopic() == null) {
+ throw new IllegalArgumentException(
+ "To read from Pubsub, a subscription name or a topic name must be provided");
}
- /**
- * Sets the {@link PubsubClient.PubsubClientFactory}.
- *
- * <p>Used for testing.
- */
- void setClientFactory(PubsubClient.PubsubClientFactory value) {
- this.clientFactory = value;
+ if (configuration.getSubscription() != null && configuration.getTopic() != null) {
+ throw new IllegalArgumentException(
+ "To read from Pubsub, a subscription name or a topic name must be provided. Not both.");
}
- /**
- * Sets the {@link Clock}.
- *
- * <p>Used for testing.
- */
- void setClock(Clock clock) {
- this.clock = clock;
+ if ((Strings.isNullOrEmpty(configuration.getSchema())
+ && !Strings.isNullOrEmpty(configuration.getFormat()))
+ || (!Strings.isNullOrEmpty(configuration.getSchema())
+ && Strings.isNullOrEmpty(configuration.getFormat()))) {
+ throw new IllegalArgumentException(
+ "A schema was provided without a data format (or viceversa). Please provide "
+ + "both of these parameters to read from Pubsub, or if you would like to use the Pubsub schema service,"
+ + " please leave both of these blank.");
}
- /** Validates the {@link PubsubReadSchemaTransformConfiguration}. */
- @Override
- public void validate(@Nullable PipelineOptions options) {
- if (configuration.getSubscription() == null && configuration.getTopic() == null) {
- throw new IllegalArgumentException(
+ Schema beamSchema;
+ SerializableFunction<byte[], Row> valueMapper;
+ if (configuration.getSchema() == null && configuration.getFormat() == null) {
+ try {
+ KV<Schema, SerializableFunction<byte[], Row>> schemaFunctionPair =
+ PubsubUtils.getTopicInfo(configuration.getTopic(), configuration.getSubscription());
+ beamSchema = schemaFunctionPair.getKey();
+ valueMapper = schemaFunctionPair.getValue();
+ } catch (IOException e) {
+ throw new RuntimeException(
String.format(
- "%s needs to set either the topic or the subscription",
- PubsubReadSchemaTransformConfiguration.class));
+ "Unable to retrieve schema information for topic %s or subscription %s",
+ configuration.getTopic(), configuration.getSubscription()),
+ e);
}
-
- if (configuration.getSubscription() != null && configuration.getTopic() != null) {
+ } else {
+ if (!VALID_DATA_FORMATS.contains(configuration.getFormat())) {
throw new IllegalArgumentException(
String.format(
- "%s should not set both the topic or the subscription",
- PubsubReadSchemaTransformConfiguration.class));
- }
-
- try {
- PayloadSerializers.getSerializer(
- configuration.getFormat(), configuration.getDataSchema(), new HashMap<>());
- } catch (IllegalArgumentException e) {
- throw new IllegalArgumentException(
- String.format(
- "Invalid %s, no serializer provider exists for format `%s`",
- PubsubReadSchemaTransformConfiguration.class, configuration.getFormat()));
+ "Format %s not supported. Only supported formats are %s",
+ configuration.getFormat(), VALID_FORMATS_STR));
}
+ beamSchema =
+ Objects.equals(configuration.getFormat(), "JSON")
+ ? JsonUtils.beamSchemaFromJsonSchema(configuration.getSchema())
+ : AvroUtils.toBeamSchema(
+ new org.apache.avro.Schema.Parser().parse(configuration.getSchema()));
+ valueMapper =
+ Objects.equals(configuration.getFormat(), "JSON")
+ ? JsonUtils.getJsonBytesToRowFunction(beamSchema)
+ : AvroUtils.getAvroBytesToRowFunction(beamSchema);
}
- /** Reads from Pub/Sub according to {@link PubsubReadSchemaTransformConfiguration}. */
- @Override
- public PCollectionRowTuple expand(PCollectionRowTuple input) {
- if (!input.getAll().isEmpty()) {
- throw new IllegalArgumentException(
- String.format(
- "%s %s input is expected to be empty",
- input.getClass().getSimpleName(), getClass().getSimpleName()));
- }
+ PubsubReadSchemaTransform transform =
+ new PubsubReadSchemaTransform(
+ configuration.getTopic(), configuration.getSubscription(), beamSchema, valueMapper);
- PCollectionTuple rowsWithDlq =
- input
- .getPipeline()
- .apply("ReadFromPubsub", buildPubsubRead())
- .apply("PubsubMessageToRow", pubsubMessageToRow);
+ if (configuration.getClientFactory() != null) {
+ transform.setClientFactory(configuration.getClientFactory());
+ }
+ if (configuration.getClock() != null) {
+ transform.setClock(configuration.getClock());
+ }
- writeToDeadLetterQueue(rowsWithDlq);
+ return transform;
+ }
- return PCollectionRowTuple.of(OUTPUT_TAG, rowsWithDlq.get(MAIN_TAG));
+ private static class PubsubReadSchemaTransform extends SchemaTransform implements Serializable {
+ final Schema beamSchema;
+ final SerializableFunction<byte[], Row> valueMapper;
+ final @Nullable String topic;
+ final @Nullable String subscription;
+ @Nullable PubsubTestClientFactory clientFactory;
+ @Nullable Clock clock;
+
+ PubsubReadSchemaTransform(
+ @Nullable String topic,
+ @Nullable String subscription,
+ Schema beamSchema,
+ SerializableFunction<byte[], Row> valueMapper) {
+ this.topic = topic;
+ this.subscription = subscription;
+ this.beamSchema = beamSchema;
+ this.valueMapper = valueMapper;
}
- private void writeToDeadLetterQueue(PCollectionTuple rowsWithDlq) {
- PubsubIO.Write<PubsubMessage> deadLetterQueue = buildDeadLetterQueueWrite();
- if (deadLetterQueue == null) {
- return;
- }
- rowsWithDlq.get(DLQ_TAG).apply("WriteToDeadLetterQueue", deadLetterQueue);
- }
+ private static class ErrorCounterFn extends DoFn<PubsubMessage, Row> {
+ private Counter pubsubErrorCounter;
+ private Long errorsInBundle = 0L;
+ private SerializableFunction<byte[], Row> valueMapper;
- /**
- * Builds {@link PubsubIO.Write} dead letter queue from {@link
- * PubsubReadSchemaTransformConfiguration}.
- */
- PubsubIO.Write<PubsubMessage> buildDeadLetterQueueWrite() {
- if (configuration.getDeadLetterQueue() == null) {
- return null;
+ ErrorCounterFn(String name, SerializableFunction<byte[], Row> valueMapper) {
+ this.pubsubErrorCounter = Metrics.counter(PubsubReadSchemaTransformProvider.class, name);
+ this.valueMapper = valueMapper;
}
- PubsubIO.Write<PubsubMessage> writeDlq =
- PubsubIO.writeMessages().to(configuration.getDeadLetterQueue());
+ @ProcessElement
+ public void process(@DoFn.Element PubsubMessage message, MultiOutputReceiver receiver) {
+
+ try {
+ receiver.get(OUTPUT_TAG).output(valueMapper.apply(message.getPayload()));
+ } catch (Exception e) {
+ errorsInBundle += 1;
+ System.out.println("Error while parsing the element" + e.toString());
Review Comment:
Maybe we don't need this log
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubWriteSchemaTransformProvider.java:
##########
@@ -76,360 +50,113 @@
* provide no backwards compatibility guarantees, and it should not be implemented outside the Beam
* repository.
*/
-@SuppressWarnings({
- "nullness" // TODO(https://github.com/apache/beam/issues/20497)
-})
-@Internal
+// @SuppressWarnings({
+// "nullness" // TODO(https://github.com/apache/beam/issues/20497)
+// })
@AutoService(SchemaTransformProvider.class)
public class PubsubWriteSchemaTransformProvider
extends TypedSchemaTransformProvider<PubsubWriteSchemaTransformConfiguration> {
- private static final String IDENTIFIER = "beam:schematransform:org.apache.beam:pubsub_write:v1";
- static final String INPUT_TAG = "input";
- static final String ERROR_TAG = "error";
- /** Returns the expected class of the configuration. */
- @Override
- protected Class<PubsubWriteSchemaTransformConfiguration> configurationClass() {
- return PubsubWriteSchemaTransformConfiguration.class;
- }
+ public static final TupleTag<PubsubMessage> OUTPUT_TAG = new TupleTag<PubsubMessage>() {};
+ public static final TupleTag<Row> ERROR_TAG = new TupleTag<Row>() {};
- /** Returns the expected {@link SchemaTransform} of the configuration. */
- @Override
- public SchemaTransform from(PubsubWriteSchemaTransformConfiguration configuration) {
- return new PubsubWriteSchemaTransform(configuration);
- }
-
- /** Implementation of the {@link SchemaTransformProvider} identifier method. */
- @Override
- public String identifier() {
- return IDENTIFIER;
- }
-
- /**
- * Implementation of the {@link TypedSchemaTransformProvider} inputCollectionNames method. Since a
- * single input is expected, this returns a list with a single name.
- */
- @Override
- public List<String> inputCollectionNames() {
- return Collections.singletonList(INPUT_TAG);
- }
+ public static final String VALID_FORMATS_STR = "AVRO,JSON";
+ public static final Set<String> VALID_DATA_FORMATS =
+ Sets.newHashSet(VALID_FORMATS_STR.split(","));
- /**
- * Implementation of the {@link TypedSchemaTransformProvider} outputCollectionNames method. The
- * only expected output is the {@link #ERROR_TAG}.
- */
@Override
- public List<String> outputCollectionNames() {
- return Collections.singletonList(ERROR_TAG);
+ public Class<PubsubWriteSchemaTransformConfiguration> configurationClass() {
+ return PubsubWriteSchemaTransformConfiguration.class;
}
- /**
- * An implementation of {@link SchemaTransform} for Pub/Sub writes configured using {@link
- * PubsubWriteSchemaTransformConfiguration}.
- */
- static class PubsubWriteSchemaTransform extends SchemaTransform {
-
- private final PubsubWriteSchemaTransformConfiguration configuration;
+ public static class ErrorFn extends DoFn<Row, PubsubMessage> {
+ private SerializableFunction<Row, byte[]> valueMapper;
+ private Schema errorSchema;
- private PubsubClient.PubsubClientFactory pubsubClientFactory;
-
- PubsubWriteSchemaTransform(PubsubWriteSchemaTransformConfiguration configuration) {
- this.configuration = configuration;
+ ErrorFn(SerializableFunction<Row, byte[]> valueMapper, Schema errorSchema) {
+ this.valueMapper = valueMapper;
+ this.errorSchema = errorSchema;
}
- PubsubWriteSchemaTransform withPubsubClientFactory(PubsubClient.PubsubClientFactory factory) {
- this.pubsubClientFactory = factory;
- return this;
- }
-
- @Override
- public PCollectionRowTuple expand(PCollectionRowTuple input) {
- if (input.getAll().size() != 1 || !input.has(INPUT_TAG)) {
- throw new IllegalArgumentException(
- String.format(
- "%s %s input is expected to contain a single %s tagged PCollection<Row>",
- input.getClass().getSimpleName(), getClass().getSimpleName(), INPUT_TAG));
- }
-
- PCollection<Row> rows = input.get(INPUT_TAG);
- if (rows.getSchema().getFieldCount() == 0) {
- throw new IllegalArgumentException(String.format("empty Schema for %s", INPUT_TAG));
- }
-
- Schema targetSchema = buildTargetSchema(rows.getSchema());
-
- rows =
- rows.apply(
- ConvertForRowToMessage.class.getSimpleName(),
- convertForRowToMessage(targetSchema))
- .setRowSchema(targetSchema);
-
- Schema schema = rows.getSchema();
-
- Schema serializableSchema =
- removeFields(schema, DEFAULT_ATTRIBUTES_KEY_NAME, DEFAULT_EVENT_TIMESTAMP_KEY_NAME);
- FieldMatcher payloadRowMatcher = FieldMatcher.of(DEFAULT_PAYLOAD_KEY_NAME, TypeName.ROW);
- if (payloadRowMatcher.match(serializableSchema)) {
- serializableSchema =
- serializableSchema.getField(DEFAULT_PAYLOAD_KEY_NAME).getType().getRowSchema();
- }
-
- validateTargetSchemaAgainstPubsubSchema(serializableSchema, input.getPipeline().getOptions());
-
- PCollectionTuple pct =
- rows.apply(
- PubsubRowToMessage.class.getSimpleName(),
- buildPubsubRowToMessage(serializableSchema));
-
- PCollection<PubsubMessage> messages = pct.get(OUTPUT);
- messages.apply(PubsubIO.Write.class.getSimpleName(), buildPubsubWrite());
- return PCollectionRowTuple.of(ERROR_TAG, pct.get(ERROR));
- }
-
- PayloadSerializer getPayloadSerializer(Schema schema) {
- if (configuration.getFormat() == null) {
- return null;
- }
- String format = configuration.getFormat();
- Set<String> availableFormats =
- Providers.loadProviders(PayloadSerializerProvider.class).keySet();
- if (!availableFormats.contains(format)) {
- String availableFormatsString = String.join(",", availableFormats);
- throw new IllegalArgumentException(
- String.format(
- "%s is not among the valid formats: [%s]", format, availableFormatsString));
- }
- return PayloadSerializers.getSerializer(configuration.getFormat(), schema, ImmutableMap.of());
- }
-
- PubsubRowToMessage buildPubsubRowToMessage(Schema schema) {
- PubsubRowToMessage.Builder builder =
- PubsubRowToMessage.builder().setPayloadSerializer(getPayloadSerializer(schema));
-
- if (configuration.getTarget() != null) {
- builder =
- builder.setTargetTimestampAttributeName(
- configuration.getTarget().getTimestampAttributeKey());
- }
-
- return builder.build();
- }
-
- PubsubIO.Write<PubsubMessage> buildPubsubWrite() {
- PubsubIO.Write<PubsubMessage> write = PubsubIO.writeMessages().to(configuration.getTopic());
-
- if (configuration.getIdAttribute() != null) {
- write = write.withIdAttribute(configuration.getIdAttribute());
- }
-
- if (pubsubClientFactory != null) {
- write = write.withClientFactory(pubsubClientFactory);
+ @ProcessElement
+ public void processElement(@Element Row row, MultiOutputReceiver receiver) {
+ try {
+ receiver.get(OUTPUT_TAG).output(new PubsubMessage(valueMapper.apply(row), null));
+ } catch (Exception e) {
+ System.out.println("Error while parsing the element" + e.toString());
Review Comment:
Cleanup this log?
##########
sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubReadSchemaTransformProviderTest.java:
##########
@@ -18,300 +18,240 @@
package org.apache.beam.sdk.io.gcp.pubsub;
import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertThrows;
import com.google.api.client.util.Clock;
-import com.google.gson.Gson;
-import com.google.gson.JsonObject;
-import com.google.gson.JsonPrimitive;
import com.google.protobuf.ByteString;
import com.google.protobuf.Timestamp;
import java.io.IOException;
import java.io.Serializable;
-import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
-import java.util.Map;
-import java.util.Objects;
import java.util.UUID;
import java.util.stream.Collectors;
+import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.extensions.avro.schemas.io.payloads.AvroPayloadSerializerProvider;
-import org.apache.beam.sdk.schemas.AutoValueSchema;
+import org.apache.beam.sdk.extensions.avro.schemas.utils.AvroUtils;
+import org.apache.beam.sdk.io.gcp.pubsub.PubsubTestClient.PubsubTestClientFactory;
+import org.apache.beam.sdk.metrics.MetricNameFilter;
+import org.apache.beam.sdk.metrics.MetricQueryResults;
+import org.apache.beam.sdk.metrics.MetricResult;
+import org.apache.beam.sdk.metrics.MetricResults;
+import org.apache.beam.sdk.metrics.MetricsFilter;
import org.apache.beam.sdk.schemas.Schema;
import org.apache.beam.sdk.schemas.io.payloads.PayloadSerializer;
+import org.apache.beam.sdk.schemas.transforms.SchemaTransform;
import org.apache.beam.sdk.testing.PAssert;
import org.apache.beam.sdk.testing.TestPipeline;
-import org.apache.beam.sdk.transforms.Create;
-import org.apache.beam.sdk.transforms.SerializableFunction;
-import org.apache.beam.sdk.transforms.display.DisplayData;
import org.apache.beam.sdk.values.PCollectionRowTuple;
import org.apache.beam.sdk.values.Row;
-import org.apache.beam.sdk.values.TypeDescriptor;
import org.junit.Rule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;
-/** Tests for {@link PubsubReadSchemaTransformProvider}. */
+/** Tests for {@link org.apache.beam.sdk.io.gcp.pubsub.PubsubReadSchemaTransformProvider}. */
@RunWith(JUnit4.class)
public class PubsubReadSchemaTransformProviderTest {
- private static final Schema SCHEMA =
+ private static final Schema BEAMSCHEMA =
Schema.of(
Schema.Field.of("name", Schema.FieldType.STRING),
Schema.Field.of("number", Schema.FieldType.INT64));
-
+ private static final Schema BEAMSCHEMAWITHERROR =
+ Schema.of(Schema.Field.of("error", Schema.FieldType.STRING));
+ private static final String SCHEMA = AvroUtils.toAvroSchema(BEAMSCHEMA).toString();
private static final String SUBSCRIPTION = "projects/project/subscriptions/subscription";
private static final String TOPIC = "projects/project/topics/topic";
- private static final List<TestCase> cases =
- Arrays.asList(
- testCase(
- "no configured topic or subscription",
- PubsubReadSchemaTransformConfiguration.builder().setDataSchema(SCHEMA).build())
- .expectInvalidConfiguration(),
- testCase(
- "both topic and subscription configured",
- PubsubReadSchemaTransformConfiguration.builder()
- .setSubscription(SUBSCRIPTION)
- .setSubscription(TOPIC)
- .setDataSchema(SCHEMA)
- .build())
- .expectInvalidConfiguration(),
- testCase(
- "invalid format configured",
- PubsubReadSchemaTransformConfiguration.builder()
- .setSubscription(SUBSCRIPTION)
- .setDataSchema(SCHEMA)
- .setFormat("invalidformat")
- .build())
- .expectInvalidConfiguration(),
- testCase(
- "configuration with subscription",
- PubsubReadSchemaTransformConfiguration.builder()
- .setSubscription(SUBSCRIPTION)
- .setDataSchema(SCHEMA)
- .build())
- .withExpectedPubsubRead(PubsubIO.readMessages().fromSubscription(SUBSCRIPTION)),
- testCase(
- "configuration with topic",
- PubsubReadSchemaTransformConfiguration.builder()
- .setTopic(TOPIC)
- .setDataSchema(SCHEMA)
- .build())
- .withExpectedPubsubRead(PubsubIO.readMessages().fromTopic(TOPIC)),
- testCase(
- "configuration with subscription, timestamp and id attributes",
- PubsubReadSchemaTransformConfiguration.builder()
- .setSubscription(SUBSCRIPTION)
- .setTimestampAttribute("timestampAttribute")
- .setIdAttribute("idAttribute")
- .setDataSchema(SCHEMA)
- .build())
- .withExpectedPubsubRead(
- PubsubIO.readMessages()
- .fromSubscription(SUBSCRIPTION)
- .withTimestampAttribute("timestampAttribute")
- .withIdAttribute("idAttribute")),
- testCase(
- "configuration with subscription and dead letter queue",
- PubsubReadSchemaTransformConfiguration.builder()
- .setSubscription(SUBSCRIPTION)
- .setDataSchema(SCHEMA)
- .setDeadLetterQueue(TOPIC)
- .build())
- .withExpectedPubsubRead(PubsubIO.readMessages().fromSubscription(SUBSCRIPTION))
- .withExpectedDeadLetterQueue(PubsubIO.writeMessages().to(TOPIC)),
- testCase(
- "configuration with subscription, timestamp attribute, and dead letter queue",
- PubsubReadSchemaTransformConfiguration.builder()
- .setSubscription(SUBSCRIPTION)
- .setTimestampAttribute("timestampAttribute")
- .setDataSchema(SCHEMA)
- .setDeadLetterQueue(TOPIC)
- .build())
- .withExpectedPubsubRead(
- PubsubIO.readMessages()
- .fromSubscription(SUBSCRIPTION)
- .withTimestampAttribute("timestampAttribute"))
- .withExpectedDeadLetterQueue(
- PubsubIO.writeMessages().to(TOPIC).withTimestampAttribute("timestampAttribute")));
-
- private static final AutoValueSchema AUTO_VALUE_SCHEMA = new AutoValueSchema();
- private static final TypeDescriptor<PubsubReadSchemaTransformConfiguration> TYPE_DESCRIPTOR =
- TypeDescriptor.of(PubsubReadSchemaTransformConfiguration.class);
- private static final SerializableFunction<PubsubReadSchemaTransformConfiguration, Row>
- ROW_SERIALIZABLE_FUNCTION = AUTO_VALUE_SCHEMA.toRowFunction(TYPE_DESCRIPTOR);
-
private static final List<Row> ROWS =
Arrays.asList(
- Row.withSchema(SCHEMA).withFieldValue("name", "a").withFieldValue("number", 100L).build(),
- Row.withSchema(SCHEMA).withFieldValue("name", "b").withFieldValue("number", 200L).build(),
- Row.withSchema(SCHEMA)
+ Row.withSchema(BEAMSCHEMA)
+ .withFieldValue("name", "a")
+ .withFieldValue("number", 100L)
+ .build(),
+ Row.withSchema(BEAMSCHEMA)
+ .withFieldValue("name", "b")
+ .withFieldValue("number", 200L)
+ .build(),
+ Row.withSchema(BEAMSCHEMA)
.withFieldValue("name", "c")
.withFieldValue("number", 300L)
.build());
- private static final Clock CLOCK = (Clock & Serializable) () -> 1656788475425L;
+ private static final List<Row> ROWSWITHERROR =
+ Arrays.asList(
+ Row.withSchema(BEAMSCHEMAWITHERROR).withFieldValue("error", "a").build(),
+ Row.withSchema(BEAMSCHEMAWITHERROR).withFieldValue("error", "b").build(),
+ Row.withSchema(BEAMSCHEMAWITHERROR).withFieldValue("error", "c").build());
Review Comment:
Naming convention for these constants is uppercase with underscores separating words (e.g. `BEAM_SCHEMA`)
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubWriteSchemaTransformProvider.java:
##########
@@ -76,360 +50,113 @@
* provide no backwards compatibility guarantees, and it should not be implemented outside the Beam
* repository.
*/
-@SuppressWarnings({
- "nullness" // TODO(https://github.com/apache/beam/issues/20497)
-})
-@Internal
+// @SuppressWarnings({
+// "nullness" // TODO(https://github.com/apache/beam/issues/20497)
+// })
@AutoService(SchemaTransformProvider.class)
public class PubsubWriteSchemaTransformProvider
extends TypedSchemaTransformProvider<PubsubWriteSchemaTransformConfiguration> {
- private static final String IDENTIFIER = "beam:schematransform:org.apache.beam:pubsub_write:v1";
- static final String INPUT_TAG = "input";
- static final String ERROR_TAG = "error";
- /** Returns the expected class of the configuration. */
- @Override
- protected Class<PubsubWriteSchemaTransformConfiguration> configurationClass() {
- return PubsubWriteSchemaTransformConfiguration.class;
- }
+ public static final TupleTag<PubsubMessage> OUTPUT_TAG = new TupleTag<PubsubMessage>() {};
+ public static final TupleTag<Row> ERROR_TAG = new TupleTag<Row>() {};
- /** Returns the expected {@link SchemaTransform} of the configuration. */
- @Override
- public SchemaTransform from(PubsubWriteSchemaTransformConfiguration configuration) {
- return new PubsubWriteSchemaTransform(configuration);
- }
-
- /** Implementation of the {@link SchemaTransformProvider} identifier method. */
- @Override
- public String identifier() {
- return IDENTIFIER;
- }
-
- /**
- * Implementation of the {@link TypedSchemaTransformProvider} inputCollectionNames method. Since a
- * single input is expected, this returns a list with a single name.
- */
- @Override
- public List<String> inputCollectionNames() {
- return Collections.singletonList(INPUT_TAG);
- }
+ public static final String VALID_FORMATS_STR = "AVRO,JSON";
+ public static final Set<String> VALID_DATA_FORMATS =
+ Sets.newHashSet(VALID_FORMATS_STR.split(","));
- /**
- * Implementation of the {@link TypedSchemaTransformProvider} outputCollectionNames method. The
- * only expected output is the {@link #ERROR_TAG}.
- */
@Override
- public List<String> outputCollectionNames() {
- return Collections.singletonList(ERROR_TAG);
+ public Class<PubsubWriteSchemaTransformConfiguration> configurationClass() {
+ return PubsubWriteSchemaTransformConfiguration.class;
}
- /**
- * An implementation of {@link SchemaTransform} for Pub/Sub writes configured using {@link
- * PubsubWriteSchemaTransformConfiguration}.
- */
- static class PubsubWriteSchemaTransform extends SchemaTransform {
-
- private final PubsubWriteSchemaTransformConfiguration configuration;
+ public static class ErrorFn extends DoFn<Row, PubsubMessage> {
+ private SerializableFunction<Row, byte[]> valueMapper;
+ private Schema errorSchema;
- private PubsubClient.PubsubClientFactory pubsubClientFactory;
-
- PubsubWriteSchemaTransform(PubsubWriteSchemaTransformConfiguration configuration) {
- this.configuration = configuration;
+ ErrorFn(SerializableFunction<Row, byte[]> valueMapper, Schema errorSchema) {
+ this.valueMapper = valueMapper;
+ this.errorSchema = errorSchema;
}
- PubsubWriteSchemaTransform withPubsubClientFactory(PubsubClient.PubsubClientFactory factory) {
- this.pubsubClientFactory = factory;
- return this;
- }
-
- @Override
- public PCollectionRowTuple expand(PCollectionRowTuple input) {
- if (input.getAll().size() != 1 || !input.has(INPUT_TAG)) {
- throw new IllegalArgumentException(
- String.format(
- "%s %s input is expected to contain a single %s tagged PCollection<Row>",
- input.getClass().getSimpleName(), getClass().getSimpleName(), INPUT_TAG));
- }
-
- PCollection<Row> rows = input.get(INPUT_TAG);
- if (rows.getSchema().getFieldCount() == 0) {
- throw new IllegalArgumentException(String.format("empty Schema for %s", INPUT_TAG));
- }
-
- Schema targetSchema = buildTargetSchema(rows.getSchema());
-
- rows =
- rows.apply(
- ConvertForRowToMessage.class.getSimpleName(),
- convertForRowToMessage(targetSchema))
- .setRowSchema(targetSchema);
-
- Schema schema = rows.getSchema();
-
- Schema serializableSchema =
- removeFields(schema, DEFAULT_ATTRIBUTES_KEY_NAME, DEFAULT_EVENT_TIMESTAMP_KEY_NAME);
- FieldMatcher payloadRowMatcher = FieldMatcher.of(DEFAULT_PAYLOAD_KEY_NAME, TypeName.ROW);
- if (payloadRowMatcher.match(serializableSchema)) {
- serializableSchema =
- serializableSchema.getField(DEFAULT_PAYLOAD_KEY_NAME).getType().getRowSchema();
- }
-
- validateTargetSchemaAgainstPubsubSchema(serializableSchema, input.getPipeline().getOptions());
-
- PCollectionTuple pct =
- rows.apply(
- PubsubRowToMessage.class.getSimpleName(),
- buildPubsubRowToMessage(serializableSchema));
-
- PCollection<PubsubMessage> messages = pct.get(OUTPUT);
- messages.apply(PubsubIO.Write.class.getSimpleName(), buildPubsubWrite());
- return PCollectionRowTuple.of(ERROR_TAG, pct.get(ERROR));
- }
-
- PayloadSerializer getPayloadSerializer(Schema schema) {
- if (configuration.getFormat() == null) {
- return null;
- }
- String format = configuration.getFormat();
- Set<String> availableFormats =
- Providers.loadProviders(PayloadSerializerProvider.class).keySet();
- if (!availableFormats.contains(format)) {
- String availableFormatsString = String.join(",", availableFormats);
- throw new IllegalArgumentException(
- String.format(
- "%s is not among the valid formats: [%s]", format, availableFormatsString));
- }
- return PayloadSerializers.getSerializer(configuration.getFormat(), schema, ImmutableMap.of());
- }
-
- PubsubRowToMessage buildPubsubRowToMessage(Schema schema) {
- PubsubRowToMessage.Builder builder =
- PubsubRowToMessage.builder().setPayloadSerializer(getPayloadSerializer(schema));
-
- if (configuration.getTarget() != null) {
- builder =
- builder.setTargetTimestampAttributeName(
- configuration.getTarget().getTimestampAttributeKey());
- }
-
- return builder.build();
- }
-
- PubsubIO.Write<PubsubMessage> buildPubsubWrite() {
- PubsubIO.Write<PubsubMessage> write = PubsubIO.writeMessages().to(configuration.getTopic());
-
- if (configuration.getIdAttribute() != null) {
- write = write.withIdAttribute(configuration.getIdAttribute());
- }
-
- if (pubsubClientFactory != null) {
- write = write.withClientFactory(pubsubClientFactory);
+ @ProcessElement
+ public void processElement(@Element Row row, MultiOutputReceiver receiver) {
+ try {
+ receiver.get(OUTPUT_TAG).output(new PubsubMessage(valueMapper.apply(row), null));
+ } catch (Exception e) {
+ System.out.println("Error while parsing the element" + e.toString());
+ receiver
+ .get(ERROR_TAG)
+ .output(Row.withSchema(errorSchema).addValues(e.toString(), row).build());
}
-
- return write;
}
+ }
- void validateSourceSchemaAgainstConfiguration(Schema sourceSchema) {
- if (sourceSchema.getFieldCount() == 0) {
- throw new IllegalArgumentException(String.format("empty Schema for %s", INPUT_TAG));
- }
-
- if (configuration.getSource() == null) {
- return;
- }
-
- SourceConfiguration source = configuration.getSource();
-
- if (source.getAttributesFieldName() != null) {
- String fieldName = source.getAttributesFieldName();
- FieldType fieldType = ATTRIBUTES_FIELD_TYPE;
- FieldMatcher fieldMatcher = FieldMatcher.of(fieldName, fieldType);
- checkArgument(
- fieldMatcher.match(sourceSchema),
- String.format("schema missing field: %s for type %s: ", fieldName, fieldType));
- }
-
- if (source.getTimestampFieldName() != null) {
- String fieldName = source.getTimestampFieldName();
- FieldType fieldType = EVENT_TIMESTAMP_FIELD_TYPE;
- FieldMatcher fieldMatcher = FieldMatcher.of(fieldName, fieldType);
- checkArgument(
- fieldMatcher.match(sourceSchema),
- String.format("schema missing field: %s for type: %s", fieldName, fieldType));
- }
-
- if (source.getPayloadFieldName() == null) {
- return;
- }
-
- String fieldName = source.getPayloadFieldName();
- FieldMatcher bytesFieldMatcher = FieldMatcher.of(fieldName, PAYLOAD_BYTES_TYPE_NAME);
- FieldMatcher rowFieldMatcher = FieldMatcher.of(fieldName, PAYLOAD_ROW_TYPE_NAME);
- SchemaReflection schemaReflection = SchemaReflection.of(sourceSchema);
- checkArgument(
- schemaReflection.matchesAny(bytesFieldMatcher, rowFieldMatcher),
+ @Override
+ public SchemaTransform from(PubsubWriteSchemaTransformConfiguration configuration) {
+ if (!VALID_DATA_FORMATS.contains(configuration.getFormat())) {
+ throw new IllegalArgumentException(
String.format(
- "schema missing field: %s for types %s or %s",
- fieldName, PAYLOAD_BYTES_TYPE_NAME, PAYLOAD_ROW_TYPE_NAME));
-
- String[] fieldsToExclude =
- Stream.of(
- source.getAttributesFieldName(),
- source.getTimestampFieldName(),
- source.getPayloadFieldName())
- .filter(Objects::nonNull)
- .toArray(String[]::new);
-
- Schema userFieldsSchema = removeFields(sourceSchema, fieldsToExclude);
-
- if (userFieldsSchema.getFieldCount() > 0) {
- throw new IllegalArgumentException(
- String.format("user fields incompatible with %s field", source.getPayloadFieldName()));
- }
- }
-
- void validateTargetSchemaAgainstPubsubSchema(Schema targetSchema, PipelineOptions options) {
- checkArgument(options != null);
-
- try (PubsubClient pubsubClient = getPubsubClient(options.as(PubsubOptions.class))) {
- PubsubClient.TopicPath topicPath = PubsubClient.topicPathFromPath(configuration.getTopic());
- PubsubClient.SchemaPath schemaPath = pubsubClient.getSchemaPath(topicPath);
- if (schemaPath == null || schemaPath.equals(SchemaPath.DELETED_SCHEMA)) {
- return;
- }
- Schema expectedSchema = pubsubClient.getSchema(schemaPath);
- checkState(
- targetSchema.equals(expectedSchema),
- String.format(
- "input schema mismatch with expected schema at path: %s\ninput schema: %s\nPub/Sub schema: %s",
- schemaPath, targetSchema, expectedSchema));
- } catch (IOException e) {
- throw new IllegalStateException(e.getMessage());
- }
- }
-
- Schema buildTargetSchema(Schema sourceSchema) {
- validateSourceSchemaAgainstConfiguration(sourceSchema);
- FieldType payloadFieldType = null;
-
- List<String> fieldsToRemove = new ArrayList<>();
-
- if (configuration.getSource() != null) {
- SourceConfiguration source = configuration.getSource();
-
- if (source.getAttributesFieldName() != null) {
- fieldsToRemove.add(source.getAttributesFieldName());
- }
-
- if (source.getTimestampFieldName() != null) {
- fieldsToRemove.add(source.getTimestampFieldName());
- }
-
- if (source.getPayloadFieldName() != null) {
- String fieldName = source.getPayloadFieldName();
- Field field = sourceSchema.getField(fieldName);
- payloadFieldType = field.getType();
- fieldsToRemove.add(fieldName);
- }
- }
-
- Schema targetSchema =
- PubsubRowToMessage.builder()
- .build()
- .inputSchemaFactory(payloadFieldType)
- .buildSchema(sourceSchema.getFields().toArray(new Field[0]));
-
- return removeFields(targetSchema, fieldsToRemove.toArray(new String[0]));
+ "Format %s not supported. Only supported formats are %s",
+ configuration.getFormat(), VALID_FORMATS_STR));
}
+ return new PubsubWriteSchemaTransform(configuration.getTopic(), configuration.getFormat());
+ }
- private PubsubClient.PubsubClientFactory getPubsubClientFactory() {
- if (pubsubClientFactory != null) {
- return pubsubClientFactory;
- }
- return PubsubGrpcClient.FACTORY;
- }
+ private static class PubsubWriteSchemaTransform extends SchemaTransform implements Serializable {
+ final String topic;
+ final String format;
- private PubsubClient getPubsubClient(PubsubOptions options) throws IOException {
- return getPubsubClientFactory()
- .newClient(
- configuration.getTarget().getTimestampAttributeKey(),
- configuration.getIdAttribute(),
- options);
+ PubsubWriteSchemaTransform(String topic, String format) {
+ this.topic = topic;
+ this.format = format;
}
- ParDo.SingleOutput<Row, Row> convertForRowToMessage(Schema targetSchema) {
- return convertForRowToMessage(targetSchema, null);
+ public @UnknownKeyFor @NonNull @Initialized PTransform<
+ @UnknownKeyFor @NonNull @Initialized PCollectionRowTuple,
+ @UnknownKeyFor @NonNull @Initialized PCollectionRowTuple>
+ buildTransform() {
+ return this;
}
Review Comment:
similarly, no need for this method anymore
##########
sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubReadSchemaTransformProviderTest.java:
##########
@@ -18,300 +18,240 @@
package org.apache.beam.sdk.io.gcp.pubsub;
import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertThrows;
import com.google.api.client.util.Clock;
-import com.google.gson.Gson;
-import com.google.gson.JsonObject;
-import com.google.gson.JsonPrimitive;
import com.google.protobuf.ByteString;
import com.google.protobuf.Timestamp;
import java.io.IOException;
import java.io.Serializable;
-import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
-import java.util.Map;
-import java.util.Objects;
import java.util.UUID;
import java.util.stream.Collectors;
+import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.extensions.avro.schemas.io.payloads.AvroPayloadSerializerProvider;
-import org.apache.beam.sdk.schemas.AutoValueSchema;
+import org.apache.beam.sdk.extensions.avro.schemas.utils.AvroUtils;
+import org.apache.beam.sdk.io.gcp.pubsub.PubsubTestClient.PubsubTestClientFactory;
+import org.apache.beam.sdk.metrics.MetricNameFilter;
+import org.apache.beam.sdk.metrics.MetricQueryResults;
+import org.apache.beam.sdk.metrics.MetricResult;
+import org.apache.beam.sdk.metrics.MetricResults;
+import org.apache.beam.sdk.metrics.MetricsFilter;
import org.apache.beam.sdk.schemas.Schema;
import org.apache.beam.sdk.schemas.io.payloads.PayloadSerializer;
+import org.apache.beam.sdk.schemas.transforms.SchemaTransform;
import org.apache.beam.sdk.testing.PAssert;
import org.apache.beam.sdk.testing.TestPipeline;
-import org.apache.beam.sdk.transforms.Create;
-import org.apache.beam.sdk.transforms.SerializableFunction;
-import org.apache.beam.sdk.transforms.display.DisplayData;
import org.apache.beam.sdk.values.PCollectionRowTuple;
import org.apache.beam.sdk.values.Row;
-import org.apache.beam.sdk.values.TypeDescriptor;
import org.junit.Rule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;
-/** Tests for {@link PubsubReadSchemaTransformProvider}. */
+/** Tests for {@link org.apache.beam.sdk.io.gcp.pubsub.PubsubReadSchemaTransformProvider}. */
@RunWith(JUnit4.class)
public class PubsubReadSchemaTransformProviderTest {
- private static final Schema SCHEMA =
+ private static final Schema BEAMSCHEMA =
Schema.of(
Schema.Field.of("name", Schema.FieldType.STRING),
Schema.Field.of("number", Schema.FieldType.INT64));
-
+ private static final Schema BEAMSCHEMAWITHERROR =
+ Schema.of(Schema.Field.of("error", Schema.FieldType.STRING));
+ private static final String SCHEMA = AvroUtils.toAvroSchema(BEAMSCHEMA).toString();
private static final String SUBSCRIPTION = "projects/project/subscriptions/subscription";
private static final String TOPIC = "projects/project/topics/topic";
- private static final List<TestCase> cases =
- Arrays.asList(
- testCase(
- "no configured topic or subscription",
- PubsubReadSchemaTransformConfiguration.builder().setDataSchema(SCHEMA).build())
- .expectInvalidConfiguration(),
- testCase(
- "both topic and subscription configured",
- PubsubReadSchemaTransformConfiguration.builder()
- .setSubscription(SUBSCRIPTION)
- .setSubscription(TOPIC)
- .setDataSchema(SCHEMA)
- .build())
- .expectInvalidConfiguration(),
- testCase(
- "invalid format configured",
- PubsubReadSchemaTransformConfiguration.builder()
- .setSubscription(SUBSCRIPTION)
- .setDataSchema(SCHEMA)
- .setFormat("invalidformat")
- .build())
- .expectInvalidConfiguration(),
- testCase(
- "configuration with subscription",
- PubsubReadSchemaTransformConfiguration.builder()
- .setSubscription(SUBSCRIPTION)
- .setDataSchema(SCHEMA)
- .build())
- .withExpectedPubsubRead(PubsubIO.readMessages().fromSubscription(SUBSCRIPTION)),
- testCase(
- "configuration with topic",
- PubsubReadSchemaTransformConfiguration.builder()
- .setTopic(TOPIC)
- .setDataSchema(SCHEMA)
- .build())
- .withExpectedPubsubRead(PubsubIO.readMessages().fromTopic(TOPIC)),
- testCase(
- "configuration with subscription, timestamp and id attributes",
- PubsubReadSchemaTransformConfiguration.builder()
- .setSubscription(SUBSCRIPTION)
- .setTimestampAttribute("timestampAttribute")
- .setIdAttribute("idAttribute")
- .setDataSchema(SCHEMA)
- .build())
- .withExpectedPubsubRead(
- PubsubIO.readMessages()
- .fromSubscription(SUBSCRIPTION)
- .withTimestampAttribute("timestampAttribute")
- .withIdAttribute("idAttribute")),
- testCase(
- "configuration with subscription and dead letter queue",
- PubsubReadSchemaTransformConfiguration.builder()
- .setSubscription(SUBSCRIPTION)
- .setDataSchema(SCHEMA)
- .setDeadLetterQueue(TOPIC)
- .build())
- .withExpectedPubsubRead(PubsubIO.readMessages().fromSubscription(SUBSCRIPTION))
- .withExpectedDeadLetterQueue(PubsubIO.writeMessages().to(TOPIC)),
- testCase(
- "configuration with subscription, timestamp attribute, and dead letter queue",
- PubsubReadSchemaTransformConfiguration.builder()
- .setSubscription(SUBSCRIPTION)
- .setTimestampAttribute("timestampAttribute")
- .setDataSchema(SCHEMA)
- .setDeadLetterQueue(TOPIC)
- .build())
- .withExpectedPubsubRead(
- PubsubIO.readMessages()
- .fromSubscription(SUBSCRIPTION)
- .withTimestampAttribute("timestampAttribute"))
- .withExpectedDeadLetterQueue(
- PubsubIO.writeMessages().to(TOPIC).withTimestampAttribute("timestampAttribute")));
-
- private static final AutoValueSchema AUTO_VALUE_SCHEMA = new AutoValueSchema();
- private static final TypeDescriptor<PubsubReadSchemaTransformConfiguration> TYPE_DESCRIPTOR =
- TypeDescriptor.of(PubsubReadSchemaTransformConfiguration.class);
- private static final SerializableFunction<PubsubReadSchemaTransformConfiguration, Row>
- ROW_SERIALIZABLE_FUNCTION = AUTO_VALUE_SCHEMA.toRowFunction(TYPE_DESCRIPTOR);
-
private static final List<Row> ROWS =
Arrays.asList(
- Row.withSchema(SCHEMA).withFieldValue("name", "a").withFieldValue("number", 100L).build(),
- Row.withSchema(SCHEMA).withFieldValue("name", "b").withFieldValue("number", 200L).build(),
- Row.withSchema(SCHEMA)
+ Row.withSchema(BEAMSCHEMA)
+ .withFieldValue("name", "a")
+ .withFieldValue("number", 100L)
+ .build(),
+ Row.withSchema(BEAMSCHEMA)
+ .withFieldValue("name", "b")
+ .withFieldValue("number", 200L)
+ .build(),
+ Row.withSchema(BEAMSCHEMA)
.withFieldValue("name", "c")
.withFieldValue("number", 300L)
.build());
- private static final Clock CLOCK = (Clock & Serializable) () -> 1656788475425L;
+ private static final List<Row> ROWSWITHERROR =
+ Arrays.asList(
+ Row.withSchema(BEAMSCHEMAWITHERROR).withFieldValue("error", "a").build(),
+ Row.withSchema(BEAMSCHEMAWITHERROR).withFieldValue("error", "b").build(),
+ Row.withSchema(BEAMSCHEMAWITHERROR).withFieldValue("error", "c").build());
+
+ private static final Clock CLOCK = (Clock & Serializable) () -> 1678988970000L;
private static final AvroPayloadSerializerProvider AVRO_PAYLOAD_SERIALIZER_PROVIDER =
new AvroPayloadSerializerProvider();
private static final PayloadSerializer AVRO_PAYLOAD_SERIALIZER =
- AVRO_PAYLOAD_SERIALIZER_PROVIDER.getSerializer(SCHEMA, new HashMap<>());
-
- @Rule public TestPipeline p = TestPipeline.create().enableAbandonedNodeEnforcement(false);
+ AVRO_PAYLOAD_SERIALIZER_PROVIDER.getSerializer(BEAMSCHEMA, new HashMap<>());
+ private static final PayloadSerializer AVRO_PAYLOAD_SERIALIZER_WITH_ERROR =
+ AVRO_PAYLOAD_SERIALIZER_PROVIDER.getSerializer(BEAMSCHEMAWITHERROR, new HashMap<>());
- @Test
- public void testBuildDeadLetterQueueWrite() {
- for (TestCase testCase : cases) {
- PubsubIO.Write<PubsubMessage> dlq =
- testCase.pubsubReadSchemaTransform().buildDeadLetterQueueWrite();
-
- if (testCase.expectedDeadLetterQueue == null) {
- assertNull(testCase.name, dlq);
- return;
- }
-
- Map<DisplayData.Identifier, DisplayData.Item> actual = DisplayData.from(dlq).asMap();
- Map<DisplayData.Identifier, DisplayData.Item> expected = testCase.expectedDeadLetterQueue;
-
- assertEquals(testCase.name, expected, actual);
- }
- }
+ @Rule public transient TestPipeline p = TestPipeline.create();
@Test
- public void testReadAvro() throws IOException {
+ public void testInvalidConfigNoTopicOrSubscription() {
PCollectionRowTuple begin = PCollectionRowTuple.empty(p);
- PubsubReadSchemaTransformProvider.PubsubReadSchemaTransform transform =
- schemaTransformWithClock("avro");
- PubsubTestClient.PubsubTestClientFactory clientFactory =
- clientFactory(incomingAvroMessagesOf(CLOCK.currentTimeMillis()));
- transform.setClientFactory(clientFactory);
- PCollectionRowTuple reads = begin.apply(transform);
-
- PAssert.that(reads.get(PubsubReadSchemaTransformProvider.OUTPUT_TAG)).containsInAnyOrder(ROWS);
-
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ begin.apply(
+ new PubsubReadSchemaTransformProvider()
+ .from(
+ PubsubReadSchemaTransformConfiguration.builder()
+ .setSchema(SCHEMA)
+ .setFormat("AVRO")
+ .build())));
p.run().waitUntilFinish();
Review Comment:
Looks like validation checks live in PubsubReadSchemaTransformProvider::from, so we actually don't need a pipeline to trigger these exceptions (we just need a `PubsubReadSchemaTransformProvider().from(...)` call)
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubReadSchemaTransformProvider.java:
##########
@@ -43,196 +62,211 @@
* provide no backwards compatibility guarantees, and it should not be implemented outside the Beam
* repository.
*/
-@SuppressWarnings({
- "nullness" // TODO(https://github.com/apache/beam/issues/20497)
-})
-@Internal
@AutoService(SchemaTransformProvider.class)
public class PubsubReadSchemaTransformProvider
extends TypedSchemaTransformProvider<PubsubReadSchemaTransformConfiguration> {
- static final String OUTPUT_TAG = "OUTPUT";
- /** Returns the expected class of the configuration. */
- @Override
- protected Class<PubsubReadSchemaTransformConfiguration> configurationClass() {
- return PubsubReadSchemaTransformConfiguration.class;
- }
+ public static final String VALID_FORMATS_STR = "AVRO,JSON";
+ public static final Set<String> VALID_DATA_FORMATS =
+ Sets.newHashSet(VALID_FORMATS_STR.split(","));
- /** Returns the expected {@link SchemaTransform} of the configuration. */
- @Override
- protected SchemaTransform from(PubsubReadSchemaTransformConfiguration configuration) {
- PubsubMessageToRow toRowTransform =
- PubsubSchemaTransformMessageToRowFactory.from(configuration).buildMessageToRow();
- return new PubsubReadSchemaTransform(configuration, toRowTransform);
- }
-
- /** Implementation of the {@link TypedSchemaTransformProvider} identifier method. */
- @Override
- public String identifier() {
- return "beam:schematransform:org.apache.beam:pubsub_read:v1";
- }
+ public static final TupleTag<Row> OUTPUT_TAG = new TupleTag<Row>() {};
+ public static final TupleTag<Row> ERROR_TAG = new TupleTag<Row>() {};
+ public static final Schema ERROR_SCHEMA =
+ Schema.builder().addStringField("error").addNullableByteArrayField("row").build();
- /**
- * Implementation of the {@link TypedSchemaTransformProvider} inputCollectionNames method. Since
- * no input is expected, this returns an empty list.
- */
@Override
- public List<String> inputCollectionNames() {
- return Collections.emptyList();
+ public Class<PubsubReadSchemaTransformConfiguration> configurationClass() {
+ return PubsubReadSchemaTransformConfiguration.class;
}
- /**
- * Implementation of the {@link TypedSchemaTransformProvider} outputCollectionNames method. Since
- * a single output is expected, this returns a list with a single name.
- */
@Override
- public List<String> outputCollectionNames() {
- return Collections.singletonList(OUTPUT_TAG);
- }
-
- /**
- * An implementation of {@link SchemaTransform} for Pub/Sub reads configured using {@link
- * PubsubReadSchemaTransformConfiguration}.
- */
- static class PubsubReadSchemaTransform extends SchemaTransform {
-
- private final PubsubReadSchemaTransformConfiguration configuration;
- private final PubsubMessageToRow pubsubMessageToRow;
-
- private PubsubClient.PubsubClientFactory clientFactory;
-
- private Clock clock;
-
- private PubsubReadSchemaTransform(
- PubsubReadSchemaTransformConfiguration configuration,
- PubsubMessageToRow pubsubMessageToRow) {
- this.configuration = configuration;
- this.pubsubMessageToRow = pubsubMessageToRow;
+ public SchemaTransform from(PubsubReadSchemaTransformConfiguration configuration) {
+ if (configuration.getSubscription() == null && configuration.getTopic() == null) {
+ throw new IllegalArgumentException(
+ "To read from Pubsub, a subscription name or a topic name must be provided");
}
- /**
- * Sets the {@link PubsubClient.PubsubClientFactory}.
- *
- * <p>Used for testing.
- */
- void setClientFactory(PubsubClient.PubsubClientFactory value) {
- this.clientFactory = value;
+ if (configuration.getSubscription() != null && configuration.getTopic() != null) {
+ throw new IllegalArgumentException(
+ "To read from Pubsub, a subscription name or a topic name must be provided. Not both.");
}
- /**
- * Sets the {@link Clock}.
- *
- * <p>Used for testing.
- */
- void setClock(Clock clock) {
- this.clock = clock;
+ if ((Strings.isNullOrEmpty(configuration.getSchema())
+ && !Strings.isNullOrEmpty(configuration.getFormat()))
+ || (!Strings.isNullOrEmpty(configuration.getSchema())
+ && Strings.isNullOrEmpty(configuration.getFormat()))) {
+ throw new IllegalArgumentException(
+ "A schema was provided without a data format (or viceversa). Please provide "
+ + "both of these parameters to read from Pubsub, or if you would like to use the Pubsub schema service,"
+ + " please leave both of these blank.");
}
- /** Validates the {@link PubsubReadSchemaTransformConfiguration}. */
- @Override
- public void validate(@Nullable PipelineOptions options) {
- if (configuration.getSubscription() == null && configuration.getTopic() == null) {
- throw new IllegalArgumentException(
+ Schema beamSchema;
+ SerializableFunction<byte[], Row> valueMapper;
+ if (configuration.getSchema() == null && configuration.getFormat() == null) {
+ try {
+ KV<Schema, SerializableFunction<byte[], Row>> schemaFunctionPair =
+ PubsubUtils.getTopicInfo(configuration.getTopic(), configuration.getSubscription());
+ beamSchema = schemaFunctionPair.getKey();
+ valueMapper = schemaFunctionPair.getValue();
+ } catch (IOException e) {
+ throw new RuntimeException(
String.format(
- "%s needs to set either the topic or the subscription",
- PubsubReadSchemaTransformConfiguration.class));
+ "Unable to retrieve schema information for topic %s or subscription %s",
+ configuration.getTopic(), configuration.getSubscription()),
+ e);
}
-
- if (configuration.getSubscription() != null && configuration.getTopic() != null) {
+ } else {
+ if (!VALID_DATA_FORMATS.contains(configuration.getFormat())) {
throw new IllegalArgumentException(
String.format(
- "%s should not set both the topic or the subscription",
- PubsubReadSchemaTransformConfiguration.class));
- }
-
- try {
- PayloadSerializers.getSerializer(
- configuration.getFormat(), configuration.getDataSchema(), new HashMap<>());
- } catch (IllegalArgumentException e) {
- throw new IllegalArgumentException(
- String.format(
- "Invalid %s, no serializer provider exists for format `%s`",
- PubsubReadSchemaTransformConfiguration.class, configuration.getFormat()));
+ "Format %s not supported. Only supported formats are %s",
+ configuration.getFormat(), VALID_FORMATS_STR));
}
+ beamSchema =
+ Objects.equals(configuration.getFormat(), "JSON")
+ ? JsonUtils.beamSchemaFromJsonSchema(configuration.getSchema())
+ : AvroUtils.toBeamSchema(
+ new org.apache.avro.Schema.Parser().parse(configuration.getSchema()));
+ valueMapper =
+ Objects.equals(configuration.getFormat(), "JSON")
+ ? JsonUtils.getJsonBytesToRowFunction(beamSchema)
+ : AvroUtils.getAvroBytesToRowFunction(beamSchema);
}
- /** Reads from Pub/Sub according to {@link PubsubReadSchemaTransformConfiguration}. */
- @Override
- public PCollectionRowTuple expand(PCollectionRowTuple input) {
- if (!input.getAll().isEmpty()) {
- throw new IllegalArgumentException(
- String.format(
- "%s %s input is expected to be empty",
- input.getClass().getSimpleName(), getClass().getSimpleName()));
- }
+ PubsubReadSchemaTransform transform =
+ new PubsubReadSchemaTransform(
+ configuration.getTopic(), configuration.getSubscription(), beamSchema, valueMapper);
- PCollectionTuple rowsWithDlq =
- input
- .getPipeline()
- .apply("ReadFromPubsub", buildPubsubRead())
- .apply("PubsubMessageToRow", pubsubMessageToRow);
+ if (configuration.getClientFactory() != null) {
+ transform.setClientFactory(configuration.getClientFactory());
+ }
+ if (configuration.getClock() != null) {
+ transform.setClock(configuration.getClock());
+ }
- writeToDeadLetterQueue(rowsWithDlq);
+ return transform;
+ }
- return PCollectionRowTuple.of(OUTPUT_TAG, rowsWithDlq.get(MAIN_TAG));
+ private static class PubsubReadSchemaTransform extends SchemaTransform implements Serializable {
+ final Schema beamSchema;
+ final SerializableFunction<byte[], Row> valueMapper;
+ final @Nullable String topic;
+ final @Nullable String subscription;
+ @Nullable PubsubTestClientFactory clientFactory;
+ @Nullable Clock clock;
+
+ PubsubReadSchemaTransform(
+ @Nullable String topic,
+ @Nullable String subscription,
+ Schema beamSchema,
+ SerializableFunction<byte[], Row> valueMapper) {
+ this.topic = topic;
+ this.subscription = subscription;
+ this.beamSchema = beamSchema;
+ this.valueMapper = valueMapper;
}
- private void writeToDeadLetterQueue(PCollectionTuple rowsWithDlq) {
- PubsubIO.Write<PubsubMessage> deadLetterQueue = buildDeadLetterQueueWrite();
- if (deadLetterQueue == null) {
- return;
- }
- rowsWithDlq.get(DLQ_TAG).apply("WriteToDeadLetterQueue", deadLetterQueue);
- }
+ private static class ErrorCounterFn extends DoFn<PubsubMessage, Row> {
+ private Counter pubsubErrorCounter;
+ private Long errorsInBundle = 0L;
+ private SerializableFunction<byte[], Row> valueMapper;
- /**
- * Builds {@link PubsubIO.Write} dead letter queue from {@link
- * PubsubReadSchemaTransformConfiguration}.
- */
- PubsubIO.Write<PubsubMessage> buildDeadLetterQueueWrite() {
- if (configuration.getDeadLetterQueue() == null) {
- return null;
+ ErrorCounterFn(String name, SerializableFunction<byte[], Row> valueMapper) {
+ this.pubsubErrorCounter = Metrics.counter(PubsubReadSchemaTransformProvider.class, name);
+ this.valueMapper = valueMapper;
}
- PubsubIO.Write<PubsubMessage> writeDlq =
- PubsubIO.writeMessages().to(configuration.getDeadLetterQueue());
+ @ProcessElement
+ public void process(@DoFn.Element PubsubMessage message, MultiOutputReceiver receiver) {
+
+ try {
+ receiver.get(OUTPUT_TAG).output(valueMapper.apply(message.getPayload()));
+ } catch (Exception e) {
+ errorsInBundle += 1;
+ System.out.println("Error while parsing the element" + e.toString());
+ receiver
+ .get(ERROR_TAG)
+ .output(
+ Row.withSchema(ERROR_SCHEMA)
+ .addValues(e.toString(), message.getPayload())
+ .build());
+ }
+ }
- if (configuration.getTimestampAttribute() != null) {
- writeDlq = writeDlq.withTimestampAttribute(configuration.getTimestampAttribute());
+ @FinishBundle
+ public void finish(FinishBundleContext c) {
+ pubsubErrorCounter.inc(errorsInBundle);
+ errorsInBundle = 0L;
}
+ }
- return writeDlq;
+ void setClientFactory(@Nullable PubsubTestClientFactory factory) {
+ this.clientFactory = factory;
}
- /** Builds {@link PubsubIO.Read} from a {@link PubsubReadSchemaTransformConfiguration}. */
- PubsubIO.Read<PubsubMessage> buildPubsubRead() {
- PubsubIO.Read<PubsubMessage> read = PubsubIO.readMessagesWithAttributes();
+ void setClock(@Nullable Clock clock) {
+ this.clock = clock;
+ }
- if (configuration.getSubscription() != null) {
- read = read.fromSubscription(configuration.getSubscription());
+ @SuppressWarnings("nullness")
+ PubsubIO.Read<PubsubMessage> buildPubsubRead() {
+ PubsubIO.Read<PubsubMessage> pubsubRead = PubsubIO.readMessages();
+ if (!Strings.isNullOrEmpty(topic)) {
+ pubsubRead = pubsubRead.fromTopic(topic);
+ } else {
+ pubsubRead = pubsubRead.fromSubscription(subscription);
}
-
- if (configuration.getTopic() != null) {
- read = read.fromTopic(configuration.getTopic());
+ if (clientFactory != null && clock != null) {
+ pubsubRead = pubsubRead.withClientFactory(clientFactory);
+ pubsubRead = clientFactory.setClock(pubsubRead, clock);
+ } else if (clientFactory != null || clock != null) {
+ throw new IllegalArgumentException(
+ "Both PubsubTestClientFactory and Clock need to be specified for testing, but only one is provided");
}
+ return pubsubRead;
+ }
- if (configuration.getTimestampAttribute() != null) {
- read = read.withTimestampAttribute(configuration.getTimestampAttribute());
- }
+ // @Override
+ public PTransform<PCollectionRowTuple, PCollectionRowTuple> buildTransform() {
+ return this;
+ }
Review Comment:
No need for this method anymore
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] Dippatel98 commented on pull request #27366: Moving Pubsub Transforms to beam
Posted by "Dippatel98 (via GitHub)" <gi...@apache.org>.
Dippatel98 commented on PR #27366:
URL: https://github.com/apache/beam/pull/27366#issuecomment-1622346628
r: @johnjcasey, @ahmedabu98, @manavgarg
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] johnjcasey commented on a diff in pull request #27366: Moving Pubsub Transforms to beam
Posted by "johnjcasey (via GitHub)" <gi...@apache.org>.
johnjcasey commented on code in PR #27366:
URL: https://github.com/apache/beam/pull/27366#discussion_r1256168087
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/common/ProtoSchemaUtils.java:
##########
@@ -0,0 +1,180 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.common;
+
+import com.squareup.wire.schema.Field;
+import com.squareup.wire.schema.Location;
+import com.squareup.wire.schema.internal.parser.EnumElement;
+import com.squareup.wire.schema.internal.parser.FieldElement;
+import com.squareup.wire.schema.internal.parser.MessageElement;
+import com.squareup.wire.schema.internal.parser.ProtoFileElement;
+import com.squareup.wire.schema.internal.parser.ProtoParser;
+import com.squareup.wire.schema.internal.parser.TypeElement;
+import java.util.Map;
+import java.util.Objects;
+import java.util.stream.Collectors;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.schemas.logicaltypes.EnumerationType;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+/**
+ * The ProtoSchemaUtils class provides utility methods for converting Protocol Buffers schema
+ * definitions into Beam schemas.
+ */
+public class ProtoSchemaUtils {
+
+ private static final Location DEFAULT_LOCATION = Location.get("");
+
+ /**
+ * Converts a Protocol Buffers schema definition into a Beam schema.
+ *
+ * @param protobufDef the Protocol Buffers schema definition as a string.
+ * @return the corresponding Beam schema.
+ * @throws SchemaParseException if the Protocol Buffers schema is invalid.
+ */
+ public static Schema beamSchemaFromProtoSchemaDescription(String protobufDef)
+ throws SchemaParseException {
+ ProtoFileElement elm = ProtoParser.Companion.parse(DEFAULT_LOCATION, protobufDef);
+ if (elm.getTypes().size() != 1) {
+ throw new SchemaParseException(
+ String.format(
+ "Expected a single type defined in schema input, but found %s. "
+ + "Provide only a single top-level proto type to use as the "
+ + "message schema.",
+ elm.getTypes().size()));
+ }
+ TypeElement typeElement = elm.getTypes().stream().findFirst().get();
+
+ if (!(typeElement instanceof MessageElement)) {
+ throw new SchemaParseException(
+ "Expected a message type defined, but found a different type definition for "
+ + typeElement.getName());
+ }
+
+ return parsedProtoSchemaToBeamSchema(typeElement, null);
+ }
+
+ /**
+ * Converts a parsed Protocol Buffers schema to a Beam schema.
+ *
+ * @param typeElement the type element representing the Protocol Buffers schema.
+ * @param typeCatalog a catalog of types referenced in the schema.
+ * @return the corresponding Beam schema.
+ */
+ @SuppressWarnings("nullness")
Review Comment:
I'd prefer to avoid adding suppression to the repo
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/common/ProtoSchemaUtils.java:
##########
@@ -0,0 +1,180 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.common;
+
+import com.squareup.wire.schema.Field;
+import com.squareup.wire.schema.Location;
+import com.squareup.wire.schema.internal.parser.EnumElement;
+import com.squareup.wire.schema.internal.parser.FieldElement;
+import com.squareup.wire.schema.internal.parser.MessageElement;
+import com.squareup.wire.schema.internal.parser.ProtoFileElement;
+import com.squareup.wire.schema.internal.parser.ProtoParser;
+import com.squareup.wire.schema.internal.parser.TypeElement;
+import java.util.Map;
+import java.util.Objects;
+import java.util.stream.Collectors;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.schemas.logicaltypes.EnumerationType;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+/**
+ * The ProtoSchemaUtils class provides utility methods for converting Protocol Buffers schema
+ * definitions into Beam schemas.
+ */
+public class ProtoSchemaUtils {
Review Comment:
Can we add tests for this class?
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubWriteSchemaTransformProvider.java:
##########
@@ -76,360 +50,113 @@
* provide no backwards compatibility guarantees, and it should not be implemented outside the Beam
* repository.
*/
-@SuppressWarnings({
- "nullness" // TODO(https://github.com/apache/beam/issues/20497)
-})
-@Internal
+// @SuppressWarnings({
Review Comment:
please remove this.
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubUtils.java:
##########
@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.pubsub;
+
+import com.google.cloud.pubsub.v1.SchemaServiceClient;
+import com.google.cloud.pubsub.v1.SubscriptionAdminClient;
+import com.google.cloud.pubsub.v1.TopicAdminClient;
+import com.google.pubsub.v1.Encoding;
+import com.google.pubsub.v1.Schema;
+import com.google.pubsub.v1.Topic;
+import java.io.IOException;
+import org.apache.beam.sdk.io.gcp.common.ProtoSchemaUtils;
+import org.apache.beam.sdk.schemas.utils.AvroUtils;
+import org.apache.beam.sdk.schemas.utils.JsonUtils;
+import org.apache.beam.sdk.transforms.SerializableFunction;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.Row;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+/** A utility class for working with Google Cloud Pub/Sub topics and subscriptions. */
+public class PubsubUtils {
+
+ /**
+ * Retrieves the schema and data conversion function for a Pub/Sub topic.
+ *
+ * @param topicName the name of the Pub/Sub topic
+ * @param subscriptionName the name of the subscription to the topic (optional)
+ * @return a {@link KV} object containing the schema and data conversion function
+ * @throws IOException if an error occurs while retrieving the schema from Pub/Sub
+ * @throws IllegalArgumentException if the schema or encoding is not supported
+ */
+ public static KV<org.apache.beam.sdk.schemas.Schema, SerializableFunction<byte[], Row>>
Review Comment:
This method has several branches. Can we be sure to test each of them?
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] Dippatel98 commented on a diff in pull request #27366: Moving Pubsub Transforms to beam
Posted by "Dippatel98 (via GitHub)" <gi...@apache.org>.
Dippatel98 commented on code in PR #27366:
URL: https://github.com/apache/beam/pull/27366#discussion_r1256467983
##########
sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubReadSchemaTransformProviderTest.java:
##########
@@ -18,300 +18,240 @@
package org.apache.beam.sdk.io.gcp.pubsub;
import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertThrows;
import com.google.api.client.util.Clock;
-import com.google.gson.Gson;
-import com.google.gson.JsonObject;
-import com.google.gson.JsonPrimitive;
import com.google.protobuf.ByteString;
import com.google.protobuf.Timestamp;
import java.io.IOException;
import java.io.Serializable;
-import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
-import java.util.Map;
-import java.util.Objects;
import java.util.UUID;
import java.util.stream.Collectors;
+import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.extensions.avro.schemas.io.payloads.AvroPayloadSerializerProvider;
-import org.apache.beam.sdk.schemas.AutoValueSchema;
+import org.apache.beam.sdk.extensions.avro.schemas.utils.AvroUtils;
+import org.apache.beam.sdk.io.gcp.pubsub.PubsubTestClient.PubsubTestClientFactory;
+import org.apache.beam.sdk.metrics.MetricNameFilter;
+import org.apache.beam.sdk.metrics.MetricQueryResults;
+import org.apache.beam.sdk.metrics.MetricResult;
+import org.apache.beam.sdk.metrics.MetricResults;
+import org.apache.beam.sdk.metrics.MetricsFilter;
import org.apache.beam.sdk.schemas.Schema;
import org.apache.beam.sdk.schemas.io.payloads.PayloadSerializer;
+import org.apache.beam.sdk.schemas.transforms.SchemaTransform;
import org.apache.beam.sdk.testing.PAssert;
import org.apache.beam.sdk.testing.TestPipeline;
-import org.apache.beam.sdk.transforms.Create;
-import org.apache.beam.sdk.transforms.SerializableFunction;
-import org.apache.beam.sdk.transforms.display.DisplayData;
import org.apache.beam.sdk.values.PCollectionRowTuple;
import org.apache.beam.sdk.values.Row;
-import org.apache.beam.sdk.values.TypeDescriptor;
import org.junit.Rule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;
-/** Tests for {@link PubsubReadSchemaTransformProvider}. */
+/** Tests for {@link org.apache.beam.sdk.io.gcp.pubsub.PubsubReadSchemaTransformProvider}. */
@RunWith(JUnit4.class)
public class PubsubReadSchemaTransformProviderTest {
- private static final Schema SCHEMA =
+ private static final Schema BEAMSCHEMA =
Schema.of(
Schema.Field.of("name", Schema.FieldType.STRING),
Schema.Field.of("number", Schema.FieldType.INT64));
-
+ private static final Schema BEAMSCHEMAWITHERROR =
+ Schema.of(Schema.Field.of("error", Schema.FieldType.STRING));
+ private static final String SCHEMA = AvroUtils.toAvroSchema(BEAMSCHEMA).toString();
private static final String SUBSCRIPTION = "projects/project/subscriptions/subscription";
private static final String TOPIC = "projects/project/topics/topic";
- private static final List<TestCase> cases =
- Arrays.asList(
- testCase(
- "no configured topic or subscription",
- PubsubReadSchemaTransformConfiguration.builder().setDataSchema(SCHEMA).build())
- .expectInvalidConfiguration(),
- testCase(
- "both topic and subscription configured",
- PubsubReadSchemaTransformConfiguration.builder()
- .setSubscription(SUBSCRIPTION)
- .setSubscription(TOPIC)
- .setDataSchema(SCHEMA)
- .build())
- .expectInvalidConfiguration(),
- testCase(
- "invalid format configured",
- PubsubReadSchemaTransformConfiguration.builder()
- .setSubscription(SUBSCRIPTION)
- .setDataSchema(SCHEMA)
- .setFormat("invalidformat")
- .build())
- .expectInvalidConfiguration(),
- testCase(
- "configuration with subscription",
- PubsubReadSchemaTransformConfiguration.builder()
- .setSubscription(SUBSCRIPTION)
- .setDataSchema(SCHEMA)
- .build())
- .withExpectedPubsubRead(PubsubIO.readMessages().fromSubscription(SUBSCRIPTION)),
- testCase(
- "configuration with topic",
- PubsubReadSchemaTransformConfiguration.builder()
- .setTopic(TOPIC)
- .setDataSchema(SCHEMA)
- .build())
- .withExpectedPubsubRead(PubsubIO.readMessages().fromTopic(TOPIC)),
- testCase(
- "configuration with subscription, timestamp and id attributes",
- PubsubReadSchemaTransformConfiguration.builder()
- .setSubscription(SUBSCRIPTION)
- .setTimestampAttribute("timestampAttribute")
- .setIdAttribute("idAttribute")
- .setDataSchema(SCHEMA)
- .build())
- .withExpectedPubsubRead(
- PubsubIO.readMessages()
- .fromSubscription(SUBSCRIPTION)
- .withTimestampAttribute("timestampAttribute")
- .withIdAttribute("idAttribute")),
- testCase(
- "configuration with subscription and dead letter queue",
- PubsubReadSchemaTransformConfiguration.builder()
- .setSubscription(SUBSCRIPTION)
- .setDataSchema(SCHEMA)
- .setDeadLetterQueue(TOPIC)
- .build())
- .withExpectedPubsubRead(PubsubIO.readMessages().fromSubscription(SUBSCRIPTION))
- .withExpectedDeadLetterQueue(PubsubIO.writeMessages().to(TOPIC)),
- testCase(
- "configuration with subscription, timestamp attribute, and dead letter queue",
- PubsubReadSchemaTransformConfiguration.builder()
- .setSubscription(SUBSCRIPTION)
- .setTimestampAttribute("timestampAttribute")
- .setDataSchema(SCHEMA)
- .setDeadLetterQueue(TOPIC)
- .build())
- .withExpectedPubsubRead(
- PubsubIO.readMessages()
- .fromSubscription(SUBSCRIPTION)
- .withTimestampAttribute("timestampAttribute"))
- .withExpectedDeadLetterQueue(
- PubsubIO.writeMessages().to(TOPIC).withTimestampAttribute("timestampAttribute")));
-
- private static final AutoValueSchema AUTO_VALUE_SCHEMA = new AutoValueSchema();
- private static final TypeDescriptor<PubsubReadSchemaTransformConfiguration> TYPE_DESCRIPTOR =
- TypeDescriptor.of(PubsubReadSchemaTransformConfiguration.class);
- private static final SerializableFunction<PubsubReadSchemaTransformConfiguration, Row>
- ROW_SERIALIZABLE_FUNCTION = AUTO_VALUE_SCHEMA.toRowFunction(TYPE_DESCRIPTOR);
-
private static final List<Row> ROWS =
Arrays.asList(
- Row.withSchema(SCHEMA).withFieldValue("name", "a").withFieldValue("number", 100L).build(),
- Row.withSchema(SCHEMA).withFieldValue("name", "b").withFieldValue("number", 200L).build(),
- Row.withSchema(SCHEMA)
+ Row.withSchema(BEAMSCHEMA)
+ .withFieldValue("name", "a")
+ .withFieldValue("number", 100L)
+ .build(),
+ Row.withSchema(BEAMSCHEMA)
+ .withFieldValue("name", "b")
+ .withFieldValue("number", 200L)
+ .build(),
+ Row.withSchema(BEAMSCHEMA)
.withFieldValue("name", "c")
.withFieldValue("number", 300L)
.build());
- private static final Clock CLOCK = (Clock & Serializable) () -> 1656788475425L;
+ private static final List<Row> ROWSWITHERROR =
+ Arrays.asList(
+ Row.withSchema(BEAMSCHEMAWITHERROR).withFieldValue("error", "a").build(),
+ Row.withSchema(BEAMSCHEMAWITHERROR).withFieldValue("error", "b").build(),
+ Row.withSchema(BEAMSCHEMAWITHERROR).withFieldValue("error", "c").build());
+
+ private static final Clock CLOCK = (Clock & Serializable) () -> 1678988970000L;
private static final AvroPayloadSerializerProvider AVRO_PAYLOAD_SERIALIZER_PROVIDER =
new AvroPayloadSerializerProvider();
private static final PayloadSerializer AVRO_PAYLOAD_SERIALIZER =
- AVRO_PAYLOAD_SERIALIZER_PROVIDER.getSerializer(SCHEMA, new HashMap<>());
-
- @Rule public TestPipeline p = TestPipeline.create().enableAbandonedNodeEnforcement(false);
+ AVRO_PAYLOAD_SERIALIZER_PROVIDER.getSerializer(BEAMSCHEMA, new HashMap<>());
+ private static final PayloadSerializer AVRO_PAYLOAD_SERIALIZER_WITH_ERROR =
+ AVRO_PAYLOAD_SERIALIZER_PROVIDER.getSerializer(BEAMSCHEMAWITHERROR, new HashMap<>());
- @Test
- public void testBuildDeadLetterQueueWrite() {
- for (TestCase testCase : cases) {
- PubsubIO.Write<PubsubMessage> dlq =
- testCase.pubsubReadSchemaTransform().buildDeadLetterQueueWrite();
-
- if (testCase.expectedDeadLetterQueue == null) {
- assertNull(testCase.name, dlq);
- return;
- }
-
- Map<DisplayData.Identifier, DisplayData.Item> actual = DisplayData.from(dlq).asMap();
- Map<DisplayData.Identifier, DisplayData.Item> expected = testCase.expectedDeadLetterQueue;
-
- assertEquals(testCase.name, expected, actual);
- }
- }
+ @Rule public transient TestPipeline p = TestPipeline.create();
@Test
- public void testReadAvro() throws IOException {
+ public void testInvalidConfigNoTopicOrSubscription() {
PCollectionRowTuple begin = PCollectionRowTuple.empty(p);
- PubsubReadSchemaTransformProvider.PubsubReadSchemaTransform transform =
- schemaTransformWithClock("avro");
- PubsubTestClient.PubsubTestClientFactory clientFactory =
- clientFactory(incomingAvroMessagesOf(CLOCK.currentTimeMillis()));
- transform.setClientFactory(clientFactory);
- PCollectionRowTuple reads = begin.apply(transform);
-
- PAssert.that(reads.get(PubsubReadSchemaTransformProvider.OUTPUT_TAG)).containsInAnyOrder(ROWS);
-
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ begin.apply(
+ new PubsubReadSchemaTransformProvider()
+ .from(
+ PubsubReadSchemaTransformConfiguration.builder()
+ .setSchema(SCHEMA)
+ .setFormat("AVRO")
+ .build())));
p.run().waitUntilFinish();
Review Comment:
Done
##########
sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubReadSchemaTransformProviderTest.java:
##########
@@ -18,300 +18,240 @@
package org.apache.beam.sdk.io.gcp.pubsub;
import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertThrows;
import com.google.api.client.util.Clock;
-import com.google.gson.Gson;
-import com.google.gson.JsonObject;
-import com.google.gson.JsonPrimitive;
import com.google.protobuf.ByteString;
import com.google.protobuf.Timestamp;
import java.io.IOException;
import java.io.Serializable;
-import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
-import java.util.Map;
-import java.util.Objects;
import java.util.UUID;
import java.util.stream.Collectors;
+import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.extensions.avro.schemas.io.payloads.AvroPayloadSerializerProvider;
-import org.apache.beam.sdk.schemas.AutoValueSchema;
+import org.apache.beam.sdk.extensions.avro.schemas.utils.AvroUtils;
+import org.apache.beam.sdk.io.gcp.pubsub.PubsubTestClient.PubsubTestClientFactory;
+import org.apache.beam.sdk.metrics.MetricNameFilter;
+import org.apache.beam.sdk.metrics.MetricQueryResults;
+import org.apache.beam.sdk.metrics.MetricResult;
+import org.apache.beam.sdk.metrics.MetricResults;
+import org.apache.beam.sdk.metrics.MetricsFilter;
import org.apache.beam.sdk.schemas.Schema;
import org.apache.beam.sdk.schemas.io.payloads.PayloadSerializer;
+import org.apache.beam.sdk.schemas.transforms.SchemaTransform;
import org.apache.beam.sdk.testing.PAssert;
import org.apache.beam.sdk.testing.TestPipeline;
-import org.apache.beam.sdk.transforms.Create;
-import org.apache.beam.sdk.transforms.SerializableFunction;
-import org.apache.beam.sdk.transforms.display.DisplayData;
import org.apache.beam.sdk.values.PCollectionRowTuple;
import org.apache.beam.sdk.values.Row;
-import org.apache.beam.sdk.values.TypeDescriptor;
import org.junit.Rule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;
-/** Tests for {@link PubsubReadSchemaTransformProvider}. */
+/** Tests for {@link org.apache.beam.sdk.io.gcp.pubsub.PubsubReadSchemaTransformProvider}. */
@RunWith(JUnit4.class)
public class PubsubReadSchemaTransformProviderTest {
- private static final Schema SCHEMA =
+ private static final Schema BEAMSCHEMA =
Schema.of(
Schema.Field.of("name", Schema.FieldType.STRING),
Schema.Field.of("number", Schema.FieldType.INT64));
-
+ private static final Schema BEAMSCHEMAWITHERROR =
+ Schema.of(Schema.Field.of("error", Schema.FieldType.STRING));
+ private static final String SCHEMA = AvroUtils.toAvroSchema(BEAMSCHEMA).toString();
private static final String SUBSCRIPTION = "projects/project/subscriptions/subscription";
private static final String TOPIC = "projects/project/topics/topic";
- private static final List<TestCase> cases =
- Arrays.asList(
- testCase(
- "no configured topic or subscription",
- PubsubReadSchemaTransformConfiguration.builder().setDataSchema(SCHEMA).build())
- .expectInvalidConfiguration(),
- testCase(
- "both topic and subscription configured",
- PubsubReadSchemaTransformConfiguration.builder()
- .setSubscription(SUBSCRIPTION)
- .setSubscription(TOPIC)
- .setDataSchema(SCHEMA)
- .build())
- .expectInvalidConfiguration(),
- testCase(
- "invalid format configured",
- PubsubReadSchemaTransformConfiguration.builder()
- .setSubscription(SUBSCRIPTION)
- .setDataSchema(SCHEMA)
- .setFormat("invalidformat")
- .build())
- .expectInvalidConfiguration(),
- testCase(
- "configuration with subscription",
- PubsubReadSchemaTransformConfiguration.builder()
- .setSubscription(SUBSCRIPTION)
- .setDataSchema(SCHEMA)
- .build())
- .withExpectedPubsubRead(PubsubIO.readMessages().fromSubscription(SUBSCRIPTION)),
- testCase(
- "configuration with topic",
- PubsubReadSchemaTransformConfiguration.builder()
- .setTopic(TOPIC)
- .setDataSchema(SCHEMA)
- .build())
- .withExpectedPubsubRead(PubsubIO.readMessages().fromTopic(TOPIC)),
- testCase(
- "configuration with subscription, timestamp and id attributes",
- PubsubReadSchemaTransformConfiguration.builder()
- .setSubscription(SUBSCRIPTION)
- .setTimestampAttribute("timestampAttribute")
- .setIdAttribute("idAttribute")
- .setDataSchema(SCHEMA)
- .build())
- .withExpectedPubsubRead(
- PubsubIO.readMessages()
- .fromSubscription(SUBSCRIPTION)
- .withTimestampAttribute("timestampAttribute")
- .withIdAttribute("idAttribute")),
- testCase(
- "configuration with subscription and dead letter queue",
- PubsubReadSchemaTransformConfiguration.builder()
- .setSubscription(SUBSCRIPTION)
- .setDataSchema(SCHEMA)
- .setDeadLetterQueue(TOPIC)
- .build())
- .withExpectedPubsubRead(PubsubIO.readMessages().fromSubscription(SUBSCRIPTION))
- .withExpectedDeadLetterQueue(PubsubIO.writeMessages().to(TOPIC)),
- testCase(
- "configuration with subscription, timestamp attribute, and dead letter queue",
- PubsubReadSchemaTransformConfiguration.builder()
- .setSubscription(SUBSCRIPTION)
- .setTimestampAttribute("timestampAttribute")
- .setDataSchema(SCHEMA)
- .setDeadLetterQueue(TOPIC)
- .build())
- .withExpectedPubsubRead(
- PubsubIO.readMessages()
- .fromSubscription(SUBSCRIPTION)
- .withTimestampAttribute("timestampAttribute"))
- .withExpectedDeadLetterQueue(
- PubsubIO.writeMessages().to(TOPIC).withTimestampAttribute("timestampAttribute")));
-
- private static final AutoValueSchema AUTO_VALUE_SCHEMA = new AutoValueSchema();
- private static final TypeDescriptor<PubsubReadSchemaTransformConfiguration> TYPE_DESCRIPTOR =
- TypeDescriptor.of(PubsubReadSchemaTransformConfiguration.class);
- private static final SerializableFunction<PubsubReadSchemaTransformConfiguration, Row>
- ROW_SERIALIZABLE_FUNCTION = AUTO_VALUE_SCHEMA.toRowFunction(TYPE_DESCRIPTOR);
-
private static final List<Row> ROWS =
Arrays.asList(
- Row.withSchema(SCHEMA).withFieldValue("name", "a").withFieldValue("number", 100L).build(),
- Row.withSchema(SCHEMA).withFieldValue("name", "b").withFieldValue("number", 200L).build(),
- Row.withSchema(SCHEMA)
+ Row.withSchema(BEAMSCHEMA)
+ .withFieldValue("name", "a")
+ .withFieldValue("number", 100L)
+ .build(),
+ Row.withSchema(BEAMSCHEMA)
+ .withFieldValue("name", "b")
+ .withFieldValue("number", 200L)
+ .build(),
+ Row.withSchema(BEAMSCHEMA)
.withFieldValue("name", "c")
.withFieldValue("number", 300L)
.build());
- private static final Clock CLOCK = (Clock & Serializable) () -> 1656788475425L;
+ private static final List<Row> ROWSWITHERROR =
+ Arrays.asList(
+ Row.withSchema(BEAMSCHEMAWITHERROR).withFieldValue("error", "a").build(),
+ Row.withSchema(BEAMSCHEMAWITHERROR).withFieldValue("error", "b").build(),
+ Row.withSchema(BEAMSCHEMAWITHERROR).withFieldValue("error", "c").build());
Review Comment:
Done
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubWriteSchemaTransformProvider.java:
##########
@@ -76,360 +50,113 @@
* provide no backwards compatibility guarantees, and it should not be implemented outside the Beam
* repository.
*/
-@SuppressWarnings({
- "nullness" // TODO(https://github.com/apache/beam/issues/20497)
-})
-@Internal
+// @SuppressWarnings({
+// "nullness" // TODO(https://github.com/apache/beam/issues/20497)
+// })
@AutoService(SchemaTransformProvider.class)
public class PubsubWriteSchemaTransformProvider
extends TypedSchemaTransformProvider<PubsubWriteSchemaTransformConfiguration> {
- private static final String IDENTIFIER = "beam:schematransform:org.apache.beam:pubsub_write:v1";
- static final String INPUT_TAG = "input";
- static final String ERROR_TAG = "error";
- /** Returns the expected class of the configuration. */
- @Override
- protected Class<PubsubWriteSchemaTransformConfiguration> configurationClass() {
- return PubsubWriteSchemaTransformConfiguration.class;
- }
+ public static final TupleTag<PubsubMessage> OUTPUT_TAG = new TupleTag<PubsubMessage>() {};
+ public static final TupleTag<Row> ERROR_TAG = new TupleTag<Row>() {};
- /** Returns the expected {@link SchemaTransform} of the configuration. */
- @Override
- public SchemaTransform from(PubsubWriteSchemaTransformConfiguration configuration) {
- return new PubsubWriteSchemaTransform(configuration);
- }
-
- /** Implementation of the {@link SchemaTransformProvider} identifier method. */
- @Override
- public String identifier() {
- return IDENTIFIER;
- }
-
- /**
- * Implementation of the {@link TypedSchemaTransformProvider} inputCollectionNames method. Since a
- * single input is expected, this returns a list with a single name.
- */
- @Override
- public List<String> inputCollectionNames() {
- return Collections.singletonList(INPUT_TAG);
- }
+ public static final String VALID_FORMATS_STR = "AVRO,JSON";
+ public static final Set<String> VALID_DATA_FORMATS =
+ Sets.newHashSet(VALID_FORMATS_STR.split(","));
- /**
- * Implementation of the {@link TypedSchemaTransformProvider} outputCollectionNames method. The
- * only expected output is the {@link #ERROR_TAG}.
- */
@Override
- public List<String> outputCollectionNames() {
- return Collections.singletonList(ERROR_TAG);
+ public Class<PubsubWriteSchemaTransformConfiguration> configurationClass() {
+ return PubsubWriteSchemaTransformConfiguration.class;
}
- /**
- * An implementation of {@link SchemaTransform} for Pub/Sub writes configured using {@link
- * PubsubWriteSchemaTransformConfiguration}.
- */
- static class PubsubWriteSchemaTransform extends SchemaTransform {
-
- private final PubsubWriteSchemaTransformConfiguration configuration;
+ public static class ErrorFn extends DoFn<Row, PubsubMessage> {
+ private SerializableFunction<Row, byte[]> valueMapper;
+ private Schema errorSchema;
- private PubsubClient.PubsubClientFactory pubsubClientFactory;
-
- PubsubWriteSchemaTransform(PubsubWriteSchemaTransformConfiguration configuration) {
- this.configuration = configuration;
+ ErrorFn(SerializableFunction<Row, byte[]> valueMapper, Schema errorSchema) {
+ this.valueMapper = valueMapper;
+ this.errorSchema = errorSchema;
}
- PubsubWriteSchemaTransform withPubsubClientFactory(PubsubClient.PubsubClientFactory factory) {
- this.pubsubClientFactory = factory;
- return this;
- }
-
- @Override
- public PCollectionRowTuple expand(PCollectionRowTuple input) {
- if (input.getAll().size() != 1 || !input.has(INPUT_TAG)) {
- throw new IllegalArgumentException(
- String.format(
- "%s %s input is expected to contain a single %s tagged PCollection<Row>",
- input.getClass().getSimpleName(), getClass().getSimpleName(), INPUT_TAG));
- }
-
- PCollection<Row> rows = input.get(INPUT_TAG);
- if (rows.getSchema().getFieldCount() == 0) {
- throw new IllegalArgumentException(String.format("empty Schema for %s", INPUT_TAG));
- }
-
- Schema targetSchema = buildTargetSchema(rows.getSchema());
-
- rows =
- rows.apply(
- ConvertForRowToMessage.class.getSimpleName(),
- convertForRowToMessage(targetSchema))
- .setRowSchema(targetSchema);
-
- Schema schema = rows.getSchema();
-
- Schema serializableSchema =
- removeFields(schema, DEFAULT_ATTRIBUTES_KEY_NAME, DEFAULT_EVENT_TIMESTAMP_KEY_NAME);
- FieldMatcher payloadRowMatcher = FieldMatcher.of(DEFAULT_PAYLOAD_KEY_NAME, TypeName.ROW);
- if (payloadRowMatcher.match(serializableSchema)) {
- serializableSchema =
- serializableSchema.getField(DEFAULT_PAYLOAD_KEY_NAME).getType().getRowSchema();
- }
-
- validateTargetSchemaAgainstPubsubSchema(serializableSchema, input.getPipeline().getOptions());
-
- PCollectionTuple pct =
- rows.apply(
- PubsubRowToMessage.class.getSimpleName(),
- buildPubsubRowToMessage(serializableSchema));
-
- PCollection<PubsubMessage> messages = pct.get(OUTPUT);
- messages.apply(PubsubIO.Write.class.getSimpleName(), buildPubsubWrite());
- return PCollectionRowTuple.of(ERROR_TAG, pct.get(ERROR));
- }
-
- PayloadSerializer getPayloadSerializer(Schema schema) {
- if (configuration.getFormat() == null) {
- return null;
- }
- String format = configuration.getFormat();
- Set<String> availableFormats =
- Providers.loadProviders(PayloadSerializerProvider.class).keySet();
- if (!availableFormats.contains(format)) {
- String availableFormatsString = String.join(",", availableFormats);
- throw new IllegalArgumentException(
- String.format(
- "%s is not among the valid formats: [%s]", format, availableFormatsString));
- }
- return PayloadSerializers.getSerializer(configuration.getFormat(), schema, ImmutableMap.of());
- }
-
- PubsubRowToMessage buildPubsubRowToMessage(Schema schema) {
- PubsubRowToMessage.Builder builder =
- PubsubRowToMessage.builder().setPayloadSerializer(getPayloadSerializer(schema));
-
- if (configuration.getTarget() != null) {
- builder =
- builder.setTargetTimestampAttributeName(
- configuration.getTarget().getTimestampAttributeKey());
- }
-
- return builder.build();
- }
-
- PubsubIO.Write<PubsubMessage> buildPubsubWrite() {
- PubsubIO.Write<PubsubMessage> write = PubsubIO.writeMessages().to(configuration.getTopic());
-
- if (configuration.getIdAttribute() != null) {
- write = write.withIdAttribute(configuration.getIdAttribute());
- }
-
- if (pubsubClientFactory != null) {
- write = write.withClientFactory(pubsubClientFactory);
+ @ProcessElement
+ public void processElement(@Element Row row, MultiOutputReceiver receiver) {
+ try {
+ receiver.get(OUTPUT_TAG).output(new PubsubMessage(valueMapper.apply(row), null));
+ } catch (Exception e) {
+ System.out.println("Error while parsing the element" + e.toString());
+ receiver
+ .get(ERROR_TAG)
+ .output(Row.withSchema(errorSchema).addValues(e.toString(), row).build());
}
-
- return write;
}
+ }
- void validateSourceSchemaAgainstConfiguration(Schema sourceSchema) {
- if (sourceSchema.getFieldCount() == 0) {
- throw new IllegalArgumentException(String.format("empty Schema for %s", INPUT_TAG));
- }
-
- if (configuration.getSource() == null) {
- return;
- }
-
- SourceConfiguration source = configuration.getSource();
-
- if (source.getAttributesFieldName() != null) {
- String fieldName = source.getAttributesFieldName();
- FieldType fieldType = ATTRIBUTES_FIELD_TYPE;
- FieldMatcher fieldMatcher = FieldMatcher.of(fieldName, fieldType);
- checkArgument(
- fieldMatcher.match(sourceSchema),
- String.format("schema missing field: %s for type %s: ", fieldName, fieldType));
- }
-
- if (source.getTimestampFieldName() != null) {
- String fieldName = source.getTimestampFieldName();
- FieldType fieldType = EVENT_TIMESTAMP_FIELD_TYPE;
- FieldMatcher fieldMatcher = FieldMatcher.of(fieldName, fieldType);
- checkArgument(
- fieldMatcher.match(sourceSchema),
- String.format("schema missing field: %s for type: %s", fieldName, fieldType));
- }
-
- if (source.getPayloadFieldName() == null) {
- return;
- }
-
- String fieldName = source.getPayloadFieldName();
- FieldMatcher bytesFieldMatcher = FieldMatcher.of(fieldName, PAYLOAD_BYTES_TYPE_NAME);
- FieldMatcher rowFieldMatcher = FieldMatcher.of(fieldName, PAYLOAD_ROW_TYPE_NAME);
- SchemaReflection schemaReflection = SchemaReflection.of(sourceSchema);
- checkArgument(
- schemaReflection.matchesAny(bytesFieldMatcher, rowFieldMatcher),
+ @Override
+ public SchemaTransform from(PubsubWriteSchemaTransformConfiguration configuration) {
+ if (!VALID_DATA_FORMATS.contains(configuration.getFormat())) {
+ throw new IllegalArgumentException(
String.format(
- "schema missing field: %s for types %s or %s",
- fieldName, PAYLOAD_BYTES_TYPE_NAME, PAYLOAD_ROW_TYPE_NAME));
-
- String[] fieldsToExclude =
- Stream.of(
- source.getAttributesFieldName(),
- source.getTimestampFieldName(),
- source.getPayloadFieldName())
- .filter(Objects::nonNull)
- .toArray(String[]::new);
-
- Schema userFieldsSchema = removeFields(sourceSchema, fieldsToExclude);
-
- if (userFieldsSchema.getFieldCount() > 0) {
- throw new IllegalArgumentException(
- String.format("user fields incompatible with %s field", source.getPayloadFieldName()));
- }
- }
-
- void validateTargetSchemaAgainstPubsubSchema(Schema targetSchema, PipelineOptions options) {
- checkArgument(options != null);
-
- try (PubsubClient pubsubClient = getPubsubClient(options.as(PubsubOptions.class))) {
- PubsubClient.TopicPath topicPath = PubsubClient.topicPathFromPath(configuration.getTopic());
- PubsubClient.SchemaPath schemaPath = pubsubClient.getSchemaPath(topicPath);
- if (schemaPath == null || schemaPath.equals(SchemaPath.DELETED_SCHEMA)) {
- return;
- }
- Schema expectedSchema = pubsubClient.getSchema(schemaPath);
- checkState(
- targetSchema.equals(expectedSchema),
- String.format(
- "input schema mismatch with expected schema at path: %s\ninput schema: %s\nPub/Sub schema: %s",
- schemaPath, targetSchema, expectedSchema));
- } catch (IOException e) {
- throw new IllegalStateException(e.getMessage());
- }
- }
-
- Schema buildTargetSchema(Schema sourceSchema) {
- validateSourceSchemaAgainstConfiguration(sourceSchema);
- FieldType payloadFieldType = null;
-
- List<String> fieldsToRemove = new ArrayList<>();
-
- if (configuration.getSource() != null) {
- SourceConfiguration source = configuration.getSource();
-
- if (source.getAttributesFieldName() != null) {
- fieldsToRemove.add(source.getAttributesFieldName());
- }
-
- if (source.getTimestampFieldName() != null) {
- fieldsToRemove.add(source.getTimestampFieldName());
- }
-
- if (source.getPayloadFieldName() != null) {
- String fieldName = source.getPayloadFieldName();
- Field field = sourceSchema.getField(fieldName);
- payloadFieldType = field.getType();
- fieldsToRemove.add(fieldName);
- }
- }
-
- Schema targetSchema =
- PubsubRowToMessage.builder()
- .build()
- .inputSchemaFactory(payloadFieldType)
- .buildSchema(sourceSchema.getFields().toArray(new Field[0]));
-
- return removeFields(targetSchema, fieldsToRemove.toArray(new String[0]));
+ "Format %s not supported. Only supported formats are %s",
+ configuration.getFormat(), VALID_FORMATS_STR));
}
+ return new PubsubWriteSchemaTransform(configuration.getTopic(), configuration.getFormat());
+ }
- private PubsubClient.PubsubClientFactory getPubsubClientFactory() {
- if (pubsubClientFactory != null) {
- return pubsubClientFactory;
- }
- return PubsubGrpcClient.FACTORY;
- }
+ private static class PubsubWriteSchemaTransform extends SchemaTransform implements Serializable {
+ final String topic;
+ final String format;
- private PubsubClient getPubsubClient(PubsubOptions options) throws IOException {
- return getPubsubClientFactory()
- .newClient(
- configuration.getTarget().getTimestampAttributeKey(),
- configuration.getIdAttribute(),
- options);
+ PubsubWriteSchemaTransform(String topic, String format) {
+ this.topic = topic;
+ this.format = format;
}
- ParDo.SingleOutput<Row, Row> convertForRowToMessage(Schema targetSchema) {
- return convertForRowToMessage(targetSchema, null);
+ public @UnknownKeyFor @NonNull @Initialized PTransform<
+ @UnknownKeyFor @NonNull @Initialized PCollectionRowTuple,
+ @UnknownKeyFor @NonNull @Initialized PCollectionRowTuple>
+ buildTransform() {
+ return this;
}
Review Comment:
Done
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubWriteSchemaTransformProvider.java:
##########
@@ -76,360 +50,113 @@
* provide no backwards compatibility guarantees, and it should not be implemented outside the Beam
* repository.
*/
-@SuppressWarnings({
- "nullness" // TODO(https://github.com/apache/beam/issues/20497)
-})
-@Internal
+// @SuppressWarnings({
+// "nullness" // TODO(https://github.com/apache/beam/issues/20497)
+// })
@AutoService(SchemaTransformProvider.class)
public class PubsubWriteSchemaTransformProvider
extends TypedSchemaTransformProvider<PubsubWriteSchemaTransformConfiguration> {
- private static final String IDENTIFIER = "beam:schematransform:org.apache.beam:pubsub_write:v1";
- static final String INPUT_TAG = "input";
- static final String ERROR_TAG = "error";
- /** Returns the expected class of the configuration. */
- @Override
- protected Class<PubsubWriteSchemaTransformConfiguration> configurationClass() {
- return PubsubWriteSchemaTransformConfiguration.class;
- }
+ public static final TupleTag<PubsubMessage> OUTPUT_TAG = new TupleTag<PubsubMessage>() {};
+ public static final TupleTag<Row> ERROR_TAG = new TupleTag<Row>() {};
- /** Returns the expected {@link SchemaTransform} of the configuration. */
- @Override
- public SchemaTransform from(PubsubWriteSchemaTransformConfiguration configuration) {
- return new PubsubWriteSchemaTransform(configuration);
- }
-
- /** Implementation of the {@link SchemaTransformProvider} identifier method. */
- @Override
- public String identifier() {
- return IDENTIFIER;
- }
-
- /**
- * Implementation of the {@link TypedSchemaTransformProvider} inputCollectionNames method. Since a
- * single input is expected, this returns a list with a single name.
- */
- @Override
- public List<String> inputCollectionNames() {
- return Collections.singletonList(INPUT_TAG);
- }
+ public static final String VALID_FORMATS_STR = "AVRO,JSON";
+ public static final Set<String> VALID_DATA_FORMATS =
+ Sets.newHashSet(VALID_FORMATS_STR.split(","));
- /**
- * Implementation of the {@link TypedSchemaTransformProvider} outputCollectionNames method. The
- * only expected output is the {@link #ERROR_TAG}.
- */
@Override
- public List<String> outputCollectionNames() {
- return Collections.singletonList(ERROR_TAG);
+ public Class<PubsubWriteSchemaTransformConfiguration> configurationClass() {
+ return PubsubWriteSchemaTransformConfiguration.class;
}
- /**
- * An implementation of {@link SchemaTransform} for Pub/Sub writes configured using {@link
- * PubsubWriteSchemaTransformConfiguration}.
- */
- static class PubsubWriteSchemaTransform extends SchemaTransform {
-
- private final PubsubWriteSchemaTransformConfiguration configuration;
+ public static class ErrorFn extends DoFn<Row, PubsubMessage> {
+ private SerializableFunction<Row, byte[]> valueMapper;
+ private Schema errorSchema;
- private PubsubClient.PubsubClientFactory pubsubClientFactory;
-
- PubsubWriteSchemaTransform(PubsubWriteSchemaTransformConfiguration configuration) {
- this.configuration = configuration;
+ ErrorFn(SerializableFunction<Row, byte[]> valueMapper, Schema errorSchema) {
+ this.valueMapper = valueMapper;
+ this.errorSchema = errorSchema;
}
- PubsubWriteSchemaTransform withPubsubClientFactory(PubsubClient.PubsubClientFactory factory) {
- this.pubsubClientFactory = factory;
- return this;
- }
-
- @Override
- public PCollectionRowTuple expand(PCollectionRowTuple input) {
- if (input.getAll().size() != 1 || !input.has(INPUT_TAG)) {
- throw new IllegalArgumentException(
- String.format(
- "%s %s input is expected to contain a single %s tagged PCollection<Row>",
- input.getClass().getSimpleName(), getClass().getSimpleName(), INPUT_TAG));
- }
-
- PCollection<Row> rows = input.get(INPUT_TAG);
- if (rows.getSchema().getFieldCount() == 0) {
- throw new IllegalArgumentException(String.format("empty Schema for %s", INPUT_TAG));
- }
-
- Schema targetSchema = buildTargetSchema(rows.getSchema());
-
- rows =
- rows.apply(
- ConvertForRowToMessage.class.getSimpleName(),
- convertForRowToMessage(targetSchema))
- .setRowSchema(targetSchema);
-
- Schema schema = rows.getSchema();
-
- Schema serializableSchema =
- removeFields(schema, DEFAULT_ATTRIBUTES_KEY_NAME, DEFAULT_EVENT_TIMESTAMP_KEY_NAME);
- FieldMatcher payloadRowMatcher = FieldMatcher.of(DEFAULT_PAYLOAD_KEY_NAME, TypeName.ROW);
- if (payloadRowMatcher.match(serializableSchema)) {
- serializableSchema =
- serializableSchema.getField(DEFAULT_PAYLOAD_KEY_NAME).getType().getRowSchema();
- }
-
- validateTargetSchemaAgainstPubsubSchema(serializableSchema, input.getPipeline().getOptions());
-
- PCollectionTuple pct =
- rows.apply(
- PubsubRowToMessage.class.getSimpleName(),
- buildPubsubRowToMessage(serializableSchema));
-
- PCollection<PubsubMessage> messages = pct.get(OUTPUT);
- messages.apply(PubsubIO.Write.class.getSimpleName(), buildPubsubWrite());
- return PCollectionRowTuple.of(ERROR_TAG, pct.get(ERROR));
- }
-
- PayloadSerializer getPayloadSerializer(Schema schema) {
- if (configuration.getFormat() == null) {
- return null;
- }
- String format = configuration.getFormat();
- Set<String> availableFormats =
- Providers.loadProviders(PayloadSerializerProvider.class).keySet();
- if (!availableFormats.contains(format)) {
- String availableFormatsString = String.join(",", availableFormats);
- throw new IllegalArgumentException(
- String.format(
- "%s is not among the valid formats: [%s]", format, availableFormatsString));
- }
- return PayloadSerializers.getSerializer(configuration.getFormat(), schema, ImmutableMap.of());
- }
-
- PubsubRowToMessage buildPubsubRowToMessage(Schema schema) {
- PubsubRowToMessage.Builder builder =
- PubsubRowToMessage.builder().setPayloadSerializer(getPayloadSerializer(schema));
-
- if (configuration.getTarget() != null) {
- builder =
- builder.setTargetTimestampAttributeName(
- configuration.getTarget().getTimestampAttributeKey());
- }
-
- return builder.build();
- }
-
- PubsubIO.Write<PubsubMessage> buildPubsubWrite() {
- PubsubIO.Write<PubsubMessage> write = PubsubIO.writeMessages().to(configuration.getTopic());
-
- if (configuration.getIdAttribute() != null) {
- write = write.withIdAttribute(configuration.getIdAttribute());
- }
-
- if (pubsubClientFactory != null) {
- write = write.withClientFactory(pubsubClientFactory);
+ @ProcessElement
+ public void processElement(@Element Row row, MultiOutputReceiver receiver) {
+ try {
+ receiver.get(OUTPUT_TAG).output(new PubsubMessage(valueMapper.apply(row), null));
+ } catch (Exception e) {
+ System.out.println("Error while parsing the element" + e.toString());
Review Comment:
Done
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubReadSchemaTransformProvider.java:
##########
@@ -43,196 +62,211 @@
* provide no backwards compatibility guarantees, and it should not be implemented outside the Beam
* repository.
*/
-@SuppressWarnings({
- "nullness" // TODO(https://github.com/apache/beam/issues/20497)
-})
-@Internal
@AutoService(SchemaTransformProvider.class)
public class PubsubReadSchemaTransformProvider
extends TypedSchemaTransformProvider<PubsubReadSchemaTransformConfiguration> {
- static final String OUTPUT_TAG = "OUTPUT";
- /** Returns the expected class of the configuration. */
- @Override
- protected Class<PubsubReadSchemaTransformConfiguration> configurationClass() {
- return PubsubReadSchemaTransformConfiguration.class;
- }
+ public static final String VALID_FORMATS_STR = "AVRO,JSON";
+ public static final Set<String> VALID_DATA_FORMATS =
+ Sets.newHashSet(VALID_FORMATS_STR.split(","));
- /** Returns the expected {@link SchemaTransform} of the configuration. */
- @Override
- protected SchemaTransform from(PubsubReadSchemaTransformConfiguration configuration) {
- PubsubMessageToRow toRowTransform =
- PubsubSchemaTransformMessageToRowFactory.from(configuration).buildMessageToRow();
- return new PubsubReadSchemaTransform(configuration, toRowTransform);
- }
-
- /** Implementation of the {@link TypedSchemaTransformProvider} identifier method. */
- @Override
- public String identifier() {
- return "beam:schematransform:org.apache.beam:pubsub_read:v1";
- }
+ public static final TupleTag<Row> OUTPUT_TAG = new TupleTag<Row>() {};
+ public static final TupleTag<Row> ERROR_TAG = new TupleTag<Row>() {};
+ public static final Schema ERROR_SCHEMA =
+ Schema.builder().addStringField("error").addNullableByteArrayField("row").build();
- /**
- * Implementation of the {@link TypedSchemaTransformProvider} inputCollectionNames method. Since
- * no input is expected, this returns an empty list.
- */
@Override
- public List<String> inputCollectionNames() {
- return Collections.emptyList();
+ public Class<PubsubReadSchemaTransformConfiguration> configurationClass() {
+ return PubsubReadSchemaTransformConfiguration.class;
}
- /**
- * Implementation of the {@link TypedSchemaTransformProvider} outputCollectionNames method. Since
- * a single output is expected, this returns a list with a single name.
- */
@Override
- public List<String> outputCollectionNames() {
- return Collections.singletonList(OUTPUT_TAG);
- }
-
- /**
- * An implementation of {@link SchemaTransform} for Pub/Sub reads configured using {@link
- * PubsubReadSchemaTransformConfiguration}.
- */
- static class PubsubReadSchemaTransform extends SchemaTransform {
-
- private final PubsubReadSchemaTransformConfiguration configuration;
- private final PubsubMessageToRow pubsubMessageToRow;
-
- private PubsubClient.PubsubClientFactory clientFactory;
-
- private Clock clock;
-
- private PubsubReadSchemaTransform(
- PubsubReadSchemaTransformConfiguration configuration,
- PubsubMessageToRow pubsubMessageToRow) {
- this.configuration = configuration;
- this.pubsubMessageToRow = pubsubMessageToRow;
+ public SchemaTransform from(PubsubReadSchemaTransformConfiguration configuration) {
+ if (configuration.getSubscription() == null && configuration.getTopic() == null) {
+ throw new IllegalArgumentException(
+ "To read from Pubsub, a subscription name or a topic name must be provided");
}
- /**
- * Sets the {@link PubsubClient.PubsubClientFactory}.
- *
- * <p>Used for testing.
- */
- void setClientFactory(PubsubClient.PubsubClientFactory value) {
- this.clientFactory = value;
+ if (configuration.getSubscription() != null && configuration.getTopic() != null) {
+ throw new IllegalArgumentException(
+ "To read from Pubsub, a subscription name or a topic name must be provided. Not both.");
}
- /**
- * Sets the {@link Clock}.
- *
- * <p>Used for testing.
- */
- void setClock(Clock clock) {
- this.clock = clock;
+ if ((Strings.isNullOrEmpty(configuration.getSchema())
+ && !Strings.isNullOrEmpty(configuration.getFormat()))
+ || (!Strings.isNullOrEmpty(configuration.getSchema())
+ && Strings.isNullOrEmpty(configuration.getFormat()))) {
+ throw new IllegalArgumentException(
+ "A schema was provided without a data format (or viceversa). Please provide "
+ + "both of these parameters to read from Pubsub, or if you would like to use the Pubsub schema service,"
+ + " please leave both of these blank.");
}
- /** Validates the {@link PubsubReadSchemaTransformConfiguration}. */
- @Override
- public void validate(@Nullable PipelineOptions options) {
- if (configuration.getSubscription() == null && configuration.getTopic() == null) {
- throw new IllegalArgumentException(
+ Schema beamSchema;
+ SerializableFunction<byte[], Row> valueMapper;
+ if (configuration.getSchema() == null && configuration.getFormat() == null) {
+ try {
+ KV<Schema, SerializableFunction<byte[], Row>> schemaFunctionPair =
+ PubsubUtils.getTopicInfo(configuration.getTopic(), configuration.getSubscription());
+ beamSchema = schemaFunctionPair.getKey();
+ valueMapper = schemaFunctionPair.getValue();
+ } catch (IOException e) {
+ throw new RuntimeException(
String.format(
- "%s needs to set either the topic or the subscription",
- PubsubReadSchemaTransformConfiguration.class));
+ "Unable to retrieve schema information for topic %s or subscription %s",
+ configuration.getTopic(), configuration.getSubscription()),
+ e);
}
-
- if (configuration.getSubscription() != null && configuration.getTopic() != null) {
+ } else {
+ if (!VALID_DATA_FORMATS.contains(configuration.getFormat())) {
throw new IllegalArgumentException(
String.format(
- "%s should not set both the topic or the subscription",
- PubsubReadSchemaTransformConfiguration.class));
- }
-
- try {
- PayloadSerializers.getSerializer(
- configuration.getFormat(), configuration.getDataSchema(), new HashMap<>());
- } catch (IllegalArgumentException e) {
- throw new IllegalArgumentException(
- String.format(
- "Invalid %s, no serializer provider exists for format `%s`",
- PubsubReadSchemaTransformConfiguration.class, configuration.getFormat()));
+ "Format %s not supported. Only supported formats are %s",
+ configuration.getFormat(), VALID_FORMATS_STR));
}
+ beamSchema =
+ Objects.equals(configuration.getFormat(), "JSON")
+ ? JsonUtils.beamSchemaFromJsonSchema(configuration.getSchema())
+ : AvroUtils.toBeamSchema(
+ new org.apache.avro.Schema.Parser().parse(configuration.getSchema()));
+ valueMapper =
+ Objects.equals(configuration.getFormat(), "JSON")
+ ? JsonUtils.getJsonBytesToRowFunction(beamSchema)
+ : AvroUtils.getAvroBytesToRowFunction(beamSchema);
}
- /** Reads from Pub/Sub according to {@link PubsubReadSchemaTransformConfiguration}. */
- @Override
- public PCollectionRowTuple expand(PCollectionRowTuple input) {
- if (!input.getAll().isEmpty()) {
- throw new IllegalArgumentException(
- String.format(
- "%s %s input is expected to be empty",
- input.getClass().getSimpleName(), getClass().getSimpleName()));
- }
+ PubsubReadSchemaTransform transform =
+ new PubsubReadSchemaTransform(
+ configuration.getTopic(), configuration.getSubscription(), beamSchema, valueMapper);
- PCollectionTuple rowsWithDlq =
- input
- .getPipeline()
- .apply("ReadFromPubsub", buildPubsubRead())
- .apply("PubsubMessageToRow", pubsubMessageToRow);
+ if (configuration.getClientFactory() != null) {
+ transform.setClientFactory(configuration.getClientFactory());
+ }
+ if (configuration.getClock() != null) {
+ transform.setClock(configuration.getClock());
+ }
- writeToDeadLetterQueue(rowsWithDlq);
+ return transform;
+ }
- return PCollectionRowTuple.of(OUTPUT_TAG, rowsWithDlq.get(MAIN_TAG));
+ private static class PubsubReadSchemaTransform extends SchemaTransform implements Serializable {
+ final Schema beamSchema;
+ final SerializableFunction<byte[], Row> valueMapper;
+ final @Nullable String topic;
+ final @Nullable String subscription;
+ @Nullable PubsubTestClientFactory clientFactory;
+ @Nullable Clock clock;
+
+ PubsubReadSchemaTransform(
+ @Nullable String topic,
+ @Nullable String subscription,
+ Schema beamSchema,
+ SerializableFunction<byte[], Row> valueMapper) {
+ this.topic = topic;
+ this.subscription = subscription;
+ this.beamSchema = beamSchema;
+ this.valueMapper = valueMapper;
}
- private void writeToDeadLetterQueue(PCollectionTuple rowsWithDlq) {
- PubsubIO.Write<PubsubMessage> deadLetterQueue = buildDeadLetterQueueWrite();
- if (deadLetterQueue == null) {
- return;
- }
- rowsWithDlq.get(DLQ_TAG).apply("WriteToDeadLetterQueue", deadLetterQueue);
- }
+ private static class ErrorCounterFn extends DoFn<PubsubMessage, Row> {
+ private Counter pubsubErrorCounter;
+ private Long errorsInBundle = 0L;
+ private SerializableFunction<byte[], Row> valueMapper;
- /**
- * Builds {@link PubsubIO.Write} dead letter queue from {@link
- * PubsubReadSchemaTransformConfiguration}.
- */
- PubsubIO.Write<PubsubMessage> buildDeadLetterQueueWrite() {
- if (configuration.getDeadLetterQueue() == null) {
- return null;
+ ErrorCounterFn(String name, SerializableFunction<byte[], Row> valueMapper) {
+ this.pubsubErrorCounter = Metrics.counter(PubsubReadSchemaTransformProvider.class, name);
+ this.valueMapper = valueMapper;
}
- PubsubIO.Write<PubsubMessage> writeDlq =
- PubsubIO.writeMessages().to(configuration.getDeadLetterQueue());
+ @ProcessElement
+ public void process(@DoFn.Element PubsubMessage message, MultiOutputReceiver receiver) {
+
+ try {
+ receiver.get(OUTPUT_TAG).output(valueMapper.apply(message.getPayload()));
+ } catch (Exception e) {
+ errorsInBundle += 1;
+ System.out.println("Error while parsing the element" + e.toString());
+ receiver
+ .get(ERROR_TAG)
+ .output(
+ Row.withSchema(ERROR_SCHEMA)
+ .addValues(e.toString(), message.getPayload())
+ .build());
+ }
+ }
- if (configuration.getTimestampAttribute() != null) {
- writeDlq = writeDlq.withTimestampAttribute(configuration.getTimestampAttribute());
+ @FinishBundle
+ public void finish(FinishBundleContext c) {
+ pubsubErrorCounter.inc(errorsInBundle);
+ errorsInBundle = 0L;
}
+ }
- return writeDlq;
+ void setClientFactory(@Nullable PubsubTestClientFactory factory) {
+ this.clientFactory = factory;
}
- /** Builds {@link PubsubIO.Read} from a {@link PubsubReadSchemaTransformConfiguration}. */
- PubsubIO.Read<PubsubMessage> buildPubsubRead() {
- PubsubIO.Read<PubsubMessage> read = PubsubIO.readMessagesWithAttributes();
+ void setClock(@Nullable Clock clock) {
+ this.clock = clock;
+ }
- if (configuration.getSubscription() != null) {
- read = read.fromSubscription(configuration.getSubscription());
+ @SuppressWarnings("nullness")
+ PubsubIO.Read<PubsubMessage> buildPubsubRead() {
+ PubsubIO.Read<PubsubMessage> pubsubRead = PubsubIO.readMessages();
+ if (!Strings.isNullOrEmpty(topic)) {
+ pubsubRead = pubsubRead.fromTopic(topic);
+ } else {
+ pubsubRead = pubsubRead.fromSubscription(subscription);
}
-
- if (configuration.getTopic() != null) {
- read = read.fromTopic(configuration.getTopic());
+ if (clientFactory != null && clock != null) {
+ pubsubRead = pubsubRead.withClientFactory(clientFactory);
+ pubsubRead = clientFactory.setClock(pubsubRead, clock);
+ } else if (clientFactory != null || clock != null) {
+ throw new IllegalArgumentException(
+ "Both PubsubTestClientFactory and Clock need to be specified for testing, but only one is provided");
}
+ return pubsubRead;
+ }
- if (configuration.getTimestampAttribute() != null) {
- read = read.withTimestampAttribute(configuration.getTimestampAttribute());
- }
+ // @Override
+ public PTransform<PCollectionRowTuple, PCollectionRowTuple> buildTransform() {
+ return this;
+ }
Review Comment:
Done
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubReadSchemaTransformProvider.java:
##########
@@ -43,196 +62,211 @@
* provide no backwards compatibility guarantees, and it should not be implemented outside the Beam
* repository.
*/
-@SuppressWarnings({
- "nullness" // TODO(https://github.com/apache/beam/issues/20497)
-})
-@Internal
@AutoService(SchemaTransformProvider.class)
public class PubsubReadSchemaTransformProvider
extends TypedSchemaTransformProvider<PubsubReadSchemaTransformConfiguration> {
- static final String OUTPUT_TAG = "OUTPUT";
- /** Returns the expected class of the configuration. */
- @Override
- protected Class<PubsubReadSchemaTransformConfiguration> configurationClass() {
- return PubsubReadSchemaTransformConfiguration.class;
- }
+ public static final String VALID_FORMATS_STR = "AVRO,JSON";
+ public static final Set<String> VALID_DATA_FORMATS =
+ Sets.newHashSet(VALID_FORMATS_STR.split(","));
- /** Returns the expected {@link SchemaTransform} of the configuration. */
- @Override
- protected SchemaTransform from(PubsubReadSchemaTransformConfiguration configuration) {
- PubsubMessageToRow toRowTransform =
- PubsubSchemaTransformMessageToRowFactory.from(configuration).buildMessageToRow();
- return new PubsubReadSchemaTransform(configuration, toRowTransform);
- }
-
- /** Implementation of the {@link TypedSchemaTransformProvider} identifier method. */
- @Override
- public String identifier() {
- return "beam:schematransform:org.apache.beam:pubsub_read:v1";
- }
+ public static final TupleTag<Row> OUTPUT_TAG = new TupleTag<Row>() {};
+ public static final TupleTag<Row> ERROR_TAG = new TupleTag<Row>() {};
+ public static final Schema ERROR_SCHEMA =
+ Schema.builder().addStringField("error").addNullableByteArrayField("row").build();
- /**
- * Implementation of the {@link TypedSchemaTransformProvider} inputCollectionNames method. Since
- * no input is expected, this returns an empty list.
- */
@Override
- public List<String> inputCollectionNames() {
- return Collections.emptyList();
+ public Class<PubsubReadSchemaTransformConfiguration> configurationClass() {
+ return PubsubReadSchemaTransformConfiguration.class;
}
- /**
- * Implementation of the {@link TypedSchemaTransformProvider} outputCollectionNames method. Since
- * a single output is expected, this returns a list with a single name.
- */
@Override
- public List<String> outputCollectionNames() {
- return Collections.singletonList(OUTPUT_TAG);
- }
-
- /**
- * An implementation of {@link SchemaTransform} for Pub/Sub reads configured using {@link
- * PubsubReadSchemaTransformConfiguration}.
- */
- static class PubsubReadSchemaTransform extends SchemaTransform {
-
- private final PubsubReadSchemaTransformConfiguration configuration;
- private final PubsubMessageToRow pubsubMessageToRow;
-
- private PubsubClient.PubsubClientFactory clientFactory;
-
- private Clock clock;
-
- private PubsubReadSchemaTransform(
- PubsubReadSchemaTransformConfiguration configuration,
- PubsubMessageToRow pubsubMessageToRow) {
- this.configuration = configuration;
- this.pubsubMessageToRow = pubsubMessageToRow;
+ public SchemaTransform from(PubsubReadSchemaTransformConfiguration configuration) {
+ if (configuration.getSubscription() == null && configuration.getTopic() == null) {
+ throw new IllegalArgumentException(
+ "To read from Pubsub, a subscription name or a topic name must be provided");
}
- /**
- * Sets the {@link PubsubClient.PubsubClientFactory}.
- *
- * <p>Used for testing.
- */
- void setClientFactory(PubsubClient.PubsubClientFactory value) {
- this.clientFactory = value;
+ if (configuration.getSubscription() != null && configuration.getTopic() != null) {
+ throw new IllegalArgumentException(
+ "To read from Pubsub, a subscription name or a topic name must be provided. Not both.");
}
- /**
- * Sets the {@link Clock}.
- *
- * <p>Used for testing.
- */
- void setClock(Clock clock) {
- this.clock = clock;
+ if ((Strings.isNullOrEmpty(configuration.getSchema())
+ && !Strings.isNullOrEmpty(configuration.getFormat()))
+ || (!Strings.isNullOrEmpty(configuration.getSchema())
+ && Strings.isNullOrEmpty(configuration.getFormat()))) {
+ throw new IllegalArgumentException(
+ "A schema was provided without a data format (or viceversa). Please provide "
+ + "both of these parameters to read from Pubsub, or if you would like to use the Pubsub schema service,"
+ + " please leave both of these blank.");
}
- /** Validates the {@link PubsubReadSchemaTransformConfiguration}. */
- @Override
- public void validate(@Nullable PipelineOptions options) {
- if (configuration.getSubscription() == null && configuration.getTopic() == null) {
- throw new IllegalArgumentException(
+ Schema beamSchema;
+ SerializableFunction<byte[], Row> valueMapper;
+ if (configuration.getSchema() == null && configuration.getFormat() == null) {
+ try {
+ KV<Schema, SerializableFunction<byte[], Row>> schemaFunctionPair =
+ PubsubUtils.getTopicInfo(configuration.getTopic(), configuration.getSubscription());
+ beamSchema = schemaFunctionPair.getKey();
+ valueMapper = schemaFunctionPair.getValue();
+ } catch (IOException e) {
+ throw new RuntimeException(
String.format(
- "%s needs to set either the topic or the subscription",
- PubsubReadSchemaTransformConfiguration.class));
+ "Unable to retrieve schema information for topic %s or subscription %s",
+ configuration.getTopic(), configuration.getSubscription()),
+ e);
}
-
- if (configuration.getSubscription() != null && configuration.getTopic() != null) {
+ } else {
+ if (!VALID_DATA_FORMATS.contains(configuration.getFormat())) {
throw new IllegalArgumentException(
String.format(
- "%s should not set both the topic or the subscription",
- PubsubReadSchemaTransformConfiguration.class));
- }
-
- try {
- PayloadSerializers.getSerializer(
- configuration.getFormat(), configuration.getDataSchema(), new HashMap<>());
- } catch (IllegalArgumentException e) {
- throw new IllegalArgumentException(
- String.format(
- "Invalid %s, no serializer provider exists for format `%s`",
- PubsubReadSchemaTransformConfiguration.class, configuration.getFormat()));
+ "Format %s not supported. Only supported formats are %s",
+ configuration.getFormat(), VALID_FORMATS_STR));
}
+ beamSchema =
+ Objects.equals(configuration.getFormat(), "JSON")
+ ? JsonUtils.beamSchemaFromJsonSchema(configuration.getSchema())
+ : AvroUtils.toBeamSchema(
+ new org.apache.avro.Schema.Parser().parse(configuration.getSchema()));
+ valueMapper =
+ Objects.equals(configuration.getFormat(), "JSON")
+ ? JsonUtils.getJsonBytesToRowFunction(beamSchema)
+ : AvroUtils.getAvroBytesToRowFunction(beamSchema);
}
- /** Reads from Pub/Sub according to {@link PubsubReadSchemaTransformConfiguration}. */
- @Override
- public PCollectionRowTuple expand(PCollectionRowTuple input) {
- if (!input.getAll().isEmpty()) {
- throw new IllegalArgumentException(
- String.format(
- "%s %s input is expected to be empty",
- input.getClass().getSimpleName(), getClass().getSimpleName()));
- }
+ PubsubReadSchemaTransform transform =
+ new PubsubReadSchemaTransform(
+ configuration.getTopic(), configuration.getSubscription(), beamSchema, valueMapper);
- PCollectionTuple rowsWithDlq =
- input
- .getPipeline()
- .apply("ReadFromPubsub", buildPubsubRead())
- .apply("PubsubMessageToRow", pubsubMessageToRow);
+ if (configuration.getClientFactory() != null) {
+ transform.setClientFactory(configuration.getClientFactory());
+ }
+ if (configuration.getClock() != null) {
+ transform.setClock(configuration.getClock());
+ }
- writeToDeadLetterQueue(rowsWithDlq);
+ return transform;
+ }
- return PCollectionRowTuple.of(OUTPUT_TAG, rowsWithDlq.get(MAIN_TAG));
+ private static class PubsubReadSchemaTransform extends SchemaTransform implements Serializable {
+ final Schema beamSchema;
+ final SerializableFunction<byte[], Row> valueMapper;
+ final @Nullable String topic;
+ final @Nullable String subscription;
+ @Nullable PubsubTestClientFactory clientFactory;
+ @Nullable Clock clock;
+
+ PubsubReadSchemaTransform(
+ @Nullable String topic,
+ @Nullable String subscription,
+ Schema beamSchema,
+ SerializableFunction<byte[], Row> valueMapper) {
+ this.topic = topic;
+ this.subscription = subscription;
+ this.beamSchema = beamSchema;
+ this.valueMapper = valueMapper;
}
- private void writeToDeadLetterQueue(PCollectionTuple rowsWithDlq) {
- PubsubIO.Write<PubsubMessage> deadLetterQueue = buildDeadLetterQueueWrite();
- if (deadLetterQueue == null) {
- return;
- }
- rowsWithDlq.get(DLQ_TAG).apply("WriteToDeadLetterQueue", deadLetterQueue);
- }
+ private static class ErrorCounterFn extends DoFn<PubsubMessage, Row> {
+ private Counter pubsubErrorCounter;
+ private Long errorsInBundle = 0L;
+ private SerializableFunction<byte[], Row> valueMapper;
- /**
- * Builds {@link PubsubIO.Write} dead letter queue from {@link
- * PubsubReadSchemaTransformConfiguration}.
- */
- PubsubIO.Write<PubsubMessage> buildDeadLetterQueueWrite() {
- if (configuration.getDeadLetterQueue() == null) {
- return null;
+ ErrorCounterFn(String name, SerializableFunction<byte[], Row> valueMapper) {
+ this.pubsubErrorCounter = Metrics.counter(PubsubReadSchemaTransformProvider.class, name);
+ this.valueMapper = valueMapper;
}
- PubsubIO.Write<PubsubMessage> writeDlq =
- PubsubIO.writeMessages().to(configuration.getDeadLetterQueue());
+ @ProcessElement
+ public void process(@DoFn.Element PubsubMessage message, MultiOutputReceiver receiver) {
+
+ try {
+ receiver.get(OUTPUT_TAG).output(valueMapper.apply(message.getPayload()));
+ } catch (Exception e) {
+ errorsInBundle += 1;
+ System.out.println("Error while parsing the element" + e.toString());
Review Comment:
Done
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] github-actions[bot] commented on pull request #27366: Moving Pubsub Transforms to beam
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #27366:
URL: https://github.com/apache/beam/pull/27366#issuecomment-1622349251
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] johnjcasey commented on a diff in pull request #27366: Moving Pubsub Transforms to beam
Posted by "johnjcasey (via GitHub)" <gi...@apache.org>.
johnjcasey commented on code in PR #27366:
URL: https://github.com/apache/beam/pull/27366#discussion_r1259900975
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubUtils.java:
##########
@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.pubsub;
+
+import com.google.cloud.pubsub.v1.SchemaServiceClient;
+import com.google.cloud.pubsub.v1.SubscriptionAdminClient;
+import com.google.cloud.pubsub.v1.TopicAdminClient;
+import com.google.pubsub.v1.Encoding;
+import com.google.pubsub.v1.Schema;
+import com.google.pubsub.v1.Topic;
+import java.io.IOException;
+import org.apache.beam.sdk.io.gcp.common.ProtoSchemaUtils;
+import org.apache.beam.sdk.schemas.utils.AvroUtils;
+import org.apache.beam.sdk.schemas.utils.JsonUtils;
+import org.apache.beam.sdk.transforms.SerializableFunction;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.Row;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+/** A utility class for working with Google Cloud Pub/Sub topics and subscriptions. */
+public class PubsubUtils {
+
+ /**
+ * Retrieves the schema and data conversion function for a Pub/Sub topic.
+ *
+ * @param topicName the name of the Pub/Sub topic
+ * @param subscriptionName the name of the subscription to the topic (optional)
+ * @return a {@link KV} object containing the schema and data conversion function
+ * @throws IOException if an error occurs while retrieving the schema from Pub/Sub
+ * @throws IllegalArgumentException if the schema or encoding is not supported
+ */
+ public static KV<org.apache.beam.sdk.schemas.Schema, SerializableFunction<byte[], Row>>
Review Comment:
we could mock it out, but writing integration tests is fine if mocking isn't feasible. For this scope, I think a high quality mock would be ideal
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ahmedabu98 commented on a diff in pull request #27366: Moving Pubsub Transforms to beam
Posted by "ahmedabu98 (via GitHub)" <gi...@apache.org>.
ahmedabu98 commented on code in PR #27366:
URL: https://github.com/apache/beam/pull/27366#discussion_r1256441783
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/common/ProtoSchemaUtils.java:
##########
@@ -0,0 +1,180 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.common;
+
+import com.squareup.wire.schema.Field;
+import com.squareup.wire.schema.Location;
+import com.squareup.wire.schema.internal.parser.EnumElement;
+import com.squareup.wire.schema.internal.parser.FieldElement;
+import com.squareup.wire.schema.internal.parser.MessageElement;
+import com.squareup.wire.schema.internal.parser.ProtoFileElement;
+import com.squareup.wire.schema.internal.parser.ProtoParser;
+import com.squareup.wire.schema.internal.parser.TypeElement;
+import java.util.Map;
+import java.util.Objects;
+import java.util.stream.Collectors;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.schemas.logicaltypes.EnumerationType;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+/**
+ * The ProtoSchemaUtils class provides utility methods for converting Protocol Buffers schema
+ * definitions into Beam schemas.
+ */
+public class ProtoSchemaUtils {
+
+ private static final Location DEFAULT_LOCATION = Location.get("");
+
+ /**
+ * Converts a Protocol Buffers schema definition into a Beam schema.
+ *
+ * @param protobufDef the Protocol Buffers schema definition as a string.
+ * @return the corresponding Beam schema.
+ * @throws SchemaParseException if the Protocol Buffers schema is invalid.
+ */
+ public static Schema beamSchemaFromProtoSchemaDescription(String protobufDef)
+ throws SchemaParseException {
+ ProtoFileElement elm = ProtoParser.Companion.parse(DEFAULT_LOCATION, protobufDef);
+ if (elm.getTypes().size() != 1) {
+ throw new SchemaParseException(
+ String.format(
+ "Expected a single type defined in schema input, but found %s. "
+ + "Provide only a single top-level proto type to use as the "
+ + "message schema.",
+ elm.getTypes().size()));
+ }
+ TypeElement typeElement = elm.getTypes().stream().findFirst().get();
+
+ if (!(typeElement instanceof MessageElement)) {
+ throw new SchemaParseException(
+ "Expected a message type defined, but found a different type definition for "
+ + typeElement.getName());
+ }
+
+ return parsedProtoSchemaToBeamSchema(typeElement, null);
+ }
+
+ /**
+ * Converts a parsed Protocol Buffers schema to a Beam schema.
+ *
+ * @param typeElement the type element representing the Protocol Buffers schema.
+ * @param typeCatalog a catalog of types referenced in the schema.
+ * @return the corresponding Beam schema.
+ */
+ @SuppressWarnings("nullness")
Review Comment:
You can use Optional.ofNullable(<nullable object>).get() to take care of the checks. Example: https://github.com/apache/beam/blob/c06247bf79ec3fd02810c26b7f0da58f6a790892/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableWriteSchemaTransformProvider.java#L168C28-L168C28
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ahmedabu98 commented on a diff in pull request #27366: Moving Pubsub Transforms to beam
Posted by "ahmedabu98 (via GitHub)" <gi...@apache.org>.
ahmedabu98 commented on code in PR #27366:
URL: https://github.com/apache/beam/pull/27366#discussion_r1256441783
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/common/ProtoSchemaUtils.java:
##########
@@ -0,0 +1,180 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.common;
+
+import com.squareup.wire.schema.Field;
+import com.squareup.wire.schema.Location;
+import com.squareup.wire.schema.internal.parser.EnumElement;
+import com.squareup.wire.schema.internal.parser.FieldElement;
+import com.squareup.wire.schema.internal.parser.MessageElement;
+import com.squareup.wire.schema.internal.parser.ProtoFileElement;
+import com.squareup.wire.schema.internal.parser.ProtoParser;
+import com.squareup.wire.schema.internal.parser.TypeElement;
+import java.util.Map;
+import java.util.Objects;
+import java.util.stream.Collectors;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.schemas.logicaltypes.EnumerationType;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+/**
+ * The ProtoSchemaUtils class provides utility methods for converting Protocol Buffers schema
+ * definitions into Beam schemas.
+ */
+public class ProtoSchemaUtils {
+
+ private static final Location DEFAULT_LOCATION = Location.get("");
+
+ /**
+ * Converts a Protocol Buffers schema definition into a Beam schema.
+ *
+ * @param protobufDef the Protocol Buffers schema definition as a string.
+ * @return the corresponding Beam schema.
+ * @throws SchemaParseException if the Protocol Buffers schema is invalid.
+ */
+ public static Schema beamSchemaFromProtoSchemaDescription(String protobufDef)
+ throws SchemaParseException {
+ ProtoFileElement elm = ProtoParser.Companion.parse(DEFAULT_LOCATION, protobufDef);
+ if (elm.getTypes().size() != 1) {
+ throw new SchemaParseException(
+ String.format(
+ "Expected a single type defined in schema input, but found %s. "
+ + "Provide only a single top-level proto type to use as the "
+ + "message schema.",
+ elm.getTypes().size()));
+ }
+ TypeElement typeElement = elm.getTypes().stream().findFirst().get();
+
+ if (!(typeElement instanceof MessageElement)) {
+ throw new SchemaParseException(
+ "Expected a message type defined, but found a different type definition for "
+ + typeElement.getName());
+ }
+
+ return parsedProtoSchemaToBeamSchema(typeElement, null);
+ }
+
+ /**
+ * Converts a parsed Protocol Buffers schema to a Beam schema.
+ *
+ * @param typeElement the type element representing the Protocol Buffers schema.
+ * @param typeCatalog a catalog of types referenced in the schema.
+ * @return the corresponding Beam schema.
+ */
+ @SuppressWarnings("nullness")
Review Comment:
You can use Optional.ofNullable(<nullable_object>).get() to take care of the checks. Example: https://github.com/apache/beam/blob/c06247bf79ec3fd02810c26b7f0da58f6a790892/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableWriteSchemaTransformProvider.java#L168C28-L168C28
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] Dippatel98 commented on a diff in pull request #27366: Moving Pubsub Transforms to beam
Posted by "Dippatel98 (via GitHub)" <gi...@apache.org>.
Dippatel98 commented on code in PR #27366:
URL: https://github.com/apache/beam/pull/27366#discussion_r1256468633
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/pubsub/PubsubWriteSchemaTransformProvider.java:
##########
@@ -76,360 +50,113 @@
* provide no backwards compatibility guarantees, and it should not be implemented outside the Beam
* repository.
*/
-@SuppressWarnings({
- "nullness" // TODO(https://github.com/apache/beam/issues/20497)
-})
-@Internal
+// @SuppressWarnings({
Review Comment:
Done
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] Dippatel98 commented on a diff in pull request #27366: Moving Pubsub Transforms to beam
Posted by "Dippatel98 (via GitHub)" <gi...@apache.org>.
Dippatel98 commented on code in PR #27366:
URL: https://github.com/apache/beam/pull/27366#discussion_r1256497434
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/common/ProtoSchemaUtils.java:
##########
@@ -0,0 +1,180 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.common;
+
+import com.squareup.wire.schema.Field;
+import com.squareup.wire.schema.Location;
+import com.squareup.wire.schema.internal.parser.EnumElement;
+import com.squareup.wire.schema.internal.parser.FieldElement;
+import com.squareup.wire.schema.internal.parser.MessageElement;
+import com.squareup.wire.schema.internal.parser.ProtoFileElement;
+import com.squareup.wire.schema.internal.parser.ProtoParser;
+import com.squareup.wire.schema.internal.parser.TypeElement;
+import java.util.Map;
+import java.util.Objects;
+import java.util.stream.Collectors;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.schemas.logicaltypes.EnumerationType;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+/**
+ * The ProtoSchemaUtils class provides utility methods for converting Protocol Buffers schema
+ * definitions into Beam schemas.
+ */
+public class ProtoSchemaUtils {
Review Comment:
Done
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] Dippatel98 commented on a diff in pull request #27366: Moving Pubsub Transforms to beam
Posted by "Dippatel98 (via GitHub)" <gi...@apache.org>.
Dippatel98 commented on code in PR #27366:
URL: https://github.com/apache/beam/pull/27366#discussion_r1256555948
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/common/ProtoSchemaUtils.java:
##########
@@ -0,0 +1,180 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.common;
+
+import com.squareup.wire.schema.Field;
+import com.squareup.wire.schema.Location;
+import com.squareup.wire.schema.internal.parser.EnumElement;
+import com.squareup.wire.schema.internal.parser.FieldElement;
+import com.squareup.wire.schema.internal.parser.MessageElement;
+import com.squareup.wire.schema.internal.parser.ProtoFileElement;
+import com.squareup.wire.schema.internal.parser.ProtoParser;
+import com.squareup.wire.schema.internal.parser.TypeElement;
+import java.util.Map;
+import java.util.Objects;
+import java.util.stream.Collectors;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.schemas.logicaltypes.EnumerationType;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+/**
+ * The ProtoSchemaUtils class provides utility methods for converting Protocol Buffers schema
+ * definitions into Beam schemas.
+ */
+public class ProtoSchemaUtils {
+
+ private static final Location DEFAULT_LOCATION = Location.get("");
+
+ /**
+ * Converts a Protocol Buffers schema definition into a Beam schema.
+ *
+ * @param protobufDef the Protocol Buffers schema definition as a string.
+ * @return the corresponding Beam schema.
+ * @throws SchemaParseException if the Protocol Buffers schema is invalid.
+ */
+ public static Schema beamSchemaFromProtoSchemaDescription(String protobufDef)
+ throws SchemaParseException {
+ ProtoFileElement elm = ProtoParser.Companion.parse(DEFAULT_LOCATION, protobufDef);
+ if (elm.getTypes().size() != 1) {
+ throw new SchemaParseException(
+ String.format(
+ "Expected a single type defined in schema input, but found %s. "
+ + "Provide only a single top-level proto type to use as the "
+ + "message schema.",
+ elm.getTypes().size()));
+ }
+ TypeElement typeElement = elm.getTypes().stream().findFirst().get();
+
+ if (!(typeElement instanceof MessageElement)) {
+ throw new SchemaParseException(
+ "Expected a message type defined, but found a different type definition for "
+ + typeElement.getName());
+ }
+
+ return parsedProtoSchemaToBeamSchema(typeElement, null);
+ }
+
+ /**
+ * Converts a parsed Protocol Buffers schema to a Beam schema.
+ *
+ * @param typeElement the type element representing the Protocol Buffers schema.
+ * @param typeCatalog a catalog of types referenced in the schema.
+ * @return the corresponding Beam schema.
+ */
+ @SuppressWarnings("nullness")
Review Comment:
Ah, thanks @ahmedabu98. Made the change
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] Dippatel98 commented on pull request #27366: Moving Pubsub Transforms to beam
Posted by "Dippatel98 (via GitHub)" <gi...@apache.org>.
Dippatel98 commented on PR #27366:
URL: https://github.com/apache/beam/pull/27366#issuecomment-1638592085
Run Java_GCP_IO_Direct 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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] johnjcasey merged pull request #27366: Moving Pubsub Transforms to beam
Posted by "johnjcasey (via GitHub)" <gi...@apache.org>.
johnjcasey merged PR #27366:
URL: https://github.com/apache/beam/pull/27366
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] Dippatel98 commented on a diff in pull request #27366: Moving Pubsub Transforms to beam
Posted by "Dippatel98 (via GitHub)" <gi...@apache.org>.
Dippatel98 commented on code in PR #27366:
URL: https://github.com/apache/beam/pull/27366#discussion_r1256429111
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/common/ProtoSchemaUtils.java:
##########
@@ -0,0 +1,180 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.common;
+
+import com.squareup.wire.schema.Field;
+import com.squareup.wire.schema.Location;
+import com.squareup.wire.schema.internal.parser.EnumElement;
+import com.squareup.wire.schema.internal.parser.FieldElement;
+import com.squareup.wire.schema.internal.parser.MessageElement;
+import com.squareup.wire.schema.internal.parser.ProtoFileElement;
+import com.squareup.wire.schema.internal.parser.ProtoParser;
+import com.squareup.wire.schema.internal.parser.TypeElement;
+import java.util.Map;
+import java.util.Objects;
+import java.util.stream.Collectors;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.schemas.logicaltypes.EnumerationType;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+/**
+ * The ProtoSchemaUtils class provides utility methods for converting Protocol Buffers schema
+ * definitions into Beam schemas.
+ */
+public class ProtoSchemaUtils {
+
+ private static final Location DEFAULT_LOCATION = Location.get("");
+
+ /**
+ * Converts a Protocol Buffers schema definition into a Beam schema.
+ *
+ * @param protobufDef the Protocol Buffers schema definition as a string.
+ * @return the corresponding Beam schema.
+ * @throws SchemaParseException if the Protocol Buffers schema is invalid.
+ */
+ public static Schema beamSchemaFromProtoSchemaDescription(String protobufDef)
+ throws SchemaParseException {
+ ProtoFileElement elm = ProtoParser.Companion.parse(DEFAULT_LOCATION, protobufDef);
+ if (elm.getTypes().size() != 1) {
+ throw new SchemaParseException(
+ String.format(
+ "Expected a single type defined in schema input, but found %s. "
+ + "Provide only a single top-level proto type to use as the "
+ + "message schema.",
+ elm.getTypes().size()));
+ }
+ TypeElement typeElement = elm.getTypes().stream().findFirst().get();
+
+ if (!(typeElement instanceof MessageElement)) {
+ throw new SchemaParseException(
+ "Expected a message type defined, but found a different type definition for "
+ + typeElement.getName());
+ }
+
+ return parsedProtoSchemaToBeamSchema(typeElement, null);
+ }
+
+ /**
+ * Converts a parsed Protocol Buffers schema to a Beam schema.
+ *
+ * @param typeElement the type element representing the Protocol Buffers schema.
+ * @param typeCatalog a catalog of types referenced in the schema.
+ * @return the corresponding Beam schema.
+ */
+ @SuppressWarnings("nullness")
Review Comment:
If I don't add any suppression, then [line 98](https://github.com/apache/beam/pull/27366/files/79342ae4afbcb6101b19fba6b623ee53418978ca#diff-d7a7d5fd76e08aa3ede8a967b406a5fb0896d7026b079f1ba32fcfd522850b62R98) complains about dereferencing null values. If you look at the line, the if condition explicitly checks for nulls and wouldn't proceed if `f.getLable()` is null.
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org