[damccorm]

Summary

• Fixes a flake in :sdks:go:test:prismValidatesRunner (TestTriggerAfterSynchronizedProcessingTime failing with expected 2 collections, but got: 3 in https://github.com/apache/beam/actions/runs/43083145592 and https://github.com/apache/beam/actions/runs/43099848241).
• Previously, TriggerAfterSynchronizedProcessingTime in Prism had shouldFire return !ts.finished without checking whether processing time had advanced past the initial clock time (or past the previous firing time when wrapped in TriggerRepeatedly).
• In TestTriggerAfterSynchronizedProcessingTime (which uses trigger.Repeat(trigger.AfterSynchronizedProcessingTime()) without advancing TestStream processing time), whenever stage.Execute channel-split the initial 3-element TestStream bundle into two bundles in Docker mode, Repeat(AfterSynchronizedProcessingTime()) fired once on each split bundle in window [0, 10000), emitting 3 panes total instead of 2.
• Track afterProcessingTimeState in TriggerAfterSynchronizedProcessingTime so it requires processing time to advance past the initial TestStream clock time (mtime.FromTime(time.Time{})) before firing, and advances firingTime to s.emNow + 1 on reset when repeated.

Test Plan

• go test ./pkg/beam/runners/prism/...
• go test -v ./test/integration/primitives -run "TestTrigger.*"
• ./gradlew :runners:prism:java:prismLoopbackValidatesRunnerTests --tests "org.apache.beam.sdk.transforms.GroupByKeyTest*" --tests "org.apache.beam.sdk.testing.TestStreamTest"

Verification

• Triggered beam_PreCommit_GoPrism.yml workflow run on branch: 43119696481

[damccorm]

/gemini review

[github-actions]

📋 Review Summary

This PR fixes a flake in TestTriggerAfterSynchronizedProcessingTime when running on Prism in Docker mode. By tracking afterProcessingTimeState in TriggerAfterSynchronizedProcessingTime, the trigger now waits for processing time to advance past the initial TestStream clock time (and past previous firing times on reset) instead of firing unconditionally on every split bundle.

🔍 General Feedback

• The implementation cleanly reuses afterProcessingTimeState and aligns well with TriggerAfterProcessingTime's state lifecycle (onElement, shouldFire, onFire, reset).
• Unit tests in strategy_test.go cover both single and repeated TriggerAfterSynchronizedProcessingTime scenarios.

---

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

[codecov]

Codecov Report

❌ Patch coverage is 90.00000% with 3 lines in your changes missing coverage. Please review.
✅ Project coverage is 57.77%. Comparing base (73b1202) to head (f83ffeb).
⚠️ Report is 1 commits behind head on master.

Files with missing lines	Patch %	Lines
...pkg/beam/runners/prism/internal/engine/strategy.go	90.00%	2 Missing and 1 partial ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #41228      +/-   ##
============================================
+ Coverage     57.76%   57.77%   +0.01%     
  Complexity     2617     2617              
============================================
  Files          1202     1202              
  Lines        171675   171702      +27     
  Branches       2163     2163              
============================================
+ Hits          99162    99198      +36     
+ Misses        69851    69838      -13     
- Partials       2662     2666       +4     

Flag	Coverage Δ
go	29.25% <90.00%> (+0.04%)	⬆️
python	80.55% <ø> (-0.01%)	⬇️

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.