jrmccluskey commented on code in PR #17716:
URL: https://github.com/apache/beam/pull/17716#discussion_r877204263
##########
sdks/go/pkg/beam/core/runtime/exec/datasource.go:
##########
@@ -366,13 +378,25 @@ func (n *DataSource) Checkpoint() (SplitResult,
time.Duration, bool, error) {
ow := su.GetOutputWatermark()
- // Always split at fraction 0.0, should have no primaries left.
+ // Always split at fraction 0.0. All remaining work should be returned
as a residual, as anything left in the primaries
+ // will not be rescheduled and could represent data loss. We expect nil
primaries but will also ignore any restrictions
+ // that are bounded and of size 0 as they represent no remaining work.
ps, rs, err := su.Split(0.0)
if err != nil {
return SplitResult{}, -1 * time.Minute, false, err
}
if len(ps) != 0 {
- return SplitResult{}, -1 * time.Minute, false,
fmt.Errorf("failed to checkpoint: got %v primary roots, want none", ps)
+ for _, root := range ps {
+ tracker, size, ok := getRTrackerFromRoot(root)
+ // If type assertion didn't return a BoundableRTracker,
we move on.
+ if !ok {
+ continue
+ }
+ if !tracker.IsBounded() || size > 0.00001 {
Review Comment:
I believe it _should_ be if we're here, but if the split code is implemented
incorrectly for a given tracker it may not be. Better to check and make sure we
aren't losing anything in the primary than assume that everything is exactly as
we expect.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
