jrmccluskey commented on code in PR #17716:
URL: https://github.com/apache/beam/pull/17716#discussion_r877544245
##########
sdks/go/pkg/beam/core/runtime/exec/datasource.go:
##########
@@ -366,13 +389,30 @@ 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(rs) == 0 {
+ return SplitResult{}, -1 * time.Minute, false, nil
+ }
if len(ps) != 0 {
- return SplitResult{}, -1 * time.Minute, false,
fmt.Errorf("failed to checkpoint: got %v primary roots, want none", ps)
+ // Expected structure of the root FullValue is KV<KV<Elm,
KV<BoundedRTracker, watermarkEstimatorState>>, Size, (Timestamp?, Windows?)>
Review Comment:
Fixed
--
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]
