Abacn commented on code in PR #39572:
URL: https://github.com/apache/beam/pull/39572#discussion_r3822822050
##########
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:
These tests do use prism runner:
https://github.com/apache/beam/blob/93f3e051c3ef47238711eb4f313d7e01477c42c3/sdks/python/test-suites/direct/common.gradle#L454
My comment was about this does fix the flakiness seen in the JmsIO test to
some extent. Since this PR closes #39446, can we apply
https://github.com/apache/beam/commit/f274b1b35a14123afb17547cae9e2e5683864ab2
here (remove 5s delay) and trigger PostCommit Python Xlang Messaging Direct?
Thanks!
--
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]
