From 3b1db11e4437813f3ed2bcbd567e9134e71b1c58 Mon Sep 17 00:00:00 2001 From: Shunping Huang Date: Mon, 29 Sep 2025 23:46:14 -0400 Subject: [PATCH 01/13] Handle after-processing-time trigger with processing-time timer. --- .../prism/internal/engine/elementmanager.go | 174 ++++++++++++++++-- .../runners/prism/internal/engine/strategy.go | 43 +++++ .../prism/internal/jobservices/management.go | 2 +- 3 files changed, 204 insertions(+), 15 deletions(-) diff --git a/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go b/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go index 6af030f36228..da48d9399296 100644 --- a/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go +++ b/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go @@ -1215,7 +1215,9 @@ type stageKind interface { // buildEventTimeBundle handles building bundles for the stage per it's kind. buildEventTimeBundle(ss *stageState, watermark mtime.Time) (toProcess elementHeap, minTs mtime.Time, newKeys set[string], holdsInBundle map[mtime.Time]int, panesInBundle []bundlePane, schedulable bool, pendingAdjustment int) - + // buildProcessingTimeBundle handles building processing-time bundles for the stage per it's kind. + buildProcessingTimeBundle(ss *stageState, em *ElementManager, emNow mtime.Time) (toProcess elementHeap, minTs mtime.Time, newKeys set[string], + holdsInBundle map[mtime.Time]int, schedulable bool) // getPaneOrDefault based on the stage state, element metadata, and bundle id. getPaneOrDefault(ss *stageState, defaultPane typex.PaneInfo, w typex.Window, keyBytes []byte, bundID string) typex.PaneInfo } @@ -1327,17 +1329,54 @@ func (ss *stageState) injectTriggeredBundlesIfReady(em *ElementManager, window t ready := ss.strat.IsTriggerReady(triggerInput{ newElementCount: 1, endOfWindowReached: endOfWindowReached, + emNow: em.ProcessingTimeNow(), }, &state) if ready { state.Pane = computeNextTriggeredPane(state.Pane, endOfWindowReached) + } else { + if pts := ss.strat.GetAfterProcessingTimeTriggers(); pts != nil { + for _, t := range pts { + ts := (&state).getTriggerState(t) + if ts.extra == nil || t.shouldFire((&state)) { + // Skipping inserting a processing time timer if the firing time + // is not set or it already should fire. + // When the after processing time triggers should fire, there are + // two scenarios: + // (1) the entire trigger of this window is ready to fire. In this + // case, `ready` should be true and we won't reach here. + // (2) we are still waiting for other triggers (subtriggers) to + // fire (e.g. AfterAll). + continue + } + firingTime := ts.extra.(afterProcessingTimeState).firingTime + notYetHolds := map[mtime.Time]int{} + timer := element{ + window: window, + timestamp: firingTime, + holdTimestamp: window.MaxTimestamp(), + pane: typex.NoFiringPane(), + transform: ss.ID, // Use stage id to fake transform id + family: "AfterProcessingTime", + tag: "", + sequence: 1, + elmBytes: nil, + keyBytes: []byte(key), + } + // TODO: how to deal with watermark holds for this implicit processing time timer + // ss.watermarkHolds.Add(timer.holdTimestamp, 1) + ss.processingTimeTimers.Persist(firingTime, timer, notYetHolds) + em.processTimeEvents.Schedule(firingTime, ss.ID) + em.wakeUpAt(firingTime) + } + } } // Store the state as triggers may have changed it. ss.state[LinkID{}][window][key] = state // If we're ready, it's time to fire! if ready { - count += ss.buildTriggeredBundle(em, key, window) + count += ss.startTriggeredBundle(em, key, window) } return count } @@ -1524,16 +1563,11 @@ func (ss *stageState) savePanes(bundID string, panesInBundle []bundlePane) { } } -// buildTriggeredBundle must be called with the stage.mu lock held. -// When in discarding mode, returns 0. -// When in accumulating mode, returns the number of fired elements to maintain a correct pending count. -func (ss *stageState) buildTriggeredBundle(em *ElementManager, key string, win typex.Window) int { +func (ss *stageState) buildTriggeredBundle(em *ElementManager, key string, win typex.Window) ([]element, int) { var toProcess []element dnt := ss.pendingByKeys[key] var notYet []element - rb := RunBundle{StageID: ss.ID, BundleID: "agg-" + em.nextBundID(), Watermark: ss.input} - // Look at all elements for this key, and only for this window. for dnt.elements.Len() > 0 { e := heap.Pop(&dnt.elements).(element) @@ -1564,6 +1598,18 @@ func (ss *stageState) buildTriggeredBundle(em *ElementManager, key string, win t heap.Init(&dnt.elements) } + return toProcess, accumulationDiff +} + +// startTriggeredBundle must be called with the stage.mu lock held. +// When in discarding mode, returns 0. +// When in accumulating mode, returns the number of fired elements to maintain a correct pending count. +func (ss *stageState) startTriggeredBundle(em *ElementManager, key string, win typex.Window) int { + toProcess, accumulationDiff := ss.buildTriggeredBundle(em, key, win) + if len(toProcess) == 0 { + return accumulationDiff + } + if ss.inprogressKeys == nil { ss.inprogressKeys = set[string]{} } @@ -1575,6 +1621,7 @@ func (ss *stageState) buildTriggeredBundle(em *ElementManager, key string, win t }, } + rb := RunBundle{StageID: ss.ID, BundleID: "agg-" + em.nextBundID(), Watermark: ss.input} ss.makeInProgressBundle( func() string { return rb.BundleID }, toProcess, @@ -1927,6 +1974,18 @@ func (ss *stageState) startProcessingTimeBundle(em *ElementManager, emNow mtime. ss.mu.Lock() defer ss.mu.Unlock() + toProcess, minTs, newKeys, holdsInBundle, stillSchedulable := ss.kind.buildProcessingTimeBundle(ss, em, emNow) + + if len(toProcess) == 0 { + // If we have nothing + return "", false, stillSchedulable + } + bundID := ss.makeInProgressBundle(genBundID, toProcess, minTs, newKeys, holdsInBundle, nil) + slog.Debug("started a processing time bundle", "stageID", ss.ID, "bundleID", bundID, "size", len(toProcess), "emNow", emNow) + return bundID, true, stillSchedulable +} + +func (*statefulStageKind) buildProcessingTimeBundle(ss *stageState, em *ElementManager, emNow mtime.Time) (elementHeap, mtime.Time, set[string], map[mtime.Time]int, bool) { // TODO: Determine if it's possible and a good idea to treat all EventTime processing as a MinTime // Special Case for ProcessingTime handling. // Eg. Always queue EventTime elements at minTime. @@ -1986,19 +2045,92 @@ func (ss *stageState) startProcessingTimeBundle(em *ElementManager, emNow mtime. for _, v := range notYet { ss.processingTimeTimers.Persist(v.firing, v.timer, notYetHolds) em.processTimeEvents.Schedule(v.firing, ss.ID) + em.wakeUpAt(v.firing) } // Add a refresh if there are still processing time events to process. stillSchedulable := (nextTime < emNow && nextTime != mtime.MaxTimestamp || len(notYet) > 0) - if len(toProcess) == 0 { - // If we have nothing - return "", false, stillSchedulable + return toProcess, minTs, newKeys, holdsInBundle, stillSchedulable +} + +func (*ordinaryStageKind) buildProcessingTimeBundle(ss *stageState, em *ElementManager, emNow mtime.Time) (elementHeap, mtime.Time, set[string], map[mtime.Time]int, bool) { + slog.Error("ordinary stages can't have processing time elements") + return nil, mtime.MinTimestamp, nil, nil, false +} + +func (*aggregateStageKind) buildProcessingTimeBundle(ss *stageState, em *ElementManager, emNow mtime.Time) (elementHeap, mtime.Time, set[string], map[mtime.Time]int, bool) { + var toProcess []element + minTs := mtime.MaxTimestamp + holdsInBundle := map[mtime.Time]int{} + + var notYet []fireElement + + nextTime := ss.processingTimeTimers.Peek() + keyCounts := map[string]int{} + newKeys := set[string]{} + + for nextTime <= emNow { + elems := ss.processingTimeTimers.FireAt(nextTime) + for _, e := range elems { + // Check if we're already executing this timer's key. + if ss.inprogressKeys.present(string(e.keyBytes)) { + notYet = append(notYet, fireElement{firing: nextTime, timer: e}) + continue + } + + // If we are set to have OneKeyPerBundle, and we already have a key for this bundle, we process it later. + if len(keyCounts) > 0 && OneKeyPerBundle { + notYet = append(notYet, fireElement{firing: nextTime, timer: e}) + continue + } + // If we are set to have OneElementPerKey, and we already have an element for this key we set this to process later. + if v := keyCounts[string(e.keyBytes)]; v > 0 && OneElementPerKey { + notYet = append(notYet, fireElement{firing: nextTime, timer: e}) + continue + } + keyCounts[string(e.keyBytes)]++ + newKeys.insert(string(e.keyBytes)) + if e.timestamp < minTs { + minTs = e.timestamp + } + // TODO: how to deal with watermark holds for this implicit processing time timer + // holdsInBundle[e.holdTimestamp]++ + + state := ss.state[LinkID{}][e.window][string(e.keyBytes)] + endOfWindowReached := e.window.MaxTimestamp() < ss.input + ready := ss.strat.IsTriggerReady(triggerInput{ + newElementCount: 0, + endOfWindowReached: endOfWindowReached, + emNow: emNow, + }, &state) + + if ready { + // We're going to process this trigger! + elems, _ := ss.buildTriggeredBundle(em, string(e.keyBytes), e.window) + toProcess = append(toProcess, elems...) + } + } + + nextTime = ss.processingTimeTimers.Peek() + if nextTime == mtime.MaxTimestamp { + // Escape the loop if there are no more events. + break + } } - bundID := ss.makeInProgressBundle(genBundID, toProcess, minTs, newKeys, holdsInBundle, nil) - slog.Debug("started a processing time bundle", "stageID", ss.ID, "bundleID", bundID, "size", len(toProcess), "emNow", emNow) - return bundID, true, stillSchedulable + // Reschedule unfired timers. + notYetHolds := map[mtime.Time]int{} + for _, v := range notYet { + ss.processingTimeTimers.Persist(v.firing, v.timer, notYetHolds) + em.processTimeEvents.Schedule(v.firing, ss.ID) + em.wakeUpAt(v.firing) + } + + // Add a refresh if there are still processing time events to process. + stillSchedulable := (nextTime < emNow && nextTime != mtime.MaxTimestamp || len(notYet) > 0) + + return toProcess, minTs, newKeys, holdsInBundle, stillSchedulable } // makeInProgressBundle is common code to store a set of elements as a bundle in progress. @@ -2329,3 +2461,17 @@ func (em *ElementManager) ProcessingTimeNow() (ret mtime.Time) { func rebaseProcessingTime(localNow, scheduled mtime.Time) mtime.Time { return localNow + (scheduled - mtime.Now()) } + +// wakeUpAt schedules a wakeup signal for the bundle processing loop. +// This is used for processing time timers to ensure the loop re-evaluates +// stages when a processing time timer is expected to fire. +func (em *ElementManager) wakeUpAt(t mtime.Time) { + if em.testStreamHandler == nil && em.config.EnableRTC { + // only create this goroutine if we have real-time clock enabled and the pipeline does not have TestStream. + go func(fireAt time.Time) { + time.AfterFunc(time.Until(fireAt), func() { + em.refreshCond.Broadcast() + }) + }(t.ToTime()) + } +} diff --git a/sdks/go/pkg/beam/runners/prism/internal/engine/strategy.go b/sdks/go/pkg/beam/runners/prism/internal/engine/strategy.go index 044b9806c1b1..2aef5fcf332f 100644 --- a/sdks/go/pkg/beam/runners/prism/internal/engine/strategy.go +++ b/sdks/go/pkg/beam/runners/prism/internal/engine/strategy.go @@ -73,6 +73,49 @@ func (ws WinStrat) IsNeverTrigger() bool { return ok } +func getAfterProcessingTimeTriggers(t Trigger) []*TriggerAfterProcessingTime { + if t == nil { + return nil + } + var triggers []*TriggerAfterProcessingTime + switch at := t.(type) { + case *TriggerAfterProcessingTime: + return []*TriggerAfterProcessingTime{at} + case *TriggerAfterAll: + for _, st := range at.SubTriggers { + triggers = append(triggers, getAfterProcessingTimeTriggers(st)...) + } + return triggers + case *TriggerAfterAny: + for _, st := range at.SubTriggers { + triggers = append(triggers, getAfterProcessingTimeTriggers(st)...) + } + return triggers + case *TriggerAfterEach: + for _, st := range at.SubTriggers { + triggers = append(triggers, getAfterProcessingTimeTriggers(st)...) + } + return triggers + case *TriggerAfterEndOfWindow: + triggers = append(triggers, getAfterProcessingTimeTriggers(at.Early)...) + triggers = append(triggers, getAfterProcessingTimeTriggers(at.Late)...) + return triggers + case *TriggerOrFinally: + triggers = append(triggers, getAfterProcessingTimeTriggers(at.Main)...) + triggers = append(triggers, getAfterProcessingTimeTriggers(at.Finally)...) + return triggers + case *TriggerRepeatedly: + return getAfterProcessingTimeTriggers(at.Repeated) + default: + return nil + } +} + +// GetAfterProcessingTimeTriggers returns all AfterProcessingTime triggers within the trigger. +func (ws WinStrat) GetAfterProcessingTimeTriggers() []*TriggerAfterProcessingTime { + return getAfterProcessingTimeTriggers(ws.Trigger) +} + func (ws WinStrat) String() string { return fmt.Sprintf("WinStrat[AllowedLateness:%v Trigger:%v]", ws.AllowedLateness, ws.Trigger) } diff --git a/sdks/go/pkg/beam/runners/prism/internal/jobservices/management.go b/sdks/go/pkg/beam/runners/prism/internal/jobservices/management.go index f00838152111..12c3c42c2e92 100644 --- a/sdks/go/pkg/beam/runners/prism/internal/jobservices/management.go +++ b/sdks/go/pkg/beam/runners/prism/internal/jobservices/management.go @@ -316,7 +316,7 @@ func (s *Server) Prepare(ctx context.Context, req *jobpb.PrepareJobRequest) (_ * func hasUnsupportedTriggers(tpb *pipepb.Trigger) bool { unsupported := false switch at := tpb.GetTrigger().(type) { - case *pipepb.Trigger_AfterProcessingTime_, *pipepb.Trigger_AfterSynchronizedProcessingTime_: + case *pipepb.Trigger_AfterSynchronizedProcessingTime_: return true case *pipepb.Trigger_AfterAll_: for _, st := range at.AfterAll.GetSubtriggers() { From b3cd9536ba2b9e29dbc94f6b5dc9ef1cefff7608 Mon Sep 17 00:00:00 2001 From: Shunping Huang Date: Tue, 30 Sep 2025 12:49:37 -0400 Subject: [PATCH 02/13] Consolidate the buildProcessingTimeBundle for stateful and aggregate kinds into one code path. --- .../prism/internal/engine/elementmanager.go | 110 ++++++------------ 1 file changed, 33 insertions(+), 77 deletions(-) diff --git a/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go b/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go index da48d9399296..4366071d8302 100644 --- a/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go +++ b/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go @@ -1985,7 +1985,8 @@ func (ss *stageState) startProcessingTimeBundle(em *ElementManager, emNow mtime. return bundID, true, stillSchedulable } -func (*statefulStageKind) buildProcessingTimeBundle(ss *stageState, em *ElementManager, emNow mtime.Time) (elementHeap, mtime.Time, set[string], map[mtime.Time]int, bool) { +func handleProcessingTimeTimer(ss *stageState, em *ElementManager, emNow mtime.Time, + processTimerFn func(e element, toProcess []element, holdsInBundle map[mtime.Time]int) []element) (elementHeap, mtime.Time, set[string], map[mtime.Time]int, bool) { // TODO: Determine if it's possible and a good idea to treat all EventTime processing as a MinTime // Special Case for ProcessingTime handling. // Eg. Always queue EventTime elements at minTime. @@ -2027,10 +2028,8 @@ func (*statefulStageKind) buildProcessingTimeBundle(ss *stageState, em *ElementM if e.timestamp < minTs { minTs = e.timestamp } - holdsInBundle[e.holdTimestamp]++ - // We're going to process this timer! - toProcess = append(toProcess, e) + toProcess = processTimerFn(e, toProcess, holdsInBundle) } nextTime = ss.processingTimeTimers.Peek() @@ -2054,83 +2053,40 @@ func (*statefulStageKind) buildProcessingTimeBundle(ss *stageState, em *ElementM return toProcess, minTs, newKeys, holdsInBundle, stillSchedulable } -func (*ordinaryStageKind) buildProcessingTimeBundle(ss *stageState, em *ElementManager, emNow mtime.Time) (elementHeap, mtime.Time, set[string], map[mtime.Time]int, bool) { - slog.Error("ordinary stages can't have processing time elements") - return nil, mtime.MinTimestamp, nil, nil, false +func (*statefulStageKind) buildProcessingTimeBundle(ss *stageState, em *ElementManager, emNow mtime.Time) (elementHeap, mtime.Time, set[string], map[mtime.Time]int, bool) { + return handleProcessingTimeTimer(ss, em, emNow, func(e element, toProcess []element, holdsInBundle map[mtime.Time]int) []element { + holdsInBundle[e.holdTimestamp]++ + // We're going to process this timer! + toProcess = append(toProcess, e) + return toProcess + }) } func (*aggregateStageKind) buildProcessingTimeBundle(ss *stageState, em *ElementManager, emNow mtime.Time) (elementHeap, mtime.Time, set[string], map[mtime.Time]int, bool) { - var toProcess []element - minTs := mtime.MaxTimestamp - holdsInBundle := map[mtime.Time]int{} - - var notYet []fireElement - - nextTime := ss.processingTimeTimers.Peek() - keyCounts := map[string]int{} - newKeys := set[string]{} - - for nextTime <= emNow { - elems := ss.processingTimeTimers.FireAt(nextTime) - for _, e := range elems { - // Check if we're already executing this timer's key. - if ss.inprogressKeys.present(string(e.keyBytes)) { - notYet = append(notYet, fireElement{firing: nextTime, timer: e}) - continue - } - - // If we are set to have OneKeyPerBundle, and we already have a key for this bundle, we process it later. - if len(keyCounts) > 0 && OneKeyPerBundle { - notYet = append(notYet, fireElement{firing: nextTime, timer: e}) - continue - } - // If we are set to have OneElementPerKey, and we already have an element for this key we set this to process later. - if v := keyCounts[string(e.keyBytes)]; v > 0 && OneElementPerKey { - notYet = append(notYet, fireElement{firing: nextTime, timer: e}) - continue - } - keyCounts[string(e.keyBytes)]++ - newKeys.insert(string(e.keyBytes)) - if e.timestamp < minTs { - minTs = e.timestamp - } - // TODO: how to deal with watermark holds for this implicit processing time timer - // holdsInBundle[e.holdTimestamp]++ - - state := ss.state[LinkID{}][e.window][string(e.keyBytes)] - endOfWindowReached := e.window.MaxTimestamp() < ss.input - ready := ss.strat.IsTriggerReady(triggerInput{ - newElementCount: 0, - endOfWindowReached: endOfWindowReached, - emNow: emNow, - }, &state) - - if ready { - // We're going to process this trigger! - elems, _ := ss.buildTriggeredBundle(em, string(e.keyBytes), e.window) - toProcess = append(toProcess, elems...) - } - } - - nextTime = ss.processingTimeTimers.Peek() - if nextTime == mtime.MaxTimestamp { - // Escape the loop if there are no more events. - break - } - } - - // Reschedule unfired timers. - notYetHolds := map[mtime.Time]int{} - for _, v := range notYet { - ss.processingTimeTimers.Persist(v.firing, v.timer, notYetHolds) - em.processTimeEvents.Schedule(v.firing, ss.ID) - em.wakeUpAt(v.firing) - } - - // Add a refresh if there are still processing time events to process. - stillSchedulable := (nextTime < emNow && nextTime != mtime.MaxTimestamp || len(notYet) > 0) + return handleProcessingTimeTimer(ss, em, emNow, func(e element, toProcess []element, holdsInBundle map[mtime.Time]int) []element { + // TODO: how to deal with watermark holds for this implicit processing time timer + // holdsInBundle[e.holdTimestamp]++ + state := ss.state[LinkID{}][e.window][string(e.keyBytes)] + endOfWindowReached := e.window.MaxTimestamp() < ss.input + ready := ss.strat.IsTriggerReady(triggerInput{ + newElementCount: 0, + endOfWindowReached: endOfWindowReached, + emNow: emNow, + }, &state) + + if ready { + // We're going to process this trigger! + elems, _ := ss.buildTriggeredBundle(em, string(e.keyBytes), e.window) + toProcess = append(toProcess, elems...) + } + + return toProcess + }) +} - return toProcess, minTs, newKeys, holdsInBundle, stillSchedulable +func (*ordinaryStageKind) buildProcessingTimeBundle(ss *stageState, em *ElementManager, emNow mtime.Time) (elementHeap, mtime.Time, set[string], map[mtime.Time]int, bool) { + slog.Error("ordinary stages can't have processing time elements") + return nil, mtime.MinTimestamp, nil, nil, false } // makeInProgressBundle is common code to store a set of elements as a bundle in progress. From d623c6f789b577174601261002233d6dda663dab Mon Sep 17 00:00:00 2001 From: Shunping Huang Date: Wed, 1 Oct 2025 00:23:18 -0400 Subject: [PATCH 03/13] Set correct pane info. --- .../pkg/beam/runners/prism/internal/engine/elementmanager.go | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go b/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go index 4366071d8302..dcc140cd3b1f 100644 --- a/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go +++ b/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go @@ -2075,9 +2075,13 @@ func (*aggregateStageKind) buildProcessingTimeBundle(ss *stageState, em *Element }, &state) if ready { + state.Pane = computeNextTriggeredPane(state.Pane, endOfWindowReached) + // We're going to process this trigger! elems, _ := ss.buildTriggeredBundle(em, string(e.keyBytes), e.window) toProcess = append(toProcess, elems...) + + ss.state[LinkID{}][e.window][string(e.keyBytes)] = state } return toProcess From c6c0ee5d39a16004a8c94bd3a5e644bade3d68e4 Mon Sep 17 00:00:00 2001 From: Shunping Huang Date: Wed, 1 Oct 2025 10:25:34 -0400 Subject: [PATCH 04/13] Save panes when handling after-processing-time triggers --- .../prism/internal/engine/elementmanager.go | 31 ++++++++++--------- 1 file changed, 17 insertions(+), 14 deletions(-) diff --git a/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go b/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go index dcc140cd3b1f..c7ca18ebef9c 100644 --- a/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go +++ b/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go @@ -1217,7 +1217,7 @@ type stageKind interface { holdsInBundle map[mtime.Time]int, panesInBundle []bundlePane, schedulable bool, pendingAdjustment int) // buildProcessingTimeBundle handles building processing-time bundles for the stage per it's kind. buildProcessingTimeBundle(ss *stageState, em *ElementManager, emNow mtime.Time) (toProcess elementHeap, minTs mtime.Time, newKeys set[string], - holdsInBundle map[mtime.Time]int, schedulable bool) + holdsInBundle map[mtime.Time]int, panesInBundle []bundlePane, schedulable bool) // getPaneOrDefault based on the stage state, element metadata, and bundle id. getPaneOrDefault(ss *stageState, defaultPane typex.PaneInfo, w typex.Window, keyBytes []byte, bundID string) typex.PaneInfo } @@ -1974,19 +1974,19 @@ func (ss *stageState) startProcessingTimeBundle(em *ElementManager, emNow mtime. ss.mu.Lock() defer ss.mu.Unlock() - toProcess, minTs, newKeys, holdsInBundle, stillSchedulable := ss.kind.buildProcessingTimeBundle(ss, em, emNow) + toProcess, minTs, newKeys, holdsInBundle, panesInBundle, stillSchedulable := ss.kind.buildProcessingTimeBundle(ss, em, emNow) if len(toProcess) == 0 { // If we have nothing return "", false, stillSchedulable } - bundID := ss.makeInProgressBundle(genBundID, toProcess, minTs, newKeys, holdsInBundle, nil) + bundID := ss.makeInProgressBundle(genBundID, toProcess, minTs, newKeys, holdsInBundle, panesInBundle) slog.Debug("started a processing time bundle", "stageID", ss.ID, "bundleID", bundID, "size", len(toProcess), "emNow", emNow) return bundID, true, stillSchedulable } func handleProcessingTimeTimer(ss *stageState, em *ElementManager, emNow mtime.Time, - processTimerFn func(e element, toProcess []element, holdsInBundle map[mtime.Time]int) []element) (elementHeap, mtime.Time, set[string], map[mtime.Time]int, bool) { + processTimerFn func(e element, toProcess []element, holdsInBundle map[mtime.Time]int, panesInBundle []bundlePane) ([]element, []bundlePane)) (elementHeap, mtime.Time, set[string], map[mtime.Time]int, []bundlePane, bool) { // TODO: Determine if it's possible and a good idea to treat all EventTime processing as a MinTime // Special Case for ProcessingTime handling. // Eg. Always queue EventTime elements at minTime. @@ -1995,6 +1995,7 @@ func handleProcessingTimeTimer(ss *stageState, em *ElementManager, emNow mtime.T // Potentially puts too much work on the scheduling thread though. var toProcess []element + var panesInBundle []bundlePane minTs := mtime.MaxTimestamp holdsInBundle := map[mtime.Time]int{} @@ -2029,7 +2030,7 @@ func handleProcessingTimeTimer(ss *stageState, em *ElementManager, emNow mtime.T minTs = e.timestamp } - toProcess = processTimerFn(e, toProcess, holdsInBundle) + toProcess, panesInBundle = processTimerFn(e, toProcess, holdsInBundle, panesInBundle) } nextTime = ss.processingTimeTimers.Peek() @@ -2050,20 +2051,20 @@ func handleProcessingTimeTimer(ss *stageState, em *ElementManager, emNow mtime.T // Add a refresh if there are still processing time events to process. stillSchedulable := (nextTime < emNow && nextTime != mtime.MaxTimestamp || len(notYet) > 0) - return toProcess, minTs, newKeys, holdsInBundle, stillSchedulable + return toProcess, minTs, newKeys, holdsInBundle, panesInBundle, stillSchedulable } -func (*statefulStageKind) buildProcessingTimeBundle(ss *stageState, em *ElementManager, emNow mtime.Time) (elementHeap, mtime.Time, set[string], map[mtime.Time]int, bool) { - return handleProcessingTimeTimer(ss, em, emNow, func(e element, toProcess []element, holdsInBundle map[mtime.Time]int) []element { +func (*statefulStageKind) buildProcessingTimeBundle(ss *stageState, em *ElementManager, emNow mtime.Time) (elementHeap, mtime.Time, set[string], map[mtime.Time]int, []bundlePane, bool) { + return handleProcessingTimeTimer(ss, em, emNow, func(e element, toProcess []element, holdsInBundle map[mtime.Time]int, panesInBundle []bundlePane) ([]element, []bundlePane) { holdsInBundle[e.holdTimestamp]++ // We're going to process this timer! toProcess = append(toProcess, e) - return toProcess + return toProcess, nil }) } -func (*aggregateStageKind) buildProcessingTimeBundle(ss *stageState, em *ElementManager, emNow mtime.Time) (elementHeap, mtime.Time, set[string], map[mtime.Time]int, bool) { - return handleProcessingTimeTimer(ss, em, emNow, func(e element, toProcess []element, holdsInBundle map[mtime.Time]int) []element { +func (*aggregateStageKind) buildProcessingTimeBundle(ss *stageState, em *ElementManager, emNow mtime.Time) (elementHeap, mtime.Time, set[string], map[mtime.Time]int, []bundlePane, bool) { + return handleProcessingTimeTimer(ss, em, emNow, func(e element, toProcess []element, holdsInBundle map[mtime.Time]int, panesInBundle []bundlePane) ([]element, []bundlePane) { // TODO: how to deal with watermark holds for this implicit processing time timer // holdsInBundle[e.holdTimestamp]++ state := ss.state[LinkID{}][e.window][string(e.keyBytes)] @@ -2082,15 +2083,17 @@ func (*aggregateStageKind) buildProcessingTimeBundle(ss *stageState, em *Element toProcess = append(toProcess, elems...) ss.state[LinkID{}][e.window][string(e.keyBytes)] = state + + panesInBundle = append(panesInBundle, bundlePane{}) } - return toProcess + return toProcess, panesInBundle }) } -func (*ordinaryStageKind) buildProcessingTimeBundle(ss *stageState, em *ElementManager, emNow mtime.Time) (elementHeap, mtime.Time, set[string], map[mtime.Time]int, bool) { +func (*ordinaryStageKind) buildProcessingTimeBundle(ss *stageState, em *ElementManager, emNow mtime.Time) (elementHeap, mtime.Time, set[string], map[mtime.Time]int, []bundlePane, bool) { slog.Error("ordinary stages can't have processing time elements") - return nil, mtime.MinTimestamp, nil, nil, false + return nil, mtime.MinTimestamp, nil, nil, nil, false } // makeInProgressBundle is common code to store a set of elements as a bundle in progress. From 41f6fdfa41e2ccfcc534aeab34afff3cb8bf4337 Mon Sep 17 00:00:00 2001 From: Shunping Huang Date: Wed, 1 Oct 2025 10:41:48 -0400 Subject: [PATCH 05/13] Add comments --- .../beam/runners/prism/internal/engine/elementmanager.go | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go b/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go index c7ca18ebef9c..a449741ab8c4 100644 --- a/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go +++ b/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go @@ -1602,8 +1602,9 @@ func (ss *stageState) buildTriggeredBundle(em *ElementManager, key string, win t } // startTriggeredBundle must be called with the stage.mu lock held. -// When in discarding mode, returns 0. -// When in accumulating mode, returns the number of fired elements to maintain a correct pending count. +// Returns the accumulation diff that the pending work needs to be adjusted by, as completed work is subtracted from the pending count. +// When in discarding mode, returns 0, as the pending work already includes these elements. +// When in accumulating mode, returns the number of fired elements, since those elements remain pending even after this bundle is fired. func (ss *stageState) startTriggeredBundle(em *ElementManager, key string, win typex.Window) int { toProcess, accumulationDiff := ss.buildTriggeredBundle(em, key, win) if len(toProcess) == 0 { @@ -1985,6 +1986,7 @@ func (ss *stageState) startProcessingTimeBundle(em *ElementManager, emNow mtime. return bundID, true, stillSchedulable } +// handleProcessingTimeTimer contains the common code for handling processing-time timers for aggregation stages and stateful stages. func handleProcessingTimeTimer(ss *stageState, em *ElementManager, emNow mtime.Time, processTimerFn func(e element, toProcess []element, holdsInBundle map[mtime.Time]int, panesInBundle []bundlePane) ([]element, []bundlePane)) (elementHeap, mtime.Time, set[string], map[mtime.Time]int, []bundlePane, bool) { // TODO: Determine if it's possible and a good idea to treat all EventTime processing as a MinTime @@ -2054,6 +2056,7 @@ func handleProcessingTimeTimer(ss *stageState, em *ElementManager, emNow mtime.T return toProcess, minTs, newKeys, holdsInBundle, panesInBundle, stillSchedulable } +// buildProcessingTimeBundle for stateful stages prepares bundles for processing-time timers func (*statefulStageKind) buildProcessingTimeBundle(ss *stageState, em *ElementManager, emNow mtime.Time) (elementHeap, mtime.Time, set[string], map[mtime.Time]int, []bundlePane, bool) { return handleProcessingTimeTimer(ss, em, emNow, func(e element, toProcess []element, holdsInBundle map[mtime.Time]int, panesInBundle []bundlePane) ([]element, []bundlePane) { holdsInBundle[e.holdTimestamp]++ @@ -2063,6 +2066,7 @@ func (*statefulStageKind) buildProcessingTimeBundle(ss *stageState, em *ElementM }) } +// buildProcessingTimeBundle for aggregation stages prepares bundles for after-processing-time triggers func (*aggregateStageKind) buildProcessingTimeBundle(ss *stageState, em *ElementManager, emNow mtime.Time) (elementHeap, mtime.Time, set[string], map[mtime.Time]int, []bundlePane, bool) { return handleProcessingTimeTimer(ss, em, emNow, func(e element, toProcess []element, holdsInBundle map[mtime.Time]int, panesInBundle []bundlePane) ([]element, []bundlePane) { // TODO: how to deal with watermark holds for this implicit processing time timer @@ -2091,6 +2095,7 @@ func (*aggregateStageKind) buildProcessingTimeBundle(ss *stageState, em *Element }) } +// buildProcessingTimeBundle for stateless stages is not supposed to be called currently func (*ordinaryStageKind) buildProcessingTimeBundle(ss *stageState, em *ElementManager, emNow mtime.Time) (elementHeap, mtime.Time, set[string], map[mtime.Time]int, []bundlePane, bool) { slog.Error("ordinary stages can't have processing time elements") return nil, mtime.MinTimestamp, nil, nil, nil, false From 2179b24b5158c54bfc2b587c7c501c1a1ac265df Mon Sep 17 00:00:00 2001 From: Shunping Huang Date: Wed, 1 Oct 2025 14:48:44 -0400 Subject: [PATCH 06/13] Change existing after processing time trigger test and add one more. --- .../prism/internal/unimplemented_test.go | 3 ++- .../test/integration/primitives/windowinto.go | 26 ++++++++++++++++--- 2 files changed, 25 insertions(+), 4 deletions(-) diff --git a/sdks/go/pkg/beam/runners/prism/internal/unimplemented_test.go b/sdks/go/pkg/beam/runners/prism/internal/unimplemented_test.go index 7a742c22d0fb..d54955f43d46 100644 --- a/sdks/go/pkg/beam/runners/prism/internal/unimplemented_test.go +++ b/sdks/go/pkg/beam/runners/prism/internal/unimplemented_test.go @@ -53,7 +53,6 @@ func TestUnimplemented(t *testing.T) { // Currently unimplemented triggers. // https://github.com/apache/beam/issues/31438 {pipeline: primitives.TriggerAfterSynchronizedProcessingTime}, - {pipeline: primitives.TriggerAfterProcessingTime}, } for _, test := range tests { @@ -93,6 +92,8 @@ func TestImplemented(t *testing.T) { {pipeline: primitives.TriggerAfterEach}, {pipeline: primitives.TriggerAfterEndOfWindow}, {pipeline: primitives.TriggerRepeat}, + {pipeline: primitives.TriggerAfterProcessingTime}, + {pipeline: primitives.TriggerAfterProcessingTimeNotTriggered}, } for _, test := range tests { diff --git a/sdks/go/test/integration/primitives/windowinto.go b/sdks/go/test/integration/primitives/windowinto.go index d33e464b76f0..f9b2c431447a 100644 --- a/sdks/go/test/integration/primitives/windowinto.go +++ b/sdks/go/test/integration/primitives/windowinto.go @@ -224,15 +224,35 @@ func TriggerAfterProcessingTime(s beam.Scope) { con := teststream.NewConfig() con.AdvanceProcessingTime(100) con.AddElements(1000, 1.0, 2.0, 3.0) - con.AdvanceProcessingTime(2000) - con.AddElements(22000, 4.0) + con.AdvanceWatermark(1000) // must advance watermark so that elements are processed in the downstream stages + con.AdvanceProcessingTime(5000) // advance processing time to fire the trigger + con.AddElements(2000, 4.0) + con.AdvanceWatermark(2000) + col := teststream.Create(s, con) + + validateEquals(s.Scope("Global"), window.NewGlobalWindows(), col, + []beam.WindowIntoOption{ + beam.Trigger(trigger.AfterProcessingTime().PlusDelay(5 * time.Second)), + }, 6.0, 4.0) +} +// TriggerAfterProcessingTime tests the AfterProcessingTime Trigger, it fires output panes once 't' processing time has passed +// Not yet supported by the flink runner: +// java.lang.UnsupportedOperationException: Advancing Processing time is not supported by the Flink Runner. +func TriggerAfterProcessingTimeNotTriggered(s beam.Scope) { + con := teststream.NewConfig() + con.AdvanceProcessingTime(100) + con.AddElements(1000, 1.0, 2.0, 3.0) + con.AdvanceWatermark(1000) // must advance watermark so that elements are processed in the downstream stages + con.AdvanceProcessingTime(4999) // advance processing time but not enough to fire the trigger + con.AddElements(2000, 4.0) + con.AdvanceWatermark(2000) col := teststream.Create(s, con) validateEquals(s.Scope("Global"), window.NewGlobalWindows(), col, []beam.WindowIntoOption{ beam.Trigger(trigger.AfterProcessingTime().PlusDelay(5 * time.Second)), - }, 6.0) + }, 10.0) } // TriggerRepeat tests the repeat trigger. As of now is it is configure to take only one trigger as a subtrigger. From f869d53418c4b272216ab0ab2324d4dac9467aa7 Mon Sep 17 00:00:00 2001 From: Shunping Huang Date: Thu, 2 Oct 2025 00:31:13 -0400 Subject: [PATCH 07/13] Change the bundleReady criterion for ordinary stage so it does not depend on watermark. --- .../prism/internal/engine/elementmanager.go | 23 ++++++++++++------- 1 file changed, 15 insertions(+), 8 deletions(-) diff --git a/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go b/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go index a449741ab8c4..2764e2b3156f 100644 --- a/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go +++ b/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go @@ -398,7 +398,7 @@ func (em *ElementManager) Bundles(ctx context.Context, upstreamCancelFn context. // If there are no changed stages, ready processing time events, // or injected bundles available, we wait until there are. - for len(em.changedStages)+len(changedByProcessingTime)+len(em.injectedBundles) == 0 { + for len(em.changedStages)+len(changedByProcessingTime)+len(em.injectedBundles)+(len(em.testStreamHandler.events)-em.testStreamHandler.nextEventIndex) == 0 { // Check to see if we must exit select { case <-ctx.Done(): @@ -1633,9 +1633,10 @@ func (ss *stageState) startTriggeredBundle(em *ElementManager, key string, win t ) slog.Debug("started a triggered bundle", "stageID", ss.ID, "bundleID", rb.BundleID, "size", len(toProcess)) - ss.bundlesToInject = append(ss.bundlesToInject, rb) + //ss.bundlesToInject = append(ss.bundlesToInject, rb) // Bundle is marked in progress here to prevent a race condition. em.refreshCond.L.Lock() + em.injectedBundles = append(em.injectedBundles, rb) em.inprogressBundles.insert(rb.BundleID) em.refreshCond.L.Unlock() return accumulationDiff @@ -2381,12 +2382,18 @@ func (ss *stageState) bundleReady(em *ElementManager, emNow mtime.Time) (mtime.T inputW := ss.input _, upstreamW := ss.UpstreamWatermark() previousInputW := ss.previousInput - if inputW == upstreamW && previousInputW == inputW { - slog.Debug("bundleReady: unchanged upstream watermark", - slog.String("stage", ss.ID), - slog.Group("watermark", - slog.Any("upstream == input == previousInput", inputW))) - return mtime.MinTimestamp, false, ptimeEventsReady, injectedReady + if _, ok := ss.kind.(*ordinaryStageKind); !ok { + if inputW == upstreamW && previousInputW == inputW { + slog.Debug("bundleReady: unchanged upstream watermark", + slog.String("stage", ss.ID), + slog.Group("watermark", + slog.Any("upstream == input == previousInput", inputW))) + return mtime.MinTimestamp, false, ptimeEventsReady, injectedReady + } + } else { + if len(ss.pending) == 0 { + return mtime.MinTimestamp, false, ptimeEventsReady, injectedReady + } } ready := true for _, side := range ss.sides { From d6ccb90d481242f7bdcc2b01db74ec5753e96a01 Mon Sep 17 00:00:00 2001 From: Shunping Huang Date: Thu, 2 Oct 2025 00:49:21 -0400 Subject: [PATCH 08/13] Remove advance watermark steps in tests. --- .../prism/internal/engine/elementmanager.go | 5 ++-- .../test/integration/primitives/windowinto.go | 26 +++++++++---------- 2 files changed, 15 insertions(+), 16 deletions(-) diff --git a/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go b/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go index 2764e2b3156f..ab7e5b3265c2 100644 --- a/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go +++ b/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go @@ -398,7 +398,7 @@ func (em *ElementManager) Bundles(ctx context.Context, upstreamCancelFn context. // If there are no changed stages, ready processing time events, // or injected bundles available, we wait until there are. - for len(em.changedStages)+len(changedByProcessingTime)+len(em.injectedBundles)+(len(em.testStreamHandler.events)-em.testStreamHandler.nextEventIndex) == 0 { + for len(em.changedStages)+len(changedByProcessingTime)+len(em.injectedBundles) == 0 { // Check to see if we must exit select { case <-ctx.Done(): @@ -1633,7 +1633,8 @@ func (ss *stageState) startTriggeredBundle(em *ElementManager, key string, win t ) slog.Debug("started a triggered bundle", "stageID", ss.ID, "bundleID", rb.BundleID, "size", len(toProcess)) - //ss.bundlesToInject = append(ss.bundlesToInject, rb) + // TODO: Use ss.bundlesToInject rather than em.injectedBundles + // ss.bundlesToInject = append(ss.bundlesToInject, rb) // Bundle is marked in progress here to prevent a race condition. em.refreshCond.L.Lock() em.injectedBundles = append(em.injectedBundles, rb) diff --git a/sdks/go/test/integration/primitives/windowinto.go b/sdks/go/test/integration/primitives/windowinto.go index f9b2c431447a..f5d01bdfbba5 100644 --- a/sdks/go/test/integration/primitives/windowinto.go +++ b/sdks/go/test/integration/primitives/windowinto.go @@ -217,42 +217,40 @@ func TriggerElementCount(s beam.Scope) { }, 2) } -// TriggerAfterProcessingTime tests the AfterProcessingTime Trigger, it fires output panes once 't' processing time has passed +// TriggerAfterProcessingTimeNotTriggered tests the AfterProcessingTime Trigger. It won't fire because 't' processing time is not reached // Not yet supported by the flink runner: // java.lang.UnsupportedOperationException: Advancing Processing time is not supported by the Flink Runner. -func TriggerAfterProcessingTime(s beam.Scope) { +func TriggerAfterProcessingTimeNotTriggered(s beam.Scope) { con := teststream.NewConfig() con.AdvanceProcessingTime(100) con.AddElements(1000, 1.0, 2.0, 3.0) - con.AdvanceWatermark(1000) // must advance watermark so that elements are processed in the downstream stages - con.AdvanceProcessingTime(5000) // advance processing time to fire the trigger - con.AddElements(2000, 4.0) - con.AdvanceWatermark(2000) + con.AdvanceProcessingTime(4999) // advance processing time but not enough to fire the trigger + con.AddElements(22000, 4.0) + col := teststream.Create(s, con) validateEquals(s.Scope("Global"), window.NewGlobalWindows(), col, []beam.WindowIntoOption{ beam.Trigger(trigger.AfterProcessingTime().PlusDelay(5 * time.Second)), - }, 6.0, 4.0) + }, 10.0) } -// TriggerAfterProcessingTime tests the AfterProcessingTime Trigger, it fires output panes once 't' processing time has passed +// TriggerAfterProcessingTime tests the AfterProcessingTime Trigger. It fires output panes once 't' processing time has passed // Not yet supported by the flink runner: // java.lang.UnsupportedOperationException: Advancing Processing time is not supported by the Flink Runner. -func TriggerAfterProcessingTimeNotTriggered(s beam.Scope) { +func TriggerAfterProcessingTime(s beam.Scope) { con := teststream.NewConfig() con.AdvanceProcessingTime(100) con.AddElements(1000, 1.0, 2.0, 3.0) - con.AdvanceWatermark(1000) // must advance watermark so that elements are processed in the downstream stages - con.AdvanceProcessingTime(4999) // advance processing time but not enough to fire the trigger - con.AddElements(2000, 4.0) - con.AdvanceWatermark(2000) + con.AdvanceProcessingTime(5000) // advance processing time to fire the trigger + con.AddElements(22000, 4.0) + col := teststream.Create(s, con) validateEquals(s.Scope("Global"), window.NewGlobalWindows(), col, []beam.WindowIntoOption{ beam.Trigger(trigger.AfterProcessingTime().PlusDelay(5 * time.Second)), - }, 10.0) + }, 6.0, 4.0) } // TriggerRepeat tests the repeat trigger. As of now is it is configure to take only one trigger as a subtrigger. From 28b2cafd540197b178a3464056774e79f740b739 Mon Sep 17 00:00:00 2001 From: Shunping Huang Date: Thu, 2 Oct 2025 10:09:28 -0400 Subject: [PATCH 09/13] Fix failed vr tests in java related to statelss dofn with side input --- .../prism/internal/engine/elementmanager.go | 15 ++++++++++----- 1 file changed, 10 insertions(+), 5 deletions(-) diff --git a/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go b/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go index ab7e5b3265c2..ca7be5d2c2a0 100644 --- a/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go +++ b/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go @@ -2383,7 +2383,15 @@ func (ss *stageState) bundleReady(em *ElementManager, emNow mtime.Time) (mtime.T inputW := ss.input _, upstreamW := ss.UpstreamWatermark() previousInputW := ss.previousInput - if _, ok := ss.kind.(*ordinaryStageKind); !ok { + + _, isOrdinaryStage := ss.kind.(*ordinaryStageKind) + if isOrdinaryStage && len(ss.sides) == 0 { + // For ordinary stage with no side inputs, we use whether there are pending elements to determine + // whether a bundle is ready or not. + if len(ss.pending) == 0 { + return mtime.MinTimestamp, false, ptimeEventsReady, injectedReady + } + } else { if inputW == upstreamW && previousInputW == inputW { slog.Debug("bundleReady: unchanged upstream watermark", slog.String("stage", ss.ID), @@ -2391,11 +2399,8 @@ func (ss *stageState) bundleReady(em *ElementManager, emNow mtime.Time) (mtime.T slog.Any("upstream == input == previousInput", inputW))) return mtime.MinTimestamp, false, ptimeEventsReady, injectedReady } - } else { - if len(ss.pending) == 0 { - return mtime.MinTimestamp, false, ptimeEventsReady, injectedReady - } } + ready := true for _, side := range ss.sides { pID, ok := em.pcolParents[side.Global] From 1059e0f00bd18e8780c61a7b97ee594cd15008e1 Mon Sep 17 00:00:00 2001 From: Shunping Huang Date: Thu, 2 Oct 2025 11:40:57 -0400 Subject: [PATCH 10/13] Sickbay a test. --- runners/prism/java/build.gradle | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/runners/prism/java/build.gradle b/runners/prism/java/build.gradle index e75fda999e14..732ca724511e 100644 --- a/runners/prism/java/build.gradle +++ b/runners/prism/java/build.gradle @@ -94,6 +94,11 @@ def sickbayTests = [ 'org.apache.beam.sdk.testing.TestStreamTest.testProcessingTimeTrigger', 'org.apache.beam.sdk.testing.TestStreamTest.testLateDataAccumulating', // Uses processing time trigger for early firings. + // A regression introduced when we use number of pending elements rather than watermark to determine + // the bundle readiness of a stateless stage. + // Getting an output of [-999, 1, 1...], but the output array should all be 1s, + 'org.apache.beam.sdk.transforms.ParDoTest$StateTests.testRequiresTimeSortedInputWithLateDataAndAllowedLateness', + // Triggered Side Inputs not yet implemented in Prism. // https://github.com/apache/beam/issues/31438 'org.apache.beam.sdk.transforms.ViewTest.testTriggeredLatestSingleton', From 4179266d5e8e2d24d77c70905195f04c15b7f2b0 Mon Sep 17 00:00:00 2001 From: Shunping Huang Date: Thu, 2 Oct 2025 15:28:28 -0400 Subject: [PATCH 11/13] Add comments to explain the sickbay test. --- runners/prism/java/build.gradle | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/runners/prism/java/build.gradle b/runners/prism/java/build.gradle index 732ca724511e..fd3631fd4a70 100644 --- a/runners/prism/java/build.gradle +++ b/runners/prism/java/build.gradle @@ -96,7 +96,12 @@ def sickbayTests = [ // A regression introduced when we use number of pending elements rather than watermark to determine // the bundle readiness of a stateless stage. - // Getting an output of [-999, 1, 1...], but the output array should all be 1s, + // Currently, Prism processes a bundle of [100, ..., 1000] when watermark is set to 100, + // and then a second bundle of [1, ... 99] when the watermark is set to +inf. + // As a result, it yields an output of [-999, 1, 1...], where -999 comes from the difference between 1000 and 1. + // According to https://beam.apache.org/releases/javadoc/current/org/apache/beam/sdk/transforms/DoFn.RequiresTimeSortedInput.html, + // the stateful dofn with `RequiresTimeSortedInput` annotation should buffer an element until the element's timestamp + allowed_lateness. + // This stateful dofn feature is not yet supported in Prism. 'org.apache.beam.sdk.transforms.ParDoTest$StateTests.testRequiresTimeSortedInputWithLateDataAndAllowedLateness', // Triggered Side Inputs not yet implemented in Prism. From 0d6f0a9edfcbd647f15296aa94f041fe70d7ca59 Mon Sep 17 00:00:00 2001 From: Shunping Huang Date: Thu, 2 Oct 2025 17:37:27 -0400 Subject: [PATCH 12/13] Address reviewer feedback. --- .../prism/internal/engine/elementmanager.go | 19 +++++++++---------- .../integration/primitives/windowinto_test.go | 6 ++++++ 2 files changed, 15 insertions(+), 10 deletions(-) diff --git a/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go b/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go index ca7be5d2c2a0..d03d906e47de 100644 --- a/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go +++ b/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go @@ -2071,8 +2071,8 @@ func (*statefulStageKind) buildProcessingTimeBundle(ss *stageState, em *ElementM // buildProcessingTimeBundle for aggregation stages prepares bundles for after-processing-time triggers func (*aggregateStageKind) buildProcessingTimeBundle(ss *stageState, em *ElementManager, emNow mtime.Time) (elementHeap, mtime.Time, set[string], map[mtime.Time]int, []bundlePane, bool) { return handleProcessingTimeTimer(ss, em, emNow, func(e element, toProcess []element, holdsInBundle map[mtime.Time]int, panesInBundle []bundlePane) ([]element, []bundlePane) { - // TODO: how to deal with watermark holds for this implicit processing time timer - // holdsInBundle[e.holdTimestamp]++ + // Different from `buildProcessingTimeBundle` for stateful stage, + // triggers don't hold back the watermark, so no holds are in the triggered bundle. state := ss.state[LinkID{}][e.window][string(e.keyBytes)] endOfWindowReached := e.window.MaxTimestamp() < ss.input ready := ss.strat.IsTriggerReady(triggerInput{ @@ -2391,14 +2391,13 @@ func (ss *stageState) bundleReady(em *ElementManager, emNow mtime.Time) (mtime.T if len(ss.pending) == 0 { return mtime.MinTimestamp, false, ptimeEventsReady, injectedReady } - } else { - if inputW == upstreamW && previousInputW == inputW { - slog.Debug("bundleReady: unchanged upstream watermark", - slog.String("stage", ss.ID), - slog.Group("watermark", - slog.Any("upstream == input == previousInput", inputW))) - return mtime.MinTimestamp, false, ptimeEventsReady, injectedReady - } + } else if inputW == upstreamW && previousInputW == inputW { + // Otherwise, use the progression of watermark to determine the bundle readiness. + slog.Debug("bundleReady: unchanged upstream watermark", + slog.String("stage", ss.ID), + slog.Group("watermark", + slog.Any("upstream == input == previousInput", inputW))) + return mtime.MinTimestamp, false, ptimeEventsReady, injectedReady } ready := true diff --git a/sdks/go/test/integration/primitives/windowinto_test.go b/sdks/go/test/integration/primitives/windowinto_test.go index 0f2cff5d8f24..39a1df6e9e74 100644 --- a/sdks/go/test/integration/primitives/windowinto_test.go +++ b/sdks/go/test/integration/primitives/windowinto_test.go @@ -77,6 +77,12 @@ func TestTriggerAfterAny(t *testing.T) { ptest.BuildAndRun(t, TriggerAfterAny) } +func TestTriggerAfterProcessingTime(t *testing.T) { + integration.CheckFilters(t) + ptest.BuildAndRun(t, TriggerAfterProcessingTime) + ptest.BuildAndRun(t, TriggerAfterProcessingTimeNotTriggered) +} + func TestTriggerAfterSynchronizedProcessingTime(t *testing.T) { integration.CheckFilters(t) ptest.BuildAndRun(t, TriggerAfterSynchronizedProcessingTime) From a3b986ec22a17e392f341d7aa1475a922a86d0d5 Mon Sep 17 00:00:00 2001 From: Shunping Huang Date: Thu, 2 Oct 2025 17:40:42 -0400 Subject: [PATCH 13/13] Add test filter for flink. --- sdks/go/test/integration/integration.go | 1 + 1 file changed, 1 insertion(+) diff --git a/sdks/go/test/integration/integration.go b/sdks/go/test/integration/integration.go index 8d951fe8ce96..eae64dcb2053 100644 --- a/sdks/go/test/integration/integration.go +++ b/sdks/go/test/integration/integration.go @@ -171,6 +171,7 @@ var flinkFilters = []string{ "TestBigQueryIO.*", "TestBigtableIO.*", "TestSpannerIO.*", + "TestTriggerAfterProcessingTime", // The number of produced outputs in AfterSynchronizedProcessingTime varies in different runs. "TestTriggerAfterSynchronizedProcessingTime", // The flink runner does not support pipeline drain for SDF.