fix leader board trigger

.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": 65
 }

sdks/python/apache_beam/examples/complete/game/leader_board.py

@@ -212,18 +212,17 @@ 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
         # processing of late data.
         | '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),
+                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.
         | '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 seconds.
         | 'LeaderboardUserGlobalWindows' >> beam.WindowInto(
             beam.window.GlobalWindows(),
-            trigger=trigger.Repeatedly(trigger.AfterCount(10)),
-            accumulation_mode=trigger.AccumulationMode.ACCUMULATING)
+            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.
         | 'ExtractAndSumScore' >> ExtractAndSumScore('user'))
 

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(

sdks/python/apache_beam/runners/portability/fn_api_runner/execution.py

@@ -188,14 +188,16 @@ 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
     self._table: collections.defaultdict[bytes,
                                          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.

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