From 64d14896bd7109839ac22fcc79c0dbd79cc883e0 Mon Sep 17 00:00:00 2001 From: Shunping Huang Date: Wed, 28 Oct 2026 20:35:52 -0400 Subject: [PATCH 1/2] [Prism] Drop elements for windows whose trigger has finished Once a window's root trigger has finished, Prism kept accepting new elements and emitted them in an extra pane at end of window. Drop them instead, as Java's ReduceFnRunner and Python's trigger driver do. - Re-enable the TestTrigger* and TestPanes integration tests on Prism. - Move TriggerElementCount and TriggerOrFinally to TestImplemented. - Expect only 6.0 from TriggerAfterProcessingTime: the trigger fires once, then 4.0 arrives in the closed window and is dropped. --- .../prism/internal/engine/elementmanager.go | 21 +++++++++++++++++++ .../prism/internal/unimplemented_test.go | 11 ++++------ sdks/go/test/integration/integration.go | 3 --- .../test/integration/primitives/windowinto.go | 6 ++++-- 4 files changed, 29 insertions(+), 12 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 2ed73e2c6639..2e3d8e4b7e09 100644 --- a/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go +++ b/sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go @@ -1515,6 +1515,20 @@ func (ss *stageState) injectTriggeredBundlesIfReady(em *ElementManager, window t return count } +// isTriggerFinished returns true if the root trigger for the given window and key +// has finished, meaning the window is closed and no further panes will be produced. +// Callers must hold ss.mu. +func (ss *stageState) isTriggerFinished(window typex.Window, key string) bool { + if ss.strat.Trigger == nil { + return false + } + state, ok := ss.state[LinkID{}][window][key] + if !ok { + return false + } + return state.getTriggerState(ss.strat.Trigger).finished +} + // addPending for aggregate stages behaves likes stateful stages, but don't need to handle timers or a separate window // expiration condition. func (*aggregateStageKind) addPending(ss *stageState, em *ElementManager, newPending []element) int { @@ -1543,6 +1557,13 @@ func (*aggregateStageKind) addPending(ss *stageState, em *ElementManager, newPen count := 0 for _, e := range newPending { + if len(e.keyBytes) > 0 && ss.isTriggerFinished(e.window, string(e.keyBytes)) { + // The trigger for this window and key has already finished, so the + // window is closed. Drop the element, matching the Java ReduceFnRunner + // (droppedDueToClosedWindow). Otherwise the element would be emitted + // in an extra pane when the watermark passes the end of the window. + continue + } count++ if len(e.keyBytes) == 0 { panic(fmt.Sprintf("zero length key: %v %v", ss.ID, ss.inputID)) 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 b03d96b04bc1..c3739d510f55 100644 --- a/sdks/go/pkg/beam/runners/prism/internal/unimplemented_test.go +++ b/sdks/go/pkg/beam/runners/prism/internal/unimplemented_test.go @@ -42,13 +42,6 @@ func TestUnimplemented(t *testing.T) { pipeline func(s beam.Scope) }{ // {pipeline: primitives.Drain}, // Can't test drain automatically yet. - - // Implemented but the Go SDK doesn't fully handle panes and - // their associated valid behaviors for these triggers, leading - // to variable results. - // See https://github.com/apache/beam/issues/31153. - {pipeline: primitives.TriggerElementCount}, - {pipeline: primitives.TriggerOrFinally}, } for _, test := range tests { @@ -91,6 +84,10 @@ func TestImplemented(t *testing.T) { {pipeline: primitives.TriggerAfterProcessingTime}, {pipeline: primitives.TriggerAfterProcessingTimeNotTriggered}, {pipeline: primitives.TriggerAfterSynchronizedProcessingTime}, + // These rely on elements being dropped once the window's trigger has + // finished (closed window), rather than emitted in an extra pane. + {pipeline: primitives.TriggerElementCount}, + {pipeline: primitives.TriggerOrFinally}, } for _, test := range tests { diff --git a/sdks/go/test/integration/integration.go b/sdks/go/test/integration/integration.go index 033a19bcdca7..ceb2ad1a36e7 100644 --- a/sdks/go/test/integration/integration.go +++ b/sdks/go/test/integration/integration.go @@ -152,9 +152,6 @@ var portableFilters = []string{ var prismFilters = []string{ // The prism runner does not yet support Java's CoGBK. "TestXLang_CoGroupBy", - // The trigger and pane tests uses TestStream - "TestTrigger.*", - "TestPanes", // TODO(https://github.com/apache/beam/issues/21058): Xlang ios don't yet work on prism. "TestKafkaIO.*", diff --git a/sdks/go/test/integration/primitives/windowinto.go b/sdks/go/test/integration/primitives/windowinto.go index f5d01bdfbba5..8e0224aa10a4 100644 --- a/sdks/go/test/integration/primitives/windowinto.go +++ b/sdks/go/test/integration/primitives/windowinto.go @@ -235,7 +235,9 @@ func TriggerAfterProcessingTimeNotTriggered(s beam.Scope) { }, 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. +// Since the trigger is not repeated, it finishes after the first firing and closes the window, so the +// element arriving after the firing (4.0) is dropped. // 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) { @@ -250,7 +252,7 @@ func TriggerAfterProcessingTime(s beam.Scope) { validateEquals(s.Scope("Global"), window.NewGlobalWindows(), col, []beam.WindowIntoOption{ beam.Trigger(trigger.AfterProcessingTime().PlusDelay(5 * time.Second)), - }, 6.0, 4.0) + }, 6.0) } // TriggerRepeat tests the repeat trigger. As of now is it is configure to take only one trigger as a subtrigger. From df26cdf97a1fd44618ce4747287889bcde55dc39 Mon Sep 17 00:00:00 2001 From: Shunping Huang Date: Wed, 28 Oct 2026 21:19:35 -0400 Subject: [PATCH 2/2] [Python] Update Prism AfterCount streaming test for closed windows AfterCount(3) isn't repeated, so it finishes after its first firing and closes the window. Prism now drops later elements for that window instead of emitting them in an extra pane. --- .../apache_beam/runners/portability/prism_runner_test.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/sdks/python/apache_beam/runners/portability/prism_runner_test.py b/sdks/python/apache_beam/runners/portability/prism_runner_test.py index bb6b7bb04da6..dcb7de6e0c74 100644 --- a/sdks/python/apache_beam/runners/portability/prism_runner_test.py +++ b/sdks/python/apache_beam/runners/portability/prism_runner_test.py @@ -288,15 +288,17 @@ def test_after_count_trigger_streaming(self): ) | beam.GroupByKey() | beam.MapTuple(PrismRunnerTest.format_result)) + # AfterCount(3) is not repeated, so it finishes after its first firing + # and closes the window. Later elements for that window and key + # (A4, A5 and B9) are dropped. Windows whose trigger never fires + # (A in [10, 20)) still emit a final pane when they expire. assert_that( result, equal_to( list([ ('A-3', {1, 2, 3}), - ('A-2', {4, 5}), ('A-2', {10, 11}), ('B-3', {6, 7, 8}), - ('B-1', {9}), ('B-3', {10, 15, 16}), ])))