From 09fbdf9acdfd427e328718b554b459b4df6127ef Mon Sep 17 00:00:00 2001 From: aibrahiim Date: Thu, 1 Oct 2026 13:15:43 +0300 Subject: [PATCH 1/6] fix leader board trigger --- .../examples/complete/game/leader_board.py | 19 ++++++++----------- 1 file changed, 8 insertions(+), 11 deletions(-) diff --git a/sdks/python/apache_beam/examples/complete/game/leader_board.py b/sdks/python/apache_beam/examples/complete/game/leader_board.py index 96cde175c409..da146e1f20f8 100644 --- a/sdks/python/apache_beam/examples/complete/game/leader_board.py +++ b/sdks/python/apache_beam/examples/complete/game/leader_board.py @@ -212,9 +212,6 @@ def __init__(self, team_window_duration, allowed_lateness): self.allowed_lateness_seconds = allowed_lateness * 60 def expand(self, pcoll): - # NOTE: the behavior does not exactly match the Java example - # TODO: allowed_lateness not implemented yet in FixedWindows - # TODO: AfterProcessingTime not implemented yet, replace AfterCount return ( pcoll # We will get early (speculative) results as well as cumulative @@ -222,8 +219,10 @@ def expand(self, pcoll): | 'LeaderboardTeamFixedWindows' >> beam.WindowInto( beam.window.FixedWindows(self.team_window_duration), trigger=trigger.AfterWatermark( - trigger.AfterCount(10), trigger.AfterCount(20)), - accumulation_mode=trigger.AccumulationMode.ACCUMULATING) + early=trigger.AfterProcessingTime(5 * 60), + late=trigger.AfterProcessingTime(10 * 60)), + accumulation_mode=trigger.AccumulationMode.ACCUMULATING, + allowed_lateness=self.allowed_lateness_seconds) # Extract and sum teamname/score pairs from the event data. | 'ExtractAndSumScore' >> ExtractAndSumScore('team')) @@ -242,16 +241,14 @@ def __init__(self, allowed_lateness): self.allowed_lateness_seconds = allowed_lateness * 60 def expand(self, pcoll): - # NOTE: the behavior does not exactly match the Java example - # TODO: allowed_lateness not implemented yet in FixedWindows - # TODO: AfterProcessingTime not implemented yet, replace AfterCount return ( pcoll - # Get periodic results every ten events. + # Get periodic results every ten minutes. | 'LeaderboardUserGlobalWindows' >> beam.WindowInto( beam.window.GlobalWindows(), - trigger=trigger.Repeatedly(trigger.AfterCount(10)), - accumulation_mode=trigger.AccumulationMode.ACCUMULATING) + trigger=trigger.Repeatedly(trigger.AfterProcessingTime(10 * 60)), + accumulation_mode=trigger.AccumulationMode.ACCUMULATING, + allowed_lateness=self.allowed_lateness_seconds) # Extract and sum username/score pairs from the event data. | 'ExtractAndSumScore' >> ExtractAndSumScore('user')) From d413595e71758fd0f30bbb419fc568ac3677cbb7 Mon Sep 17 00:00:00 2001 From: aibrahiim Date: Thu, 1 Oct 2026 13:18:02 +0300 Subject: [PATCH 2/6] trigger workflow --- .github/trigger_files/beam_PostCommit_Python.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/trigger_files/beam_PostCommit_Python.json b/.github/trigger_files/beam_PostCommit_Python.json index e11d2260913f..a4517c89e4ee 100644 --- a/.github/trigger_files/beam_PostCommit_Python.json +++ b/.github/trigger_files/beam_PostCommit_Python.json @@ -1,5 +1,5 @@ { "comment": "Modify this file in a trivial way to cause this test suite to run.", "pr": "40241", - "modification": 63 + "modification": 64 } From da910f3bb099203ea95b8fc9ab7cd6f466103759 Mon Sep 17 00:00:00 2001 From: aibrahiim Date: Thu, 1 Oct 2026 14:20:26 +0300 Subject: [PATCH 3/6] fix leader board triggers --- .../examples/complete/game/leader_board_test.py | 4 +++- .../portability/fn_api_runner/execution.py | 17 ++++++++++------- sdks/python/apache_beam/transforms/trigger.py | 15 ++++++++++++--- 3 files changed, 25 insertions(+), 11 deletions(-) diff --git a/sdks/python/apache_beam/examples/complete/game/leader_board_test.py b/sdks/python/apache_beam/examples/complete/game/leader_board_test.py index 1c1cd6548923..34157ed054c6 100644 --- a/sdks/python/apache_beam/examples/complete/game/leader_board_test.py +++ b/sdks/python/apache_beam/examples/complete/game/leader_board_test.py @@ -49,7 +49,9 @@ def create_data(self, p): beam.window.TimestampedValue(elem, elem['timestamp']))) def test_leader_board_teams(self): - with TestPipeline() as p: + # Prism fires the early processing time pane and the on time pane. + test_options = PipelineOptions(flags=['--allow_unsafe_triggers']) + with TestPipeline(options=test_options) as p: result = ( self.create_data(p) | leader_board.CalculateTeamScores( diff --git a/sdks/python/apache_beam/runners/portability/fn_api_runner/execution.py b/sdks/python/apache_beam/runners/portability/fn_api_runner/execution.py index fe2421f8ee8a..be1da6af401b 100644 --- a/sdks/python/apache_beam/runners/portability/fn_api_runner/execution.py +++ b/sdks/python/apache_beam/runners/portability/fn_api_runner/execution.py @@ -188,7 +188,8 @@ def __init__( self, pre_grouped_coder: coders.Coder, post_grouped_coder: coders.Coder, - windowing: core.Windowing) -> None: + windowing: core.Windowing, + clock: Optional[Union[TestClock, RealClock]] = None) -> None: self._key_coder = pre_grouped_coder.key_coder() self._pre_grouped_coder = pre_grouped_coder self._post_grouped_coder = post_grouped_coder @@ -196,6 +197,7 @@ def __init__( list[Any]] = collections.defaultdict( list) self._windowing = windowing + self._clock = clock self._grouped_output: Optional[list[list[bytes]]] = None def copy(self) -> 'GroupingBuffer': @@ -250,11 +252,9 @@ def partition(self, n: int) -> list[list[bytes]]: globally_window((key, values)) ] else: - # TODO(pabloem, BEAM-7514): Trigger driver needs access to the clock - # note that this only comes through if windowing is default - but what - # about having multiple firings on the global window. - # May need to revise. - trigger_driver = trigger.create_trigger_driver(self._windowing, True) + # Processing-time triggers read the clock when an element arrives. + trigger_driver = trigger.create_trigger_driver( + self._windowing, True, clock=self._clock) windowed_key_values = trigger_driver.process_entire_key coder_impl = self._post_grouped_coder.get_impl() key_coder_impl = self._key_coder.get_impl() @@ -1148,7 +1148,10 @@ def get_buffer( self.execution_context.pipeline_components. pcollections[input_pcoll].windowing_strategy_id]]) self.execution_context.pcoll_buffers[buffer_id] = GroupingBuffer( - pre_gbk_coder, post_gbk_coder, windowing_strategy) + pre_gbk_coder, + post_gbk_coder, + windowing_strategy, + clock=self.execution_context.clock) else: # These should be the only two identifiers we produce for now, # but special side input writes may go here. diff --git a/sdks/python/apache_beam/transforms/trigger.py b/sdks/python/apache_beam/transforms/trigger.py index cc9922dd158f..c1ac0275aee2 100644 --- a/sdks/python/apache_beam/transforms/trigger.py +++ b/sdks/python/apache_beam/transforms/trigger.py @@ -1277,7 +1277,12 @@ def process_entire_key(self, key, windowed_values): MIN_TIMESTAMP): yield wvalue.with_value((key, wvalue.value)) while state.timers: - fired = state.get_and_clear_timers() + # Real-time timers that are still in the future must stay queued. + # Firing them here would ignore AfterProcessingTime's delay. + processing_time = None if self.clock is None else self.clock.time() + fired = state.get_and_clear_timers(processing_time=processing_time) + if not fired: + break for timer_window, (name, time_domain, fire_time, _) in fired: for wvalue in self.process_timer(timer_window, name, @@ -1660,6 +1665,9 @@ def get_timers( for (name, time_domain, dynamic_timer_tag), timestamp in list( timers.items()): if time_domain == TimeDomain.REAL_TIME: + if processing_time is None: + has_realtime_timer = True + continue time_marker = processing_time has_realtime_timer = True elif time_domain == TimeDomain.WATERMARK: @@ -1677,8 +1685,9 @@ def get_timers( del self.timers[window] return expired, has_realtime_timer - def get_and_clear_timers(self, watermark=MAX_TIMESTAMP): - return self.get_timers(clear=True, watermark=watermark)[0] + def get_and_clear_timers(self, watermark=MAX_TIMESTAMP, processing_time=None): + return self.get_timers( + clear=True, watermark=watermark, processing_time=processing_time)[0] def get_earliest_hold(self): earliest_hold = MAX_TIMESTAMP From 05725753096accc5085166b292e9ac0a2af522ca Mon Sep 17 00:00:00 2001 From: aibrahiim Date: Tue, 6 Oct 2026 19:02:26 +0300 Subject: [PATCH 4/6] shorten leader board team trigger delays --- .../python/apache_beam/examples/complete/game/leader_board.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/sdks/python/apache_beam/examples/complete/game/leader_board.py b/sdks/python/apache_beam/examples/complete/game/leader_board.py index da146e1f20f8..98c128b5585a 100644 --- a/sdks/python/apache_beam/examples/complete/game/leader_board.py +++ b/sdks/python/apache_beam/examples/complete/game/leader_board.py @@ -219,8 +219,8 @@ def expand(self, pcoll): | 'LeaderboardTeamFixedWindows' >> beam.WindowInto( beam.window.FixedWindows(self.team_window_duration), trigger=trigger.AfterWatermark( - early=trigger.AfterProcessingTime(5 * 60), - late=trigger.AfterProcessingTime(10 * 60)), + early=trigger.AfterProcessingTime(5), + late=trigger.AfterProcessingTime(30)), accumulation_mode=trigger.AccumulationMode.ACCUMULATING, allowed_lateness=self.allowed_lateness_seconds) # Extract and sum teamname/score pairs from the event data. From c275c97b2e0b89b7e29e1c9567d7f8fcb0d74f10 Mon Sep 17 00:00:00 2001 From: aibrahiim Date: Wed, 7 Oct 2026 10:28:37 +0300 Subject: [PATCH 5/6] trigger workflow --- .github/trigger_files/beam_PostCommit_Python.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/trigger_files/beam_PostCommit_Python.json b/.github/trigger_files/beam_PostCommit_Python.json index a4517c89e4ee..b6d790320d0b 100644 --- a/.github/trigger_files/beam_PostCommit_Python.json +++ b/.github/trigger_files/beam_PostCommit_Python.json @@ -1,5 +1,5 @@ { "comment": "Modify this file in a trivial way to cause this test suite to run.", "pr": "40241", - "modification": 64 + "modification": 65 } From 76f6e6d7c0c1b2ce582c73d068ad9260cf15c5f3 Mon Sep 17 00:00:00 2001 From: aibrahiim Date: Wed, 7 Oct 2026 22:27:23 +0300 Subject: [PATCH 6/6] fire leader board user scores every 10 seconds --- .../python/apache_beam/examples/complete/game/leader_board.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/sdks/python/apache_beam/examples/complete/game/leader_board.py b/sdks/python/apache_beam/examples/complete/game/leader_board.py index 98c128b5585a..e2d52698760e 100644 --- a/sdks/python/apache_beam/examples/complete/game/leader_board.py +++ b/sdks/python/apache_beam/examples/complete/game/leader_board.py @@ -243,10 +243,10 @@ def __init__(self, allowed_lateness): def expand(self, pcoll): return ( pcoll - # Get periodic results every ten minutes. + # Get periodic results every ten seconds. | 'LeaderboardUserGlobalWindows' >> beam.WindowInto( beam.window.GlobalWindows(), - trigger=trigger.Repeatedly(trigger.AfterProcessingTime(10 * 60)), + trigger=trigger.Repeatedly(trigger.AfterProcessingTime(10)), accumulation_mode=trigger.AccumulationMode.ACCUMULATING, allowed_lateness=self.allowed_lateness_seconds) # Extract and sum username/score pairs from the event data.