shunping opened a new pull request, #40886: URL: https://github.com/apache/beam/pull/40886
## Problem A stateful DoFn annotated with `@RequiresTimeSortedInput` should get its elements in event-time order. Per the [javadoc](https://beam.apache.org/releases/javadoc/current/org/apache/beam/sdk/transforms/DoFn.RequiresTimeSortedInput.html), the runner buffers each element until the watermark passes the element's timestamp plus the allowed lateness. Prism only made such a stage a fusion root and sorted whatever was pending when the watermark advanced. It didn't take allowed lateness into account. In `ParDoTest$StateTests.testRequiresTimeSortedInputWithLateDataAndAllowedLateness`: - the watermark advances to 100 after elements 1000..100 arrive, so Prism processed `[100, ..., 1000]` right away; - the late but allowed elements `[1, ..., 99]` were processed in a second bundle; - the output started with `-999` instead of being all ones. ## Change - A stage with a ParDo that requires time sorted input now records its main input's allowed lateness. - Such a stage holds each element until the watermark passes `timestamp + allowedLateness`, then processes held elements in timestamp order. - Other stages behave as before. ## Testing - Added the Go unit test `TestStatefulStageKind_buildEventTimeBundle_TimeSortedInput`. - Removed `testRequiresTimeSortedInputWithLateDataAndAllowedLateness` from the Prism sickbay. - The full Prism Java ValidatesRunner suite and the Go Prism tests pass. ## Known limitation A key's timers that sit behind held data in its queue also wait until that data is released. Java delivers timers on the watermark, independently of buffered elements. No current test depends on that. -- 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]
