[shunping]

Problem

A stateful DoFn annotated with @RequiresTimeSortedInput should get its elements in event-time order. Per the
javadoc, the runner buffers each element until the watermark passes the element's timestamp plus the allowed lateness.

Prism keeps each key's pending input in a heap ordered by event time, so every bundle is already sorted. But a stateful stage released all pending elements as soon as the watermark advanced. Late but allowed elements that arrived afterwards were processed after elements with later timestamps.

In testRequiresTimeSortedInputWithLateDataAndAllowedLateness, Prism processed [100, ..., 1000] at watermark 100 and then [1, ..., 99], so 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.

[codecov]

Codecov Report

❌ Patch coverage is 68.00000% with 8 lines in your changes missing coverage. Please review.
✅ Project coverage is 56.40%. Comparing base (0f1ac43) to head (06a68dd).
⚠️ Report is 19 commits behind head on master.

Files with missing lines	Patch %	Lines
...s/go/pkg/beam/runners/prism/internal/preprocess.go	33.33%	3 Missing and 1 partial ⚠️
...am/runners/prism/internal/engine/elementmanager.go	88.23%	1 Missing and 1 partial ⚠️
sdks/go/pkg/beam/runners/prism/internal/execute.go	0.00%	1 Missing and 1 partial ⚠️

Additional details and impacted files

@@             Coverage Diff             @@
##           master   #40886       +/-   ##
===========================================
+ Coverage        0   56.40%   +56.40%     
===========================================
  Files           0      876      +876     
  Lines           0   154819   +154819     
===========================================
+ Hits            0    87327    +87327     
- Misses          0    65633    +65633     
- Partials        0     1859     +1859     

Flag	Coverage Δ
go	28.92% <68.00%> (?)
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.

[github-actions]

📋 Review Summary

This PR properly implements support for @RequiresTimeSortedInput in the Prism runner by holding back elements in stateful stages until the input watermark has advanced past their timestamp plus the main input's allowed lateness (or when the watermark reaches +infinity). It also updates stage preprocessing to propagate the requirement and allowed lateness to ElementManager, adds unit tests for the new time-sorted holding behavior, and removes the newly passing Java ValidatesRunner tests from the Prism sickbay.

🔍 General Feedback

• The watermark cutoff calculation in timeSortedCutoff carefully prevents overflow at mtime.MaxTimestamp and underflow via mtime.Subtract.
• hasBuildableDataLocked and buildEventTimeBundle stay consistent with each other by sharing timeSortedReady, ensuring empty bundles are not scheduled when all pending elements are held back for time sorting.

---

💡 Tip: Comment /gemini review on this pull request to trigger a new Gemini code review.

[github-actions]

Assigning reviewers:

R: @jrmccluskey for label go.
Backup reviewer: lostluck (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).