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/06/01 19:33:38 UTC
[GitHub] [beam] Dippatel98 opened a new pull request, #26970: Added dlq and error metrics to SpannerChangestreamsReadSchemaTransformProvider
Dippatel98 opened a new pull request, #26970:
URL: https://github.com/apache/beam/pull/26970
I have added dlq support in the Spanner Changestream read schema transform by tagging with an appropriate TupleTag. I have also added a Counter to keep track of the number of failed elements in a bundle.
------------------------
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] johnjcasey commented on a diff in pull request #26970: Added dlq and error metrics to SpannerChangestreamsReadSchemaTransformProvider
Posted by "johnjcasey (via GitHub)" <gi...@apache.org>.
johnjcasey commented on code in PR #26970:
URL: https://github.com/apache/beam/pull/26970#discussion_r1218395353
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/SpannerChangestreamsReadSchemaTransformProvider.java:
##########
@@ -193,15 +218,19 @@ public abstract static class Builder {
}
}
- private static final class DataChangeRecordToRow extends DoFn<DataChangeRecord, Row> {
+ public static final class DataChangeRecordToRow extends DoFn<DataChangeRecord, Row> {
Review Comment:
ty
--
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 #26970: Added dlq and error metrics to SpannerChangestreamsReadSchemaTransformProvider
Posted by "johnjcasey (via GitHub)" <gi...@apache.org>.
johnjcasey merged PR #26970:
URL: https://github.com/apache/beam/pull/26970
--
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 #26970: Added dlq and error metrics to SpannerChangestreamsReadSchemaTransformProvider
Posted by "Dippatel98 (via GitHub)" <gi...@apache.org>.
Dippatel98 commented on code in PR #26970:
URL: https://github.com/apache/beam/pull/26970#discussion_r1218391374
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/SpannerChangestreamsReadSchemaTransformProvider.java:
##########
@@ -235,41 +264,58 @@ public void process(@DoFn.Element DataChangeRecord record, OutputReceiver<Row> r
.map(nonNullValues -> getGson().fromJson(nonNullValues, Map.class))
.orElseGet(Collections::emptyMap);
- for (Map.Entry<String, String> valueEntry : newValues.entrySet()) {
- if (valueEntry.getValue() == null) {
- continue;
+ try {
+ for (Map.Entry<String, String> valueEntry : newValues.entrySet()) {
+ if (valueEntry.getValue() == null) {
+ continue;
+ }
+ // TODO(pabloem): Understand why SpannerSchema has field names in lowercase...
+ rowBuilder =
+ rowBuilder.withFieldValue(
+ valueEntry.getKey().toLowerCase(),
+ stringToParsedValue(
+ internalRowSchema.getField(valueEntry.getKey().toLowerCase()).getType(),
+ valueEntry.getValue()));
}
- // TODO(pabloem): Understand why SpannerSchema has field names in lowercase...
- rowBuilder =
- rowBuilder.withFieldValue(
- valueEntry.getKey().toLowerCase(),
- stringToParsedValue(
- internalRowSchema.getField(valueEntry.getKey().toLowerCase()).getType(),
- valueEntry.getValue()));
- }
- for (Map.Entry<String, String> pkEntry : keyValues.entrySet()) {
- if (pkEntry.getValue() == null) {
- continue;
+ for (Map.Entry<String, String> pkEntry : keyValues.entrySet()) {
+ if (pkEntry.getValue() == null) {
+ continue;
+ }
+ // TODO(pabloem): Understand why SpannerSchema has field names in lowercase...
+ rowBuilder =
+ rowBuilder.withFieldValue(
+ pkEntry.getKey().toLowerCase(),
+ stringToParsedValue(
+ internalRowSchema.getField(pkEntry.getKey().toLowerCase()).getType(),
+ pkEntry.getValue()));
}
- // TODO(pabloem): Understand why SpannerSchema has field names in lowercase...
- rowBuilder =
- rowBuilder.withFieldValue(
- pkEntry.getKey().toLowerCase(),
- stringToParsedValue(
- internalRowSchema.getField(pkEntry.getKey().toLowerCase()).getType(),
- pkEntry.getValue()));
+ receiver
+ .get(OUTPUT_TAG)
+ .outputWithTimestamp(
+ Row.withSchema(tableChangeRecordSchema)
+ .addValue(record.getModType().toString())
+ .addValue(record.getCommitTimestamp().toString())
+ .addValue(Long.parseLong(record.getRecordSequence()))
+ .addValue(rowBuilder.build())
+ .build(),
+ timestamp);
+ } catch (Exception e) {
+ errorsInBundle += 1;
Review Comment:
The possibility of both of those are pretty low. But for sanity's sake I can add those statements inside the try block. Let me know.
--
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 #26970: Added dlq and error metrics to SpannerChangestreamsReadSchemaTransformProvider
Posted by "johnjcasey (via GitHub)" <gi...@apache.org>.
johnjcasey commented on code in PR #26970:
URL: https://github.com/apache/beam/pull/26970#discussion_r1218377814
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/SpannerChangestreamsReadSchemaTransformProvider.java:
##########
@@ -193,15 +218,19 @@ public abstract static class Builder {
}
}
- private static final class DataChangeRecordToRow extends DoFn<DataChangeRecord, Row> {
+ public static final class DataChangeRecordToRow extends DoFn<DataChangeRecord, Row> {
Review Comment:
Also, you should create a test case with the scenario that artificially forces use of the DLQ, it shouldn't remain untested
--
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 #26970: Added dlq and error metrics to SpannerChangestreamsReadSchemaTransformProvider
Posted by "Dippatel98 (via GitHub)" <gi...@apache.org>.
Dippatel98 commented on PR #26970:
URL: https://github.com/apache/beam/pull/26970#issuecomment-1572661379
r: @manavgarg, @johnjcasey, @nickuncaged1201
--
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 #26970: Added dlq and error metrics to SpannerChangestreamsReadSchemaTransformProvider
Posted by "Dippatel98 (via GitHub)" <gi...@apache.org>.
Dippatel98 commented on code in PR #26970:
URL: https://github.com/apache/beam/pull/26970#discussion_r1218315758
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/SpannerChangestreamsReadSchemaTransformProvider.java:
##########
@@ -193,15 +218,19 @@ public abstract static class Builder {
}
}
- private static final class DataChangeRecordToRow extends DoFn<DataChangeRecord, Row> {
+ public static final class DataChangeRecordToRow extends DoFn<DataChangeRecord, Row> {
Review Comment:
In order to test it in the SpannerChangestreamsReadDlqTest. There didn't seem a straight forward way to test the dlq. So I tested the DoFn itself.
--
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 #26970: Added dlq and error metrics to SpannerChangestreamsReadSchemaTransformProvider
Posted by "Dippatel98 (via GitHub)" <gi...@apache.org>.
Dippatel98 commented on code in PR #26970:
URL: https://github.com/apache/beam/pull/26970#discussion_r1218387751
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/SpannerChangestreamsReadSchemaTransformProvider.java:
##########
@@ -235,41 +264,58 @@ public void process(@DoFn.Element DataChangeRecord record, OutputReceiver<Row> r
.map(nonNullValues -> getGson().fromJson(nonNullValues, Map.class))
.orElseGet(Collections::emptyMap);
- for (Map.Entry<String, String> valueEntry : newValues.entrySet()) {
- if (valueEntry.getValue() == null) {
- continue;
+ try {
+ for (Map.Entry<String, String> valueEntry : newValues.entrySet()) {
+ if (valueEntry.getValue() == null) {
+ continue;
+ }
+ // TODO(pabloem): Understand why SpannerSchema has field names in lowercase...
+ rowBuilder =
+ rowBuilder.withFieldValue(
+ valueEntry.getKey().toLowerCase(),
+ stringToParsedValue(
+ internalRowSchema.getField(valueEntry.getKey().toLowerCase()).getType(),
+ valueEntry.getValue()));
}
- // TODO(pabloem): Understand why SpannerSchema has field names in lowercase...
- rowBuilder =
- rowBuilder.withFieldValue(
- valueEntry.getKey().toLowerCase(),
- stringToParsedValue(
- internalRowSchema.getField(valueEntry.getKey().toLowerCase()).getType(),
- valueEntry.getValue()));
- }
- for (Map.Entry<String, String> pkEntry : keyValues.entrySet()) {
- if (pkEntry.getValue() == null) {
- continue;
+ for (Map.Entry<String, String> pkEntry : keyValues.entrySet()) {
+ if (pkEntry.getValue() == null) {
+ continue;
+ }
+ // TODO(pabloem): Understand why SpannerSchema has field names in lowercase...
+ rowBuilder =
+ rowBuilder.withFieldValue(
+ pkEntry.getKey().toLowerCase(),
+ stringToParsedValue(
+ internalRowSchema.getField(pkEntry.getKey().toLowerCase()).getType(),
+ pkEntry.getValue()));
}
- // TODO(pabloem): Understand why SpannerSchema has field names in lowercase...
- rowBuilder =
- rowBuilder.withFieldValue(
- pkEntry.getKey().toLowerCase(),
- stringToParsedValue(
- internalRowSchema.getField(pkEntry.getKey().toLowerCase()).getType(),
- pkEntry.getValue()));
+ receiver
+ .get(OUTPUT_TAG)
+ .outputWithTimestamp(
+ Row.withSchema(tableChangeRecordSchema)
+ .addValue(record.getModType().toString())
+ .addValue(record.getCommitTimestamp().toString())
+ .addValue(Long.parseLong(record.getRecordSequence()))
+ .addValue(rowBuilder.build())
+ .build(),
+ timestamp);
+ } catch (Exception e) {
+ errorsInBundle += 1;
Review Comment:
I am not sure. Is it allowed to have two fields with the same name in spanner? If yes then there can be, other possibility is if the SpannerIO messes up.
--
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 #26970: Added dlq and error metrics to SpannerChangestreamsReadSchemaTransformProvider
Posted by "Dippatel98 (via GitHub)" <gi...@apache.org>.
Dippatel98 commented on code in PR #26970:
URL: https://github.com/apache/beam/pull/26970#discussion_r1218324568
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/SpannerChangestreamsReadSchemaTransformProvider.java:
##########
@@ -235,41 +264,58 @@ public void process(@DoFn.Element DataChangeRecord record, OutputReceiver<Row> r
.map(nonNullValues -> getGson().fromJson(nonNullValues, Map.class))
.orElseGet(Collections::emptyMap);
- for (Map.Entry<String, String> valueEntry : newValues.entrySet()) {
- if (valueEntry.getValue() == null) {
- continue;
+ try {
+ for (Map.Entry<String, String> valueEntry : newValues.entrySet()) {
+ if (valueEntry.getValue() == null) {
+ continue;
+ }
+ // TODO(pabloem): Understand why SpannerSchema has field names in lowercase...
+ rowBuilder =
+ rowBuilder.withFieldValue(
+ valueEntry.getKey().toLowerCase(),
+ stringToParsedValue(
+ internalRowSchema.getField(valueEntry.getKey().toLowerCase()).getType(),
+ valueEntry.getValue()));
}
- // TODO(pabloem): Understand why SpannerSchema has field names in lowercase...
- rowBuilder =
- rowBuilder.withFieldValue(
- valueEntry.getKey().toLowerCase(),
- stringToParsedValue(
- internalRowSchema.getField(valueEntry.getKey().toLowerCase()).getType(),
- valueEntry.getValue()));
- }
- for (Map.Entry<String, String> pkEntry : keyValues.entrySet()) {
- if (pkEntry.getValue() == null) {
- continue;
+ for (Map.Entry<String, String> pkEntry : keyValues.entrySet()) {
+ if (pkEntry.getValue() == null) {
+ continue;
+ }
+ // TODO(pabloem): Understand why SpannerSchema has field names in lowercase...
+ rowBuilder =
+ rowBuilder.withFieldValue(
+ pkEntry.getKey().toLowerCase(),
+ stringToParsedValue(
+ internalRowSchema.getField(pkEntry.getKey().toLowerCase()).getType(),
+ pkEntry.getValue()));
}
- // TODO(pabloem): Understand why SpannerSchema has field names in lowercase...
- rowBuilder =
- rowBuilder.withFieldValue(
- pkEntry.getKey().toLowerCase(),
- stringToParsedValue(
- internalRowSchema.getField(pkEntry.getKey().toLowerCase()).getType(),
- pkEntry.getValue()));
+ receiver
+ .get(OUTPUT_TAG)
+ .outputWithTimestamp(
+ Row.withSchema(tableChangeRecordSchema)
+ .addValue(record.getModType().toString())
+ .addValue(record.getCommitTimestamp().toString())
+ .addValue(Long.parseLong(record.getRecordSequence()))
+ .addValue(rowBuilder.build())
+ .build(),
+ timestamp);
+ } catch (Exception e) {
+ errorsInBundle += 1;
Review Comment:
I discovered that it is possible for the spanner schema to change while this transform is running. It is a case which is handled in the SpannerIO successfully, but our transform doesn't handle it. That is the only mode of failure that I could come up with. As for the .getKeysJson and .getNewValuesJson they are returning the spanner row as JSON from the Mod and mapping it to a hash map. The Mod is created by the I/O itself, so I don't think there will be any formatting issues there and converting them to a hash map should be straightforward.
--
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 #26970: Added dlq and error metrics to SpannerChangestreamsReadSchemaTransformProvider
Posted by "johnjcasey (via GitHub)" <gi...@apache.org>.
johnjcasey commented on code in PR #26970:
URL: https://github.com/apache/beam/pull/26970#discussion_r1218396209
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/SpannerChangestreamsReadSchemaTransformProvider.java:
##########
@@ -235,41 +264,58 @@ public void process(@DoFn.Element DataChangeRecord record, OutputReceiver<Row> r
.map(nonNullValues -> getGson().fromJson(nonNullValues, Map.class))
.orElseGet(Collections::emptyMap);
- for (Map.Entry<String, String> valueEntry : newValues.entrySet()) {
- if (valueEntry.getValue() == null) {
- continue;
+ try {
+ for (Map.Entry<String, String> valueEntry : newValues.entrySet()) {
+ if (valueEntry.getValue() == null) {
+ continue;
+ }
+ // TODO(pabloem): Understand why SpannerSchema has field names in lowercase...
+ rowBuilder =
+ rowBuilder.withFieldValue(
+ valueEntry.getKey().toLowerCase(),
+ stringToParsedValue(
+ internalRowSchema.getField(valueEntry.getKey().toLowerCase()).getType(),
+ valueEntry.getValue()));
}
- // TODO(pabloem): Understand why SpannerSchema has field names in lowercase...
- rowBuilder =
- rowBuilder.withFieldValue(
- valueEntry.getKey().toLowerCase(),
- stringToParsedValue(
- internalRowSchema.getField(valueEntry.getKey().toLowerCase()).getType(),
- valueEntry.getValue()));
- }
- for (Map.Entry<String, String> pkEntry : keyValues.entrySet()) {
- if (pkEntry.getValue() == null) {
- continue;
+ for (Map.Entry<String, String> pkEntry : keyValues.entrySet()) {
+ if (pkEntry.getValue() == null) {
+ continue;
+ }
+ // TODO(pabloem): Understand why SpannerSchema has field names in lowercase...
+ rowBuilder =
+ rowBuilder.withFieldValue(
+ pkEntry.getKey().toLowerCase(),
+ stringToParsedValue(
+ internalRowSchema.getField(pkEntry.getKey().toLowerCase()).getType(),
+ pkEntry.getValue()));
}
- // TODO(pabloem): Understand why SpannerSchema has field names in lowercase...
- rowBuilder =
- rowBuilder.withFieldValue(
- pkEntry.getKey().toLowerCase(),
- stringToParsedValue(
- internalRowSchema.getField(pkEntry.getKey().toLowerCase()).getType(),
- pkEntry.getValue()));
+ receiver
+ .get(OUTPUT_TAG)
+ .outputWithTimestamp(
+ Row.withSchema(tableChangeRecordSchema)
+ .addValue(record.getModType().toString())
+ .addValue(record.getCommitTimestamp().toString())
+ .addValue(Long.parseLong(record.getRecordSequence()))
+ .addValue(rowBuilder.build())
+ .build(),
+ timestamp);
+ } catch (Exception e) {
+ errorsInBundle += 1;
Review Comment:
tyvm. I agree its low, but poorly handle exceptions are the causes of recent outages
--
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 #26970: Added dlq and error metrics to SpannerChangestreamsReadSchemaTransformProvider
Posted by "johnjcasey (via GitHub)" <gi...@apache.org>.
johnjcasey commented on code in PR #26970:
URL: https://github.com/apache/beam/pull/26970#discussion_r1218378000
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/SpannerChangestreamsReadSchemaTransformProvider.java:
##########
@@ -235,41 +264,58 @@ public void process(@DoFn.Element DataChangeRecord record, OutputReceiver<Row> r
.map(nonNullValues -> getGson().fromJson(nonNullValues, Map.class))
.orElseGet(Collections::emptyMap);
- for (Map.Entry<String, String> valueEntry : newValues.entrySet()) {
- if (valueEntry.getValue() == null) {
- continue;
+ try {
+ for (Map.Entry<String, String> valueEntry : newValues.entrySet()) {
+ if (valueEntry.getValue() == null) {
+ continue;
+ }
+ // TODO(pabloem): Understand why SpannerSchema has field names in lowercase...
+ rowBuilder =
+ rowBuilder.withFieldValue(
+ valueEntry.getKey().toLowerCase(),
+ stringToParsedValue(
+ internalRowSchema.getField(valueEntry.getKey().toLowerCase()).getType(),
+ valueEntry.getValue()));
}
- // TODO(pabloem): Understand why SpannerSchema has field names in lowercase...
- rowBuilder =
- rowBuilder.withFieldValue(
- valueEntry.getKey().toLowerCase(),
- stringToParsedValue(
- internalRowSchema.getField(valueEntry.getKey().toLowerCase()).getType(),
- valueEntry.getValue()));
- }
- for (Map.Entry<String, String> pkEntry : keyValues.entrySet()) {
- if (pkEntry.getValue() == null) {
- continue;
+ for (Map.Entry<String, String> pkEntry : keyValues.entrySet()) {
+ if (pkEntry.getValue() == null) {
+ continue;
+ }
+ // TODO(pabloem): Understand why SpannerSchema has field names in lowercase...
+ rowBuilder =
+ rowBuilder.withFieldValue(
+ pkEntry.getKey().toLowerCase(),
+ stringToParsedValue(
+ internalRowSchema.getField(pkEntry.getKey().toLowerCase()).getType(),
+ pkEntry.getValue()));
}
- // TODO(pabloem): Understand why SpannerSchema has field names in lowercase...
- rowBuilder =
- rowBuilder.withFieldValue(
- pkEntry.getKey().toLowerCase(),
- stringToParsedValue(
- internalRowSchema.getField(pkEntry.getKey().toLowerCase()).getType(),
- pkEntry.getValue()));
+ receiver
+ .get(OUTPUT_TAG)
+ .outputWithTimestamp(
+ Row.withSchema(tableChangeRecordSchema)
+ .addValue(record.getModType().toString())
+ .addValue(record.getCommitTimestamp().toString())
+ .addValue(Long.parseLong(record.getRecordSequence()))
+ .addValue(rowBuilder.build())
+ .build(),
+ timestamp);
+ } catch (Exception e) {
+ errorsInBundle += 1;
Review Comment:
Is it at all possible for there to be duplicate keys?
--
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 #26970: Added dlq and error metrics to SpannerChangestreamsReadSchemaTransformProvider
Posted by "Dippatel98 (via GitHub)" <gi...@apache.org>.
Dippatel98 commented on PR #26970:
URL: https://github.com/apache/beam/pull/26970#issuecomment-1572670313
Run Spotless 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 commented on a diff in pull request #26970: Added dlq and error metrics to SpannerChangestreamsReadSchemaTransformProvider
Posted by "johnjcasey (via GitHub)" <gi...@apache.org>.
johnjcasey commented on code in PR #26970:
URL: https://github.com/apache/beam/pull/26970#discussion_r1218377305
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/SpannerChangestreamsReadSchemaTransformProvider.java:
##########
@@ -193,15 +218,19 @@ public abstract static class Builder {
}
}
- private static final class DataChangeRecordToRow extends DoFn<DataChangeRecord, Row> {
+ public static final class DataChangeRecordToRow extends DoFn<DataChangeRecord, Row> {
Review Comment:
got it. Can you annotate it as @VisibleForTesting then?
--
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 #26970: Added dlq and error metrics to SpannerChangestreamsReadSchemaTransformProvider
Posted by "Dippatel98 (via GitHub)" <gi...@apache.org>.
Dippatel98 commented on code in PR #26970:
URL: https://github.com/apache/beam/pull/26970#discussion_r1218380589
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/SpannerChangestreamsReadSchemaTransformProvider.java:
##########
@@ -193,15 +218,19 @@ public abstract static class Builder {
}
}
- private static final class DataChangeRecordToRow extends DoFn<DataChangeRecord, Row> {
+ public static final class DataChangeRecordToRow extends DoFn<DataChangeRecord, Row> {
Review Comment:
My explanation was not clear, the dlq is tested in a way, just not the end to end workflow. I provide the DoFn with a correct and incorrect record. Then check if they are tagged properly.
--
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 #26970: Added dlq and error metrics to SpannerChangestreamsReadSchemaTransformProvider
Posted by "johnjcasey (via GitHub)" <gi...@apache.org>.
johnjcasey commented on code in PR #26970:
URL: https://github.com/apache/beam/pull/26970#discussion_r1218307979
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/SpannerChangestreamsReadSchemaTransformProvider.java:
##########
@@ -193,15 +218,19 @@ public abstract static class Builder {
}
}
- private static final class DataChangeRecordToRow extends DoFn<DataChangeRecord, Row> {
+ public static final class DataChangeRecordToRow extends DoFn<DataChangeRecord, Row> {
Review Comment:
why is this changed to public?
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/SpannerChangestreamsReadSchemaTransformProvider.java:
##########
@@ -235,41 +264,58 @@ public void process(@DoFn.Element DataChangeRecord record, OutputReceiver<Row> r
.map(nonNullValues -> getGson().fromJson(nonNullValues, Map.class))
.orElseGet(Collections::emptyMap);
- for (Map.Entry<String, String> valueEntry : newValues.entrySet()) {
- if (valueEntry.getValue() == null) {
- continue;
+ try {
+ for (Map.Entry<String, String> valueEntry : newValues.entrySet()) {
+ if (valueEntry.getValue() == null) {
+ continue;
+ }
+ // TODO(pabloem): Understand why SpannerSchema has field names in lowercase...
+ rowBuilder =
+ rowBuilder.withFieldValue(
+ valueEntry.getKey().toLowerCase(),
+ stringToParsedValue(
+ internalRowSchema.getField(valueEntry.getKey().toLowerCase()).getType(),
+ valueEntry.getValue()));
}
- // TODO(pabloem): Understand why SpannerSchema has field names in lowercase...
- rowBuilder =
- rowBuilder.withFieldValue(
- valueEntry.getKey().toLowerCase(),
- stringToParsedValue(
- internalRowSchema.getField(valueEntry.getKey().toLowerCase()).getType(),
- valueEntry.getValue()));
- }
- for (Map.Entry<String, String> pkEntry : keyValues.entrySet()) {
- if (pkEntry.getValue() == null) {
- continue;
+ for (Map.Entry<String, String> pkEntry : keyValues.entrySet()) {
+ if (pkEntry.getValue() == null) {
+ continue;
+ }
+ // TODO(pabloem): Understand why SpannerSchema has field names in lowercase...
+ rowBuilder =
+ rowBuilder.withFieldValue(
+ pkEntry.getKey().toLowerCase(),
+ stringToParsedValue(
+ internalRowSchema.getField(pkEntry.getKey().toLowerCase()).getType(),
+ pkEntry.getValue()));
}
- // TODO(pabloem): Understand why SpannerSchema has field names in lowercase...
- rowBuilder =
- rowBuilder.withFieldValue(
- pkEntry.getKey().toLowerCase(),
- stringToParsedValue(
- internalRowSchema.getField(pkEntry.getKey().toLowerCase()).getType(),
- pkEntry.getValue()));
+ receiver
+ .get(OUTPUT_TAG)
+ .outputWithTimestamp(
+ Row.withSchema(tableChangeRecordSchema)
+ .addValue(record.getModType().toString())
+ .addValue(record.getCommitTimestamp().toString())
+ .addValue(Long.parseLong(record.getRecordSequence()))
+ .addValue(rowBuilder.build())
+ .build(),
+ timestamp);
+ } catch (Exception e) {
+ errorsInBundle += 1;
Review Comment:
Is is it possible for anything here to have an exception? I'm especially looking at .getKeysJson and .getNewValuesJson. If so, we would fail to output to the error queue
--
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 #26970: Added dlq and error metrics to SpannerChangestreamsReadSchemaTransformProvider
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #26970:
URL: https://github.com/apache/beam/pull/26970#issuecomment-1572663242
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] Dippatel98 commented on pull request #26970: Added dlq and error metrics to SpannerChangestreamsReadSchemaTransformProvider
Posted by "Dippatel98 (via GitHub)" <gi...@apache.org>.
Dippatel98 commented on PR #26970:
URL: https://github.com/apache/beam/pull/26970#issuecomment-1576855294
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 #26970: Added dlq and error metrics to SpannerChangestreamsReadSchemaTransformProvider
Posted by "Dippatel98 (via GitHub)" <gi...@apache.org>.
Dippatel98 commented on code in PR #26970:
URL: https://github.com/apache/beam/pull/26970#discussion_r1218422152
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/SpannerChangestreamsReadSchemaTransformProvider.java:
##########
@@ -235,41 +264,58 @@ public void process(@DoFn.Element DataChangeRecord record, OutputReceiver<Row> r
.map(nonNullValues -> getGson().fromJson(nonNullValues, Map.class))
.orElseGet(Collections::emptyMap);
- for (Map.Entry<String, String> valueEntry : newValues.entrySet()) {
- if (valueEntry.getValue() == null) {
- continue;
+ try {
+ for (Map.Entry<String, String> valueEntry : newValues.entrySet()) {
+ if (valueEntry.getValue() == null) {
+ continue;
+ }
+ // TODO(pabloem): Understand why SpannerSchema has field names in lowercase...
+ rowBuilder =
+ rowBuilder.withFieldValue(
+ valueEntry.getKey().toLowerCase(),
+ stringToParsedValue(
+ internalRowSchema.getField(valueEntry.getKey().toLowerCase()).getType(),
+ valueEntry.getValue()));
}
- // TODO(pabloem): Understand why SpannerSchema has field names in lowercase...
- rowBuilder =
- rowBuilder.withFieldValue(
- valueEntry.getKey().toLowerCase(),
- stringToParsedValue(
- internalRowSchema.getField(valueEntry.getKey().toLowerCase()).getType(),
- valueEntry.getValue()));
- }
- for (Map.Entry<String, String> pkEntry : keyValues.entrySet()) {
- if (pkEntry.getValue() == null) {
- continue;
+ for (Map.Entry<String, String> pkEntry : keyValues.entrySet()) {
+ if (pkEntry.getValue() == null) {
+ continue;
+ }
+ // TODO(pabloem): Understand why SpannerSchema has field names in lowercase...
+ rowBuilder =
+ rowBuilder.withFieldValue(
+ pkEntry.getKey().toLowerCase(),
+ stringToParsedValue(
+ internalRowSchema.getField(pkEntry.getKey().toLowerCase()).getType(),
+ pkEntry.getValue()));
}
- // TODO(pabloem): Understand why SpannerSchema has field names in lowercase...
- rowBuilder =
- rowBuilder.withFieldValue(
- pkEntry.getKey().toLowerCase(),
- stringToParsedValue(
- internalRowSchema.getField(pkEntry.getKey().toLowerCase()).getType(),
- pkEntry.getValue()));
+ receiver
+ .get(OUTPUT_TAG)
+ .outputWithTimestamp(
+ Row.withSchema(tableChangeRecordSchema)
+ .addValue(record.getModType().toString())
+ .addValue(record.getCommitTimestamp().toString())
+ .addValue(Long.parseLong(record.getRecordSequence()))
+ .addValue(rowBuilder.build())
+ .build(),
+ timestamp);
+ } catch (Exception e) {
+ errorsInBundle += 1;
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 #26970: Added dlq and error metrics to SpannerChangestreamsReadSchemaTransformProvider
Posted by "Dippatel98 (via GitHub)" <gi...@apache.org>.
Dippatel98 commented on code in PR #26970:
URL: https://github.com/apache/beam/pull/26970#discussion_r1218422744
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/SpannerChangestreamsReadSchemaTransformProvider.java:
##########
@@ -193,15 +218,19 @@ public abstract static class Builder {
}
}
- private static final class DataChangeRecordToRow extends DoFn<DataChangeRecord, Row> {
+ public static final class DataChangeRecordToRow extends DoFn<DataChangeRecord, Row> {
Review Comment:
Done (I have added @VisibleForTesting).
--
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