Repository navigation
Tag the TFRecord write error output with the schema it actually emits - #39759
Conversation
ErrorFn is constructed with errorSchema = ErrorHandling.errorSchema(
inputSchema) and emits ErrorHandling.errorRecord(errorSchema, ..), so
every row on ERROR_TAG carries that schema. The collection was then
tagged with ErrorHandling.errorSchema(errorSchema) -- the wrapper
applied twice -- declaring {failed_row: {failed_row: .., error_message},
error_message} which no element it produces can match.
The sibling in the same package gets it right five lines after the same
ErrorFn construction: TFRecordReadSchemaTransformProvider:140 does
setRowSchema(errorSchema). JavaFilter, JavaMapToFields, PubsubRowToMessage
and BigQueryStorageWriteApi all do the same.
One test, and it needs no runner -- the schema is fixed when the graph is
built. Restoring the second wrap fails it and nothing else: 8 tests, 1
failure.
KafkaWriteSchemaTransformProvider:301 has the identical double wrap. It
is a different module, so it is a separate change rather than folded in
here.
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #39759 +/- ##
============================================
+ Coverage 56.57% 57.13% +0.56%
+ Complexity 3999 3641 -358
============================================
Files 1329 1191 -138
Lines 185231 192731 +7500
Branches 2729 3816 +1087
============================================
+ Hits 104796 110125 +5329
- Misses 77466 79038 +1572
- Partials 2969 3568 +599
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
|
Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment |
|
Run Java_IOs_Direct PreCommit |
|
Assigning reviewers: R: @kennknowles for label java. Note: If you would like to opt out of this review, comment Available commands:
The PR bot will only process comments in the main thread (not review comments). |
ErrorCounterFn emits ErrorHandling.errorRecord(errorSchema, ...), where errorSchema is already ErrorHandling.errorSchema(inputSchema). The error PCollection was tagged with the wrapper applied a second time, declaring a shape no emitted element can match. Same defect as apache#39759 in TFRecordWriteSchemaTransformProvider. The existing tests apply ErrorCounterFn directly and tag the output themselves, so none of them reach the transform's expand(). Signed-off-by: Zihan Dai <dzh1436286758@gmail.com>
…39760) * Tag the Kafka write error output with the schema it actually emits ErrorCounterFn emits ErrorHandling.errorRecord(errorSchema, ...), where errorSchema is already ErrorHandling.errorSchema(inputSchema). The error PCollection was tagged with the wrapper applied a second time, declaring a shape no emitted element can match. Same defect as #39759 in TFRecordWriteSchemaTransformProvider. The existing tests apply ErrorCounterFn directly and tag the output themselves, so none of them reach the transform's expand(). Signed-off-by: Zihan Dai <dzh1436286758@gmail.com> * Add CHANGES.md entry for the Kafka write error schema fix Signed-off-by: Zihan Dai <dzh1436286758@gmail.com> --------- Signed-off-by: Zihan Dai <dzh1436286758@gmail.com> Co-authored-by: Zihan Dai <dzh1436286758@gmail.com>
Please add a meaningful description for your change here
ErrorFnis constructed witherrorSchema = ErrorHandling.errorSchema(inputSchema)and emitsErrorHandling.errorRecord(errorSchema, row, e), so every row onERROR_TAGcarries that schema. The collection was then tagged with the wrapper applied a second time:errorSchema(x)is{failed_row: Row(x), error_message: STRING}, so the declared shape becomes{failed_row: {failed_row: …, error_message: …}, error_message: …}— which no element the transform produces can match.The sibling in the same package gets it right, five lines after an identical
ErrorFnconstruction:JavaFilterTransformProvider,JavaMapToFieldsTransformProvider,PubsubRowToMessage,PubsubWriteSchemaTransformProviderandBigQueryStorageWriteApiSchemaTransformProviderall do the same.Test
One, and it needs no runner — the schema is fixed when the graph is built, so it runs in the ordinary
:sdks:java:core:testtask rather than underNeedsRunner.Restoring the second wrap fails it and nothing else:
spotlessJavaCheck,checkstyleMainandcheckstyleTeston:sdks:java:coreare clean.Not in this PR
KafkaWriteSchemaTransformProvider.java:301has the identical double wrap. It is in a different module with its own CI, so I have kept it separate rather than folding it in.Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
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, commentfixes #<ISSUE NUMBER>instead.CHANGES.mdwith noteworthy changes.See the Contributor Guide for more tips on how to make review process smoother.
To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md
GitHub Actions Tests Status (on master branch)
See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.