lostluck commented on code in PR #17558:
URL: https://github.com/apache/beam/pull/17558#discussion_r866277511
##########
sdks/go/pkg/beam/core/runtime/exec/pardo.go:
##########
@@ -174,17 +174,22 @@ func (n *ParDo) processSingleWindow(mainIn *MainInput)
(sdf.ProcessContinuation,
if err != nil {
return nil, n.fail(err)
}
- if mainIn.RTracker != nil && !mainIn.RTracker.IsDone() {
- return nil, rtErrHelper(mainIn.RTracker.GetError())
- }
// Forward direct output, if any. It is always a main output.
if val != nil {
+ // Check for incomplete processing of a restriction without a
checkpoint
+ if mainIn.RTracker != nil && !mainIn.RTracker.IsDone() &&
val.Continuation == nil {
+ return nil, rtErrHelper(mainIn.RTracker.GetError())
+ }
// We do not forward a ProcessContinuation on its own
if val.Elm == nil {
return val.Continuation, nil
}
return val.Continuation, n.Out[0].ProcessElement(n.ctx, val)
+ } else {
+ if mainIn.RTracker != nil && !mainIn.RTracker.IsDone() {
+ return nil, rtErrHelper(mainIn.RTracker.GetError())
+ }
}
Review Comment:
Style nit: If the previous clause ends with a return, drop the else.
--
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]
