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