shunping opened a new pull request, #40878:
URL: https://github.com/apache/beam/pull/40878
## 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
--
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]
