ahmedabu98 commented on code in PR #40975:
URL: https://github.com/apache/beam/pull/40975#discussion_r4473063445
##########
sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go:
##########
@@ -2589,7 +2596,21 @@ func (ss *stageState)
createOnWindowExpirationBundles(newOut mtime.Time, em *Ele
elmBytes: nil,
})
}
- em.addPending(len(toProcess))
+ // Remove the key accounting, or continue tracking which keys
still need clearing.
+ if len(busyKeys) == 0 {
+ delete(ss.keysToExpireByWindow, win)
+ } else {
+ ss.keysToExpireByWindow[win] = busyKeys
+ }
+ if len(toProcess) == 0 {
+ // Every key is busy, so the window expires once one is
done.
+ continue
+ }
+ // Hold off on garbage collecting data for these windows while
these
+ // are in progress.
+ ss.inProgressExpiredWindows[win] += 1
+ rb := RunBundle{StageID: ss.ID, BundleID: "owe-" +
em.nextBundID(), Watermark: wm}
Review Comment:
`rb := RunBundle{...}` didn’t need to move. It’s back in its original spot
`ss.inProgressExpiredWindows[win] += 1` has to stay after the decrement, so
that the counter has a chance to decrement when the window's bundle persists
##########
sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go:
##########
@@ -2589,7 +2596,21 @@ func (ss *stageState)
createOnWindowExpirationBundles(newOut mtime.Time, em *Ele
elmBytes: nil,
})
}
- em.addPending(len(toProcess))
+ // Remove the key accounting, or continue tracking which keys
still need clearing.
+ if len(busyKeys) == 0 {
+ delete(ss.keysToExpireByWindow, win)
+ } else {
+ ss.keysToExpireByWindow[win] = busyKeys
+ }
Review Comment:
No good reason, moved it back to where it was
##########
sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go:
##########
@@ -1621,13 +1621,19 @@ func (*statefulStageKind) addPending(ss *stageState, em
*ElementManager, newPend
timers: map[timerKey]timerTimes{},
}
ss.pendingByKeys[string(e.keyBytes)] = dnt
- if ss.keysToExpireByWindow != nil {
- w, ok := ss.keysToExpireByWindow[e.window]
- if !ok {
- w = make(set[string])
- ss.keysToExpireByWindow[e.window] = w
- }
+ }
+ // The key owes an OnWindowExpiration callback for each window
it uses. The callback counts
+ // as pending until its bundle persists, so the job can't end
before the callback runs.
+ // A cleared timer, with a sequence of -1, doesn't use its
window.
+ if ss.keysToExpireByWindow != nil && e.sequence >= 0 {
+ w, ok := ss.keysToExpireByWindow[e.window]
+ if !ok {
+ w = make(set[string])
+ ss.keysToExpireByWindow[e.window] = w
+ }
+ if !w.present(string(e.keyBytes)) {
w.insert(string(e.keyBytes))
+ count++
Review Comment:
The first count++ counts the element.
The second counts the OnWindowExpiration callback that the key owes for that
window.
Added a comment to clarify
--
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]
