pass the runner clock into batch triggers

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