shunping opened a new pull request, #40879:
URL: https://github.com/apache/beam/pull/40879
## Summary
The `Trigger` message in `beam_runner_api.proto` only says when each trigger
is *ready* to fire. It doesn't say what happens when a trigger **finishes**,
meaning it will never fire again. Runners depend on that: when a window's root
trigger finishes, the window is closed, no more panes are emitted, and later
elements are dropped.
Java and Python already behave this way, but the rules live only in SDK and
runner code. Prism behaved differently until #<Prism PR>. This PR adds the
rules to the model as comments only; there are no schema changes.
## Changes
**`Trigger` header comment**
- Triggers are evaluated per key and window; each time the root trigger is
ready, the current pane is output.
- When the *root* trigger finishes, the window is closed for that key:
- the firing that finished it is the final pane (`PaneInfo.is_last`);
- no further panes are produced, including at end of window or window
expiry, regardless of `ClosingBehavior` / `OnTimeBehavior`;
- elements arriving after the window closes are dropped, whether or not
they are late relative to the watermark.
- A root trigger that may finish before the window expires can lose data, so
SDKs call it "unsafe". `Repeat` makes a trigger never finish. Runners must
still implement the closed-window rules for pipelines that use finishing
triggers.
- A finishing *subtrigger* doesn't close the window on its own; it only
affects its parent.
**When each trigger type finishes**
| Trigger | Finishes | Java state machine |
|---|---|---|
| `AfterAll` | after firing once |
[AfterAllStateMachine#L111](https://github.com/apache/beam/blob/0c5adeaff12d46edffa88adbb43fd4ea8fad0f26/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterAllStateMachine.java#L111)
|
| `AfterAny` | after firing once |
[AfterFirstStateMachine#L108](https://github.com/apache/beam/blob/0c5adeaff12d46edffa88adbb43fd4ea8fad0f26/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterFirstStateMachine.java#L108)
|
| `AfterEach` | when its last subtrigger finishes |
[AfterEachStateMachine#L161](https://github.com/apache/beam/blob/0c5adeaff12d46edffa88adbb43fd4ea8fad0f26/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterEachStateMachine.java#L161)
|
| `AfterEndOfWindow` | with no `late_firings`, on its firing after end of
window; with `late_firings`, never before expiry (late trigger repeats
implicitly) |
[AfterWatermarkStateMachine#L196-L204](https://github.com/apache/beam/blob/0c5adeaff12d46edffa88adbb43fd4ea8fad0f26/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterWatermarkStateMachine.java#L196-L204),
[#L258](https://github.com/apache/beam/blob/0c5adeaff12d46edffa88adbb43fd4ea8fad0f26/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterWatermarkStateMachine.java#L258)
|
| `AfterProcessingTime`, `AfterSynchronizedProcessingTime` | after firing
once |
[AfterDelayFromFirstElementStateMachine#L231](https://github.com/apache/beam/blob/0c5adeaff12d46edffa88adbb43fd4ea8fad0f26/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterDelayFromFirstElementStateMachine.java#L231)
|
| `ElementCount` | after firing once |
[AfterPaneStateMachine#L133](https://github.com/apache/beam/blob/0c5adeaff12d46edffa88adbb43fd4ea8fad0f26/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/AfterPaneStateMachine.java#L133)
|
| `OrFinally` | when `finally` fires, or `main` finishes, whichever comes
first |
[OrFinallyStateMachine#L77-L104](https://github.com/apache/beam/blob/0c5adeaff12d46edffa88adbb43fd4ea8fad0f26/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/OrFinallyStateMachine.java#L77-L104)
|
| `Repeat` | never; the subtrigger is reset when it finishes |
[RepeatedlyStateMachine#L89](https://github.com/apache/beam/blob/0c5adeaff12d46edffa88adbb43fd4ea8fad0f26/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/RepeatedlyStateMachine.java#L89)
|
| `Default`, `Always` | never | |
| `Never` | never before window expiry | |
**`ClosingBehavior.EMIT_ALWAYS`**: notes that it doesn't apply if the window
was already closed by a finished root trigger. In that case no pane is emitted
at expiry.
## References
**What "finishing" a root trigger means**
- Java `Trigger` javadoc,
[Trigger.java#L41-L45](https://github.com/apache/beam/blob/0c5adeaff12d46edffa88adbb43fd4ea8fad0f26/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/Trigger.java#L41-L45):
> When the root trigger finishes (indicating it will never fire again),
the window is closed and any new elements assigned to that window are discarded.
- Java runner,
[ReduceFnRunner.java](https://github.com/apache/beam/blob/0c5adeaff12d46edffa88adbb43fd4ea8fad0f26/runners/core-java/src/main/java/org/apache/beam/runners/core/ReduceFnRunner.java):
-
[L643-L655](https://github.com/apache/beam/blob/0c5adeaff12d46edffa88adbb43fd4ea8fad0f26/runners/core-java/src/main/java/org/apache/beam/runners/core/ReduceFnRunner.java#L643-L655):
elements for a closed window are dropped and counted in
`droppedDueToClosedWindow`.
-
[L853-L866](https://github.com/apache/beam/blob/0c5adeaff12d46edffa88adbb43fd4ea8fad0f26/runners/core-java/src/main/java/org/apache/beam/runners/core/ReduceFnRunner.java#L853-L866)
and
[L879-L883](https://github.com/apache/beam/blob/0c5adeaff12d46edffa88adbb43fd4ea8fad0f26/runners/core-java/src/main/java/org/apache/beam/runners/core/ReduceFnRunner.java#L879-L883):
end-of-window and expiry panes fire only while `windowIsActiveAndOpen()`, i.e.
the window is active and its trigger hasn't finished. So a closed window gets
no extra pane, whatever `ClosingBehavior` says.
- Python: once a window's trigger finishes, `GeneralTriggerDriver` in
[trigger.py](https://github.com/apache/beam/blob/0c5adeaff12d46edffa88adbb43fd4ea8fad0f26/sdks/python/apache_beam/transforms/trigger.py)
marks it with a `TOMBSTONE` and skips later elements for it.
- Programming guide, [§9.5 Composite
triggers](https://github.com/apache/beam/blob/0c5adeaff12d46edffa88adbb43fd4ea8fad0f26/website/www/site/content/en/documentation/programming-guide.md#L6082-L6109):
triggers can emit "repeatedly, at most once", and `orFinally` makes a trigger
"fire one final time and never fire again".
**"Unsafe" triggers**
- Java `Trigger.mayFinish()`,
[Trigger.java#L143-L147](https://github.com/apache/beam/blob/0c5adeaff12d46edffa88adbb43fd4ea8fad0f26/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/windowing/Trigger.java#L143-L147):
"A top level trigger that finishes can cause data loss, so is rejected by
GroupByKey validation."
- Java `GroupByKey`,
[GroupByKey.java#L185-L245](https://github.com/apache/beam/blob/0c5adeaff12d46edffa88adbb43fd4ea8fad0f26/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupByKey.java#L185-L245):
`triggerIsSafe` rejects root triggers that may finish, except `Never`, and
end-of-window triggers with zero allowed lateness or with late firings. The
error reads "Unsafe trigger '…' may lose data, did you mean to wrap it in
`Repeatedly.forever(...)`?".
- Python:
-
[trigger.py#L160-L181](https://github.com/apache/beam/blob/0c5adeaff12d46edffa88adbb43fd4ea8fad0f26/sdks/python/apache_beam/transforms/trigger.py#L160-L181):
`DataLossReason.MAY_FINISH`, "data that comes in after the trigger may be
lost. Example: AfterCount(1) will stop firing after the first element."
-
[trigger.py#L269-L294](https://github.com/apache/beam/blob/0c5adeaff12d46edffa88adbb43fd4ea8fad0f26/sdks/python/apache_beam/transforms/trigger.py#L269-L294):
`may_lose_data()`.
-
[core.py#L3524-L3540](https://github.com/apache/beam/blob/0c5adeaff12d46edffa88adbb43fd4ea8fad0f26/sdks/python/apache_beam/transforms/core.py#L3524-L3540):
`GroupByKey` raises "Unsafe trigger … may lose data" unless
`--allow_unsafe_triggers` is set.
- Design doc: [Guard against "Trigger Finishing", a data loss
risk](https://s.apache.org/finishing-triggers-drop-data), linked from the Java
error message.
--
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]
