From f83ffebd70faac80dcad04408f02cadab42e5bc2 Mon Sep 17 00:00:00 2001 From: Danny McCormick Date: Wed, 11 Nov 2026 23:25:00 +0000 Subject: [PATCH 1/2] Require processing time advancement before firing TriggerAfterSynchronizedProcessingTime in Prism --- .../runners/prism/internal/engine/strategy.go | 48 +++++++++++++++++-- .../prism/internal/engine/strategy_test.go | 20 ++++++++ 2 files changed, 63 insertions(+), 5 deletions(-) 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 691f249a5bea..8eb539658880 100644 --- a/sdks/go/pkg/beam/runners/prism/internal/engine/strategy.go +++ b/sdks/go/pkg/beam/runners/prism/internal/engine/strategy.go @@ -732,15 +732,39 @@ func (t *TriggerAfterProcessingTime) String() string { // TriggerAfterSynchronizedProcessingTime is supposed to fires once when processing // time across multiple workers synchronizes with the first element's processing time. -// It is a no-op in the current prism single-node architecture, because we only have -// one worker/machine. Therefore, the trigger just fires once it receives the data. +// In the current prism single-node architecture, it fires once it receives data after +// processing time has advanced past the initial clock time (or previous firing time). type TriggerAfterSynchronizedProcessingTime struct{} -func (t *TriggerAfterSynchronizedProcessingTime) onElement(triggerInput, *StateData) {} +func (t *TriggerAfterSynchronizedProcessingTime) onElement(input triggerInput, state *StateData) { + ts := state.getTriggerState(t) + if ts.finished { + return + } + + if ts.extra == nil { + ts.extra = afterProcessingTimeState{ + emNow: input.emNow, + firingTime: mtime.FromTime(time.Time{}) + 1, + endOfWindowReached: input.endOfWindowReached, + } + } else { + s, _ := ts.extra.(afterProcessingTimeState) + s.emNow = input.emNow + s.endOfWindowReached = input.endOfWindowReached + ts.extra = s + } + + state.setTriggerState(t, ts) +} func (t *TriggerAfterSynchronizedProcessingTime) shouldFire(state *StateData) bool { ts := state.getTriggerState(t) - return !ts.finished + if ts.extra == nil || ts.finished { + return false + } + s := ts.extra.(afterProcessingTimeState) + return s.emNow >= s.firingTime } func (t *TriggerAfterSynchronizedProcessingTime) onFire(state *StateData) { @@ -753,7 +777,21 @@ func (t *TriggerAfterSynchronizedProcessingTime) onFire(state *StateData) { } func (t *TriggerAfterSynchronizedProcessingTime) reset(state *StateData) { - delete(state.Trigger, t) + ts := state.getTriggerState(t) + if ts.extra != nil { + if ts.extra.(afterProcessingTimeState).endOfWindowReached { + delete(state.Trigger, t) + return + } + } + + ts.finished = false + if ts.extra != nil { + s := ts.extra.(afterProcessingTimeState) + s.firingTime = s.emNow + 1 + ts.extra = s + } + state.setTriggerState(t, ts) } func (t *TriggerAfterSynchronizedProcessingTime) String() string { diff --git a/sdks/go/pkg/beam/runners/prism/internal/engine/strategy_test.go b/sdks/go/pkg/beam/runners/prism/internal/engine/strategy_test.go index 3b928be278f8..aad056b381fe 100644 --- a/sdks/go/pkg/beam/runners/prism/internal/engine/strategy_test.go +++ b/sdks/go/pkg/beam/runners/prism/internal/engine/strategy_test.go @@ -549,6 +549,26 @@ func TestTriggers_isReady(t *testing.T) { {triggerInput{emNow: 5000, newElementCount: 1}, false}, // ElmCount = 1 {triggerInput{emNow: 5500, newElementCount: 1}, true}, // ElmCount = 2, fire ElmCount trigger }, + }, { + name: "afterSynchronizedProcessingTime", + trig: &TriggerAfterSynchronizedProcessingTime{}, + inputs: []io{ + {triggerInput{emNow: mtime.FromTime(time.Time{}), newElementCount: 1}, false}, + {triggerInput{emNow: mtime.FromTime(time.Time{}), newElementCount: 1}, false}, + {triggerInput{emNow: 1000, newElementCount: 1}, true}, + {triggerInput{emNow: 2000, newElementCount: 1}, false}, + }, + }, { + name: "afterSynchronizedProcessingTime_Repeated", + trig: &TriggerRepeatedly{&TriggerAfterSynchronizedProcessingTime{}}, + inputs: []io{ + {triggerInput{emNow: mtime.FromTime(time.Time{}), newElementCount: 1}, false}, + {triggerInput{emNow: mtime.FromTime(time.Time{}), newElementCount: 1}, false}, + {triggerInput{emNow: 1000, newElementCount: 1}, true}, + {triggerInput{emNow: 1000, newElementCount: 1}, false}, + {triggerInput{emNow: 2000, newElementCount: 1}, true}, + {triggerInput{emNow: 2000, newElementCount: 1}, false}, + }, }, { name: "default", trig: &TriggerDefault{}, From 129a34a695a28add6b12698c681733aac7cf6cc0 Mon Sep 17 00:00:00 2001 From: Danny McCormick Date: Fri, 13 Nov 2026 05:03:56 +0000 Subject: [PATCH 2/2] Test TriggerAfterSynchronizedProcessingTime without Repeat wrapper in windowinto.go --- .../runners/prism/internal/engine/strategy.go | 48 ++----------------- .../prism/internal/engine/strategy_test.go | 20 -------- .../test/integration/primitives/windowinto.go | 2 +- 3 files changed, 6 insertions(+), 64 deletions(-) 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 8eb539658880..691f249a5bea 100644 --- a/sdks/go/pkg/beam/runners/prism/internal/engine/strategy.go +++ b/sdks/go/pkg/beam/runners/prism/internal/engine/strategy.go @@ -732,39 +732,15 @@ func (t *TriggerAfterProcessingTime) String() string { // TriggerAfterSynchronizedProcessingTime is supposed to fires once when processing // time across multiple workers synchronizes with the first element's processing time. -// In the current prism single-node architecture, it fires once it receives data after -// processing time has advanced past the initial clock time (or previous firing time). +// It is a no-op in the current prism single-node architecture, because we only have +// one worker/machine. Therefore, the trigger just fires once it receives the data. type TriggerAfterSynchronizedProcessingTime struct{} -func (t *TriggerAfterSynchronizedProcessingTime) onElement(input triggerInput, state *StateData) { - ts := state.getTriggerState(t) - if ts.finished { - return - } - - if ts.extra == nil { - ts.extra = afterProcessingTimeState{ - emNow: input.emNow, - firingTime: mtime.FromTime(time.Time{}) + 1, - endOfWindowReached: input.endOfWindowReached, - } - } else { - s, _ := ts.extra.(afterProcessingTimeState) - s.emNow = input.emNow - s.endOfWindowReached = input.endOfWindowReached - ts.extra = s - } - - state.setTriggerState(t, ts) -} +func (t *TriggerAfterSynchronizedProcessingTime) onElement(triggerInput, *StateData) {} func (t *TriggerAfterSynchronizedProcessingTime) shouldFire(state *StateData) bool { ts := state.getTriggerState(t) - if ts.extra == nil || ts.finished { - return false - } - s := ts.extra.(afterProcessingTimeState) - return s.emNow >= s.firingTime + return !ts.finished } func (t *TriggerAfterSynchronizedProcessingTime) onFire(state *StateData) { @@ -777,21 +753,7 @@ func (t *TriggerAfterSynchronizedProcessingTime) onFire(state *StateData) { } func (t *TriggerAfterSynchronizedProcessingTime) reset(state *StateData) { - ts := state.getTriggerState(t) - if ts.extra != nil { - if ts.extra.(afterProcessingTimeState).endOfWindowReached { - delete(state.Trigger, t) - return - } - } - - ts.finished = false - if ts.extra != nil { - s := ts.extra.(afterProcessingTimeState) - s.firingTime = s.emNow + 1 - ts.extra = s - } - state.setTriggerState(t, ts) + delete(state.Trigger, t) } func (t *TriggerAfterSynchronizedProcessingTime) String() string { diff --git a/sdks/go/pkg/beam/runners/prism/internal/engine/strategy_test.go b/sdks/go/pkg/beam/runners/prism/internal/engine/strategy_test.go index aad056b381fe..3b928be278f8 100644 --- a/sdks/go/pkg/beam/runners/prism/internal/engine/strategy_test.go +++ b/sdks/go/pkg/beam/runners/prism/internal/engine/strategy_test.go @@ -549,26 +549,6 @@ func TestTriggers_isReady(t *testing.T) { {triggerInput{emNow: 5000, newElementCount: 1}, false}, // ElmCount = 1 {triggerInput{emNow: 5500, newElementCount: 1}, true}, // ElmCount = 2, fire ElmCount trigger }, - }, { - name: "afterSynchronizedProcessingTime", - trig: &TriggerAfterSynchronizedProcessingTime{}, - inputs: []io{ - {triggerInput{emNow: mtime.FromTime(time.Time{}), newElementCount: 1}, false}, - {triggerInput{emNow: mtime.FromTime(time.Time{}), newElementCount: 1}, false}, - {triggerInput{emNow: 1000, newElementCount: 1}, true}, - {triggerInput{emNow: 2000, newElementCount: 1}, false}, - }, - }, { - name: "afterSynchronizedProcessingTime_Repeated", - trig: &TriggerRepeatedly{&TriggerAfterSynchronizedProcessingTime{}}, - inputs: []io{ - {triggerInput{emNow: mtime.FromTime(time.Time{}), newElementCount: 1}, false}, - {triggerInput{emNow: mtime.FromTime(time.Time{}), newElementCount: 1}, false}, - {triggerInput{emNow: 1000, newElementCount: 1}, true}, - {triggerInput{emNow: 1000, newElementCount: 1}, false}, - {triggerInput{emNow: 2000, newElementCount: 1}, true}, - {triggerInput{emNow: 2000, newElementCount: 1}, false}, - }, }, { name: "default", trig: &TriggerDefault{}, diff --git a/sdks/go/test/integration/primitives/windowinto.go b/sdks/go/test/integration/primitives/windowinto.go index 8e0224aa10a4..2efaf014567b 100644 --- a/sdks/go/test/integration/primitives/windowinto.go +++ b/sdks/go/test/integration/primitives/windowinto.go @@ -374,7 +374,7 @@ func TriggerAfterSynchronizedProcessingTime(s beam.Scope) { con.AddElements(12000, 5.0, 8.0) col := teststream.Create(s, con) - trigger := trigger.Repeat(trigger.AfterSynchronizedProcessingTime()) + trigger := trigger.AfterSynchronizedProcessingTime() windowSize := 10 * time.Second validateCount(s.Scope("Global"), window.NewFixedWindows(windowSize), col, []beam.WindowIntoOption{