prodriguezdefino opened a new pull request, #41075:
URL: https://github.com/apache/beam/pull/41075
… elements
Problem:
In stageState.buildTriggeredBundle and
aggregateStageKind.buildEventTimeBundle
(sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go), firing an
aggregation pane in Accumulating mode pushes the emitted elements back onto
dnt.elements with their original event timestamps:
if ss.strat.Accumulating {
for _, e := range toProcess {
heap.Push(&dnt.elements, e)
}
}
stageState.minPendingTimestampLocked takes the minimum of
dnt.elements[0].timestamp across every key, which holds ss.input at the
earliest emitted element's timestamp even after the upstream watermark passes
window.MaxTimestamp(). When a late element arrives, endOfWindowReached :=
window.MaxTimestamp() < ss.input evaluates to false. TriggerAfterEndOfWindow
stays on its Early sub-trigger instead of switching to Late, and
computeNextTriggeredPane resets pane.Timing to PaneEarly instead of advancing
to PaneLate.
Fix:
Store already-emitted accumulating elements in StateData.Accumulated outside
dnt.elements so they do not hold minPendingTimestampLocked, track owed ON_TIME
watermark firings and window-end holds in stageState.pendingOnTime, merge
accumulated elements with new pending elements on subsequent firings, and
discard them on window expiration. Skip internal LinkID{} aggregation state in
StateForBundle, and add tests in elementmanager_test.go and
handlerunner_test.go.
Reproduction (Python SDK on PrismRunner):
```python
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.testing.test_stream import TestStream
from apache_beam.testing.util import assert_that, equal_to
from apache_beam.transforms.trigger import AccumulationMode, AfterCount,
AfterWatermark
from apache_beam.transforms.window import FixedWindows, TimestampedValue
from apache_beam.utils.windowed_value import PaneInfoTiming
class FormatPaneDoFn(beam.DoFn):
def process(self, element, pane_info=beam.DoFn.PaneInfoParam):
_, vals = element
timing = PaneInfoTiming.to_string(pane_info.timing)
yield (timing, sorted(vals))
if __name__ == "__main__":
options = PipelineOptions(["--runner=PrismRunner",
"--environment_type=LOOPBACK"])
with beam.Pipeline(options=options) as p:
out = (
p
| TestStream()
.add_elements([TimestampedValue(("k", 1), 1)])
.advance_watermark_to(20)
.add_elements([TimestampedValue(("k", 2), 2)])
.advance_watermark_to_infinity()
| beam.WindowInto(
FixedWindows(10),
trigger=AfterWatermark(early=AfterCount(1),
late=AfterCount(1)),
accumulation_mode=AccumulationMode.ACCUMULATING,
allowed_lateness=60,
)
| beam.GroupByKey()
| beam.ParDo(FormatPaneDoFn())
)
assert_that(
out,
equal_to([("EARLY", [1]), ("ON_TIME", [1]), ("LATE", [1, 2])]),
)
```
Before the fix: Prism outputs `[('EARLY', [1]), ('ON_TIME', [1]), ('EARLY',
[1, 2]), ('ON_TIME', [1, 2])]`, and `assert_that` fails.
After the fix: Prism outputs `[('EARLY', [1]), ('ON_TIME', [1]), ('LATE',
[1, 2])]`.
<!--
Thank you for your contribution! Follow this checklist to help us
incorporate your contribution quickly and easily:
- [ ] If this contribution is large, please file an Apache [Individual
Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
- [ ] Replace these instructions with a description of your change and why
you have made it.
- [ ] Mention the appropriate issue in your description (for example:
`addresses #123` or `fixes #456`).
- [ ] Update `CHANGES.md` with noteworthy changes.
See [CONTRIBUTING.md]() for more details.
-->
--
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]
