[Prism] Support injecting triggered bundle for a batch of elements.

[shunping]

Previously, Prism's streaming-first design checked trigger criteria for each element and triggered bundles immediately. This resulted in behavior that diverged from the Dataflow and FnApi runners in batch mode.

To address this, this PR introduces the ability to process and inject triggered bundles in batches, a behavior toggled by the --streaming pipeline option.

---

However, this change has surfaced issues with unsafe triggers, leading to two failing tests after re-enabling prism on pipelines with unsafe triggers.

• apache_beam.transforms.trigger_test.TriggerPipelineTest testMethod=test_after_count.

  • Error message:

  apache_beam.testing.util.BeamAssertException: Failed assert: [('A-5', {1, 2, 3, 4, 5}), ('B-4', {8, 9, 6, 7}), ('B-3', {16, 10, 15})] == [('B-4', {8, 9, 6, 7}), ('A-5', {1, 2, 3, 4, 5}), ('B-3', {16, 10, 15}), ('A-2', {10, 11})], 
  unexpected elements [('A-2', {10, 11})]
  

  • Assert:
    

    beam/sdks/python/apache_beam/transforms/trigger_test.py

    Lines 661 to 669 in da57e58

    	assert_that(
    	result,
    	equal_to(
    	list({
    	'A-5': {1, 2, 3, 4, 5},
    	# A-10, A-11 never emitted due to AfterCount(3) never firing.
    	'B-4': {6, 7, 8, 9},
    	'B-3': {10, 15, 16},
    	}.items())))

  • Running on Dataflow

• apache_beam/examples/complete/game/leader_board_test.py LeaderBoardTest.test_leader_board_users

  • Error message:

  apache_beam.testing.util.BeamAssertException: Failed assert: [] == [('user2_team2', 2), ('user3_team3', 8), ('user4_team3', 5), ('user1_team1', 50)], 
  unexpected elements [('user2_team2', 2), ('user3_team3', 8), ('user4_team3', 5), ('user1_team1', 50)]
  

  • Assert:
    

    beam/sdks/python/apache_beam/examples/complete/game/leader_board_test.py

    Lines 62 to 68 in da57e58

    	def test_leader_board_users(self):
    	test_options = PipelineOptions(flags=['--allow_unsafe_triggers'])
    	with TestPipeline(options=test_options) as p:
    	result = (
    	self.create_data(p)
    	| leader_board.CalculateUserScores(allowed_lateness=120))
    	assert_that(result, equal_to([]))

---

The two test failures are caused by differences in the window "closing behavior," which determines if a final pane is emitted when a window expires.

beam/model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/beam_runner_api.proto

Lines 1210 to 1220 in da57e58

	message ClosingBehavior {
	enum Enum {
	UNSPECIFIED = 0;
	// Emit output when a window expires, whether or not there has been
	// any new data since the last output.
	EMIT_ALWAYS = 1;
	// Only emit output when new data has arrives since the last output
	EMIT_IF_NONEMPTY = 2;
	}

The Python SDK defaults to EMIT_ALWAYS, with a TODO to support EMIT_IF_NONEMPTY.

beam/sdks/python/apache_beam/transforms/core.py

Lines 3781 to 3782 in da57e58

	# TODO(robertwb): Support EMIT_IF_NONEMPTY
	closing_behavior=beam_runner_api_pb2.ClosingBehavior.EMIT_ALWAYS,

This PR's changes try to align Prism's output with the Dataflow runner's behavior, and I believe the two failing tests will need to be re-benchmarked.

[codecov]

Codecov Report

❌ Patch coverage is 90.16393% with 6 lines in your changes missing coverage. Please review.
✅ Project coverage is 36.12%. Comparing base (bcc8961) to head (bc648d5).
⚠️ Report is 8 commits behind head on master.

Files with missing lines	Patch %	Lines
...am/runners/prism/internal/engine/elementmanager.go	94.33%	2 Missing and 1 partial ⚠️
sdks/go/pkg/beam/runners/prism/internal/execute.go	62.50%	2 Missing and 1 partial ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #36219      +/-   ##
============================================
+ Coverage     36.09%   36.12%   +0.02%     
  Complexity     1618     1618              
============================================
  Files          1057     1057              
  Lines        164358   164415      +57     
  Branches       1165     1165              
============================================
+ Hits          59331    59388      +57     
- Misses       102871   102874       +3     
+ Partials       2156     2153       -3     

Flag	Coverage Δ
go	28.28% <90.16%> (+0.05%)	⬆️

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.

[shunping]

The test apache_beam/examples/complete/game/leader_board_test.py::LeaderBoardTest::test_leader_board_users were run in two different workflows:

• In beam_PreCommit_Python_Examples, the test is successful (https://github.com/apache/beam/actions/runs/17879963633/job/50846364797?pr=36219):
  [gw4] [ 28%] PASSED apache_beam/examples/complete/game/leader_board_test.py::LeaderBoardTest::test_leader_board_users

• In Python Unit Tests, the same test failed (
  https://github.com/apache/beam/actions/runs/17879963379/workflow?pr=36219). By adding some debugging log, I found the runner was the fallback one, because no go is found.

ERROR    apache_beam.runners.direct.direct_runner:direct_runner.py:148 PrismRunnerSupportVisotr returnsTrue
ERROR    apache_beam.runners.direct.direct_runner:direct_runner.py:218 Running pipeline with PrismRunner.
ERROR    apache_beam.runners.direct.direct_runner:direct_runner.py:237 Exception with PrismRunner:
 [Errno 2] No such file or directory: 'go'

ERROR    apache_beam.runners.direct.direct_runner:direct_runner.py:238 Falling back to DirectRunner
ERROR    apache_beam.runners.direct.direct_runner:direct_runner.py:244 Running pipeline with FnApiRunner.

Looks like we need to install go in that workflow.

[shunping]

I am not able to re-enable the pipelines with unsafe triggers to run on prism.

• First, there was the missing go binary problem, which was resolved by Install go for python unit tests to use prism runner. #36221.
• However, I noticed there is a significant increase of runtime on Python Unit Test (Python Unit Tests is getting much slower since 2.67.0 #36222). So it was reverted at Revert "Install go for python unit tests to use prism runner." #36223.

When things are resolved, we can pick up the commit to re-enable and rebenchmark: bc648d5.

[github-actions]

Assigning reviewers:

R: @jrmccluskey for label python.
R: @lostluck 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).

[lostluck]

LGTM

[lostluck]

I'll note that Go does some magic when passing a string cast []byte as a map key inline, which avoids allocating. That's why this method took in a []byte for the key, instead of eagerly converting it.

[lostluck]

But this over here probably made the string(key) (when key is []byte) moot anyway.