[shunping]

Summary

In Beam, when a window's root trigger finishes (it will never fire again), the window is closed. The firing that finished the trigger is the final pane, no more panes are emitted, and any elements that arrive later are dropped.
Prism didn't do this. It kept buffering elements for a closed window and emitted them in an extra pane when the watermark passed the end of the window. This PR makes Prism drop those elements, matching the Java and Python runners, and re-enables the trigger integration tests on Prism.
Related: #31153

Changes

1. Prism: drop elements for closed windows (runners/prism/internal/engine/elementmanager.go)

• New stageState.isTriggerFinished(window, key) reports whether the root trigger for a window and key has finished.
• aggregateStageKind.addPending uses it to drop elements for closed windows. The check runs per element, so in streaming mode an element that finishes the trigger causes later elements in the same batch to be dropped.
  Observed panes from Prism before and after the fix:

Pipeline	Before	After
TriggerElementCount: AfterCount(2), fixed 10s windows	[0,10): {1,2}, {3,4,5}; [50,60): {10} → 3 panes	{1,2}, {10} → 2
TriggerOrFinally: OrFinally(Always, AfterCount(2))	[0,10): {1}, {2}, {3}; [10,20): {5}, {8} → 5 panes	{1}, {2}, {5}, {8} → 4
TriggerAfterProcessingTime: AfterProcessingTime(+5s), not repeated, global window	6, 4	6

2. Tests

• test/integration/integration.go: removed the Prism filters for TestTrigger.* and TestPanes. They now all pass on Prism.
• runners/prism/internal/unimplemented_test.go: moved TriggerElementCount and TriggerOrFinally from TestUnimplemented to TestImplemented.
• test/integration/primitives/windowinto.go: TriggerAfterProcessingTime now expects only 6.0. The trigger isn't repeated, so it finishes after its first firing; 4.0 then arrives in the closed window and is dropped. No other runner runs this test: direct, portable, spark and dataflow filter TestTrigger.*, and flink filters it explicitly.

Why this is the expected behavior

See #40879 for details.

[codecov]

Codecov Report

❌ Patch coverage is 55.55556% with 4 lines in your changes missing coverage. Please review.
✅ Project coverage is 56.40%. Comparing base (0f1ac43) to head (df26cdf).
⚠️ Report is 16 commits behind head on master.

Files with missing lines	Patch %	Lines
...am/runners/prism/internal/engine/elementmanager.go	55.55%	2 Missing and 2 partials ⚠️

Additional details and impacted files

@@             Coverage Diff             @@
##           master   #40878       +/-   ##
===========================================
+ Coverage        0   56.40%   +56.40%     
===========================================
  Files           0      876      +876     
  Lines           0   154798   +154798     
===========================================
+ Hits            0    87319    +87319     
- Misses          0    65617    +65617     
- Partials        0     1862     +1862     

Flag	Coverage Δ
go	28.93% <55.55%> (?)
python	79.82% <ø> (?)

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[shunping]

Thanks @lostluck! This is something I always want to figure out. Now it is easier with bots. Please take a look at #40879 too when you have a chance.

I still remembered our discussion on the end-of-window firing and the ClosingBehavior / OnTimeBehavior before. Good times!

[lostluck]

Yeah, this is great to have narrowed down! Especially the documentation update in the next PR. The bots are very good extensions of oneself if you already have plenty of information to start. Glad they're working out for Beam!

[github-actions]

Assigning reviewers:

R: @lostluck for label go.
Backup reviewer: jrmccluskey (if the main reviewer is unresponsive, you can escalate to them by tagging their username)

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).