[BEAM-14499] Fix TestStream integration tests #17733
Conversation
|
Can one of the admins verify this patch? |
4 similar comments
|
Can one of the admins verify this patch? |
|
Can one of the admins verify this patch? |
|
Can one of the admins verify this patch? |
|
Can one of the admins verify this patch? |
Codecov Report
@@ Coverage Diff @@
## master #17733 +/- ##
==========================================
- Coverage 73.98% 73.98% -0.01%
==========================================
Files 696 696
Lines 91851 91871 +20
==========================================
+ Hits 67958 67968 +10
- Misses 22644 22653 +9
- Partials 1249 1250 +1
Flags with carried forward coverage won't be shown. Click here to find out more.
Continue to review full report at Codecov.
|
|
Run Go Flink ValidatesRunner |
|
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control |
| // cannot be globally windowed. | ||
| func windowedEquals(s beam.Scope, wfn *window.Fn, actual, expected beam.PCollection) beam.PCollection { | ||
| unexpected, correct, missing := WindowedDiff(s, wfn, actual, expected) | ||
| beam.ParDo0(s, failIfBadEntries, beam.WindowInto(s, wfn, beam.Impulse(s)), beam.SideInput{Input: unexpected}, beam.SideInput{Input: correct}, beam.SideInput{Input: missing}) |
There was a problem hiding this comment.
Right now I don't think failIfBadEntries prints out windowing information - it probably should since that can now make a difference.
There was a problem hiding this comment.
I don't think we can be window function aware at that level.
There was a problem hiding this comment.
Yeah, I meant information about the actual window, not the windowing function. Basically, if you get different results because of windowing it would be nice if that was obvious from the error
There was a problem hiding this comment.
Okay that was more doable. We have a windowing call-out now.
| t.Errorf("Pipeline failed: %v", err) | ||
| } | ||
| } | ||
|
|
There was a problem hiding this comment.
Could we add a test or two for windowing equality? Ideally at least 1 success and 1+ failures
| @@ -44,6 +45,21 @@ func Diff(s beam.Scope, a, b beam.PCollection) (left, both, right beam.PCollecti | |||
| return beam.ParDo3(s, &diffFn{Type: beam.EncodedType{T: t.Type()}}, imp, beam.SideInput{Input: a}, beam.SideInput{Input: b}) | |||
| } | |||
There was a problem hiding this comment.
I want to call out that we no longer actually use this ourselves, so its theoretically a candidate for deprecation/removal. With that said, I think this is both public and usable enough that people might have dependencies on it, so I don't really think that's a good idea. Mostly just had the thought and wanted to see how others feel about it.
|
Current Go SDK testing utilities and primitives (namely Impulse) don't play nicely with the windowing and streaming set-up, so this approach will not work. Closing this PR and stepping back the breaking case to a warning in #17735 |
Fixes TestStream tests by introducing a new WindowedEqualsList() function that passes a window function down to wrap necessary components of the Equals() transforms in. This avoids pipeline construction-time failures in TestStream test cases.
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
R: @username).[BEAM-XXX] Fixes bug in ApproximateQuantiles, where you replaceBEAM-XXXwith the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.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.