Abacn commented on code in PR #39572:
URL: https://github.com/apache/beam/pull/39572#discussion_r3814860485
##########
sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go:
##########
@@ -1874,22 +1898,51 @@ keysPerBundle:
break
}
}
+ if len(toProcessForKey) > 0 {
+ newKeys.insert(k)
+ // Track the min-timestamp for later watermark
handling. Elements pop
+ // in timestamp order, so the first selected one is the
earliest.
+ if ts := toProcessForKey[0].timestamp; ts < minTs {
+ minTs = ts
+ }
+ }
toProcess = append(toProcess, toProcessForKey...)
if dnt.elements.Len() == 0 {
delete(ss.pendingByKeys, k)
}
- if OneKeyPerBundle {
+ // A key that yielded nothing, such as one headed by a timer
above the
+ // watermark, must not consume the single key slot, or the
bundle is empty
+ // and the stage keeps rescheduling on it.
+ if OneKeyPerBundle && len(toProcessForKey) > 0 {
break keysPerBundle
}
}
- // If we're out of data, and timers were not cleared then the watermark
is accurate.
- stillSchedulable := !(len(ss.pendingByKeys) == 0 && !timerCleared)
+ // Reschedule only when a later bundle could build something, or a
cleared
+ // timer may have held back the minimum pending timestamp.
+ stillSchedulable := timerCleared || ss.hasBuildableDataLocked(watermark)
return toProcess, minTs, newKeys, holdsInBundle, nil, stillSchedulable, 0
}
+// hasBuildableDataLocked reports whether a key that isn't in progress heads
its
+// heap with data, or with a timer the watermark has reached. Callers hold
ss.mu.
+func (ss *stageState) hasBuildableDataLocked(watermark mtime.Time) bool {
+ for k, dnt := range ss.pendingByKeys {
Review Comment:
Here it iterates over pendingByKeys again. Would there be any performance
concern?
--
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]
