[Prism] Support AfterProcessingTime triggers - part 2 #36333

shunping · 2025-09-30T16:55:44Z

This PR is a follow-up to #36126. The majority of this PR has been reviewed in #36069, and I've added comments to the code to indicate where the feedback has been addressed.

Part of the work to support AfterProcessingTime trigger (#31438)

shunping · 2025-09-30T17:03:35Z

sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go

+// This is used for processing time timers to ensure the loop re-evaluates
+// stages when a processing time timer is expected to fire.
+func (em *ElementManager) wakeUpAt(t mtime.Time) {
+	if em.testStreamHandler == nil && em.config.EnableRTC {


Addressed https://github.com/apache/beam/pull/36069/files#r2342213066.

We should also only be creating this Goroutine when the Real Time Clock is being used, and testStreamHandler is nil. Otherwise behavior will be additionally wonky.

shunping · 2025-09-30T17:05:50Z

sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go

+	})
+}
+
+func (*ordinaryStageKind) buildProcessingTimeBundle(ss *stageState, em *ElementManager, emNow mtime.Time) (elementHeap, mtime.Time, set[string], map[mtime.Time]int, bool) {


Addressed https://github.com/apache/beam/pull/36069/files#r2342012446

Feel free to drop the named return values, since we aren't using them here...

shunping · 2025-09-30T17:07:03Z

sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go

-	slog.Debug("started a processing time bundle", "stageID", ss.ID, "bundleID", bundID, "size", len(toProcess), "emNow", emNow)
-	return bundID, true, stillSchedulable
+func (*statefulStageKind) buildProcessingTimeBundle(ss *stageState, em *ElementManager, emNow mtime.Time) (elementHeap, mtime.Time, set[string], map[mtime.Time]int, bool) {
+	return handleProcessingTimeTimer(ss, em, emNow, func(e element, toProcess []element, holdsInBundle map[mtime.Time]int) {


Addressed https://github.com/apache/beam/pull/36069/files#r2342029890

Correct me if I'm wrong, I think this is the only difference preventing us from a single shared method between the general stateful handler and the aggregation handler?
(That is, a single method on ss.state that both kinds call)

…kinds into one code path.

github-actions · 2025-10-01T19:34:53Z

Assigning reviewers:

R: @jrmccluskey for label go.

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).

sdks/go/test/integration/primitives/windowinto.go

lostluck

Are you leaving non-Go tests to another PR?

Are there any Java conformance tests that start passing now that this is in?

shunping · 2025-10-01T21:56:10Z

Are you leaving non-Go tests to another PR?

Are there any Java conformance tests that start passing now that this is in?

Yes. I tried some commented Java tests and it seems that they need AfterSynchronizedProcessingTime trigger to run.

For Python ones, we will need to clear up some tests later.

lostluck · 2025-10-01T22:04:57Z

Well, the goodnews is I'm not sure there's any "work" to be done here for AfterSynchronizedProcessingTime, since it depends on all the workers to have caught up to the first element's arrival.

https://beam.apache.org/releases/javadoc/2.68.0/org/apache/beam/sdk/transforms/windowing/AfterSynchronizedProcessingTime.html

But in principle, prism is always just a single worker (there can be multiple SDK stacks, but they're not workers per-se), but the "notion" of processing time across workers is probably a no-op in the current generation of Prism as a single Node local runner.

shunping · 2025-10-01T23:43:17Z

Well, the goodnews is I'm not sure there's any "work" to be done here for AfterSynchronizedProcessingTime, since it depends on all the workers to have caught up to the first element's arrival.

https://beam.apache.org/releases/javadoc/2.68.0/org/apache/beam/sdk/transforms/windowing/AfterSynchronizedProcessingTime.html

But in principle, prism is always just a single worker (there can be multiple SDK stacks, but they're not workers per-se), but the "notion" of processing time across workers is probably a no-op in the current generation of Prism as a single Node local runner.

So in the current setting of prism, it sounds like the AfterSynchronizedProcessingTime is a AftetCount(1)? I see it is a subclass of OnceTrigger in the Java doc, so it is only fired once.

lostluck · 2025-10-02T00:32:08Z

I think so yes.

It's a trigger only implemented in Java (and Go i guess...) IIRC and labeled as an implementation detail too. It's just unfortunately externalized too much so the internality isn't maintained well.

Beam's cross worker communication doesn't report a machine's local wall time, so at present there's no way to do this. To implement it properly requires the runner to manage its own machines/worker side communications better, like how Dataflow has runnerv2 for worker work acquisition.

…pend on watermark.

shunping · 2025-10-02T05:01:16Z

Thanks!

The fact that some test pipelines are stuck if there is no watermark advancing is truly a problem.

I am looking into another approach to solve that. Basically, we will change bundleReady so for ordinary stages, it won't wait for the watermark advancing to process (I marked the change I made in the code).

PTAL. I am running tests to see if the idea works. If it works, we will no longer need TestStream.advanceWatermark calls in our processing-time trigger tests.

shunping · 2025-10-02T05:02:22Z

sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go

+	// ss.bundlesToInject = append(ss.bundlesToInject, rb)
 	// Bundle is marked in progress here to prevent a race condition.
 	em.refreshCond.L.Lock()
+	em.injectedBundles = append(em.injectedBundles, rb)


I use em.injectedBundles here, because the stage-wise bundlesToInject is causing the pipeline stuck. This needs to be consolidated in the future PR.

If using this mechanism also add "em.inprogressBundles.insert(rb.BundleID)" to avoid "losing" the bundle. injectedBundles are considered "inprogress" even when they aren't yet executing. This is to avoid incorrect premature terminations.

Very odd that there's a stuckness going on with bundlesToInject. Not sure why that would be offhand.

If using this mechanism also add "em.inprogressBundles.insert(rb.BundleID)" to avoid "losing" the bundle. injectedBundles are considered "inprogress" even when they aren't yet executing. This is to avoid incorrect premature terminations.

Yes, the line you mentioned is already there.

shunping · 2025-10-02T14:10:51Z

sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go

+	if isOrdinaryStage && len(ss.sides) == 0 {
+		// For ordinary stage with no side inputs, we use whether there are pending elements to determine
+		// whether a bundle is ready or not.
+		if len(ss.pending) == 0 {


Use watermark to determine whether to create a bundle for aggregation and stateful stages, but not stateless stages without sideinputs.

codecov · 2025-10-02T14:17:32Z

Codecov Report

❌ Patch coverage is 49.03226% with 79 lines in your changes missing coverage. Please review.
✅ Project coverage is 56.84%. Comparing base (9cd994e) to head (28b2caf).
⚠️ Report is 27 commits behind head on master.

Files with missing lines	Patch %	Lines
...am/runners/prism/internal/engine/elementmanager.go	45.29%	60 Missing and 4 partials ⚠️
...pkg/beam/runners/prism/internal/engine/strategy.go	62.16%	13 Missing and 1 partial ⚠️
...m/runners/prism/internal/jobservices/management.go	0.00%	1 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #36333      +/-   ##
============================================
- Coverage     56.84%   56.84%   -0.01%     
  Complexity     3386     3386              
============================================
  Files          1220     1220              
  Lines        185898   186056     +158     
  Branches       3523     3523              
============================================
+ Hits         105672   105757      +85     
- Misses        76885    76942      +57     
- Partials       3341     3357      +16

Flag	Coverage Δ
go	`28.43% <49.03%> (+0.06%)`	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Sentry.
📢 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.

runners/prism/java/build.gradle

lostluck

LGTM, thanks for all the detailed work!

lostluck · 2025-10-02T19:54:40Z

sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go

+	// ss.bundlesToInject = append(ss.bundlesToInject, rb)
 	// Bundle is marked in progress here to prevent a race condition.
 	em.refreshCond.L.Lock()
+	em.injectedBundles = append(em.injectedBundles, rb)


If using this mechanism also add "em.inprogressBundles.insert(rb.BundleID)" to avoid "losing" the bundle. injectedBundles are considered "inprogress" even when they aren't yet executing. This is to avoid incorrect premature terminations.

lostluck · 2025-10-02T19:56:12Z

sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go

+	// ss.bundlesToInject = append(ss.bundlesToInject, rb)
 	// Bundle is marked in progress here to prevent a race condition.
 	em.refreshCond.L.Lock()
+	em.injectedBundles = append(em.injectedBundles, rb)


Very odd that there's a stuckness going on with bundlesToInject. Not sure why that would be offhand.

lostluck · 2025-10-02T19:58:27Z

sdks/go/test/integration/primitives/windowinto.go

+// TriggerAfterProcessingTimeNotTriggered tests the AfterProcessingTime Trigger. It won't fire because 't' processing time is not reached
+// Not yet supported by the flink runner:
+// java.lang.UnsupportedOperationException: Advancing Processing time is not supported by the Flink Runner.
+func TriggerAfterProcessingTimeNotTriggered(s beam.Scope) {


Also add a Test wrapper to

https://github.com/apache/beam/blob/3903623850dcad24e4a02f1eb7f81e13d4ea4d54/sdks/go/test/integration/primitives/windowinto_test.go

So this can get executed in the other runner suites, even if it's currently getting filtered out, it's important to not "miss" it. That's also where the "test name" for it gets added so it's visible in the lists at

https://github.com/apache/beam/blob/master/sdks/go/test/integration/integration.go#L166

Why do we skip the trigger tests in prism?

beam/sdks/go/test/integration/integration.go

Line 150 in 7c88161

"TestTrigger.*",

It was easier/preferred to add them to the prism test suite (in the no longer correctly named unimplemented_test.go) than to unblock this. I think we can probably remove this skip soon, after your work here is done.

Something something re-org.

sdks/go/pkg/beam/runners/prism/internal/engine/elementmanager.go

damccorm · 2025-10-03T15:08:56Z

FYI this PR is causing ML tests to start failing - https://github.com/apache/beam/actions/workflows/beam_PreCommit_Python_ML.yml?query=branch%3Amaster

The test is wrong, and I have #36377 to fix it

Handle after-processing-time trigger with processing-time timer.

3b1db11

github-actions bot added go runners prism labels Sep 30, 2025

shunping commented Sep 30, 2025

View reviewed changes

Consolidate the buildProcessingTimeBundle for stateful and aggregate …

b3cd953

…kinds into one code path.

shunping force-pushed the prism-after-processing-time-2 branch from fe01a3d to b3cd953 Compare October 1, 2025 02:22

shunping added 4 commits October 1, 2025 00:23

Set correct pane info.

d623c6f

Save panes when handling after-processing-time triggers

c6c0ee5

Add comments

41f6fdf

Change existing after processing time trigger test and add one more.

2179b24

shunping marked this pull request as ready for review October 1, 2025 18:50

shunping requested a review from lostluck October 1, 2025 18:50

shunping self-assigned this Oct 1, 2025

github-actions bot added the Next Action: Reviewers label Oct 1, 2025

lostluck reviewed Oct 1, 2025

View reviewed changes

sdks/go/test/integration/primitives/windowinto.go Outdated Show resolved Hide resolved

lostluck approved these changes Oct 1, 2025

View reviewed changes

shunping added 2 commits October 2, 2025 00:31

Change the bundleReady criterion for ordinary stage so it does not de…

f869d53

…pend on watermark.

Remove advance watermark steps in tests.

d6ccb90

shunping force-pushed the prism-after-processing-time-2 branch from 8c0e5ca to d6ccb90 Compare October 2, 2025 04:55

shunping commented Oct 2, 2025

View reviewed changes

Fix failed vr tests in java related to statelss dofn with side input

28b2caf

shunping commented Oct 2, 2025

View reviewed changes

github-actions bot added java and removed java labels Oct 2, 2025

shunping force-pushed the prism-after-processing-time-2 branch from 923dea8 to 28b2caf Compare October 2, 2025 15:40

shunping added 2 commits October 2, 2025 11:40

Sickbay a test.

1059e0f

Add comments to explain the sickbay test.

4179266

lostluck reviewed Oct 2, 2025

View reviewed changes

runners/prism/java/build.gradle Show resolved Hide resolved

shunping mentioned this pull request Oct 2, 2025

[Feature Request]: Support correct behavior of RequiresTimeSortedInput with AllowedLateness #36362

Open

17 tasks

lostluck approved these changes Oct 2, 2025

View reviewed changes

shunping added 2 commits October 2, 2025 17:37

Address reviewer feedback.

0d6f0a9

Add test filter for flink.

a3b986e

shunping merged commit 3534960 into apache:master Oct 2, 2025
28 of 29 checks passed

damccorm mentioned this pull request Oct 3, 2025

Make ml tests portable #36377

Merged

3 tasks

shunping mentioned this pull request Oct 3, 2025

[Prism] Support AfterSynchronizedProcessingTime and enable java processing-time trigger tests #36379

Merged

[Prism] Support AfterProcessingTime triggers - part 2 #36333

[Prism] Support AfterProcessingTime triggers - part 2 #36333

Uh oh!

Conversation

shunping commented Sep 30, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

github-actions bot commented Oct 1, 2025

Uh oh!

Uh oh!

lostluck left a comment

Choose a reason for hiding this comment

Uh oh!

shunping commented Oct 1, 2025

Uh oh!

lostluck commented Oct 1, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

shunping commented Oct 1, 2025

Uh oh!

lostluck commented Oct 2, 2025

Uh oh!

shunping commented Oct 2, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

shunping Oct 2, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

codecov bot commented Oct 2, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

Uh oh!

lostluck left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

damccorm commented Oct 3, 2025

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

shunping commented Sep 30, 2025 •

edited

Loading

lostluck commented Oct 1, 2025 •

edited

Loading

shunping commented Oct 2, 2025 •

edited

Loading

shunping Oct 2, 2025 •

edited

Loading

codecov bot commented Oct 2, 2025 •

edited

Loading