jrmccluskey commented on code in PR #40975:
URL: https://github.com/apache/beam/pull/40975#discussion_r4472093135
##########
sdks/go/pkg/beam/runners/prism/internal/coders_test.go:
##########
@@ -348,7 +348,24 @@ func Test_pullDecoder(t *testing.T) {
},
},
},
- []byte{4, 0, 1, 2, 3},
+ []byte{0, 0, 0, 4, 0, 1, 2, 3},
Review Comment:
Why did this test case change?
##########
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:
Is there a reason these got moved later in the function?
##########
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:
I'm not sure I follow the purpose of the second count increment in this block
##########
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:
Is there a reason this block was moved earlier in the function?
--
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]
