diff --git a/sdks/python/apache_beam/transforms/util.py b/sdks/python/apache_beam/transforms/util.py index 309aba5b4af2..369e8843be6b 100644 --- a/sdks/python/apache_beam/transforms/util.py +++ b/sdks/python/apache_beam/transforms/util.py @@ -64,6 +64,7 @@ from apache_beam.transforms.timeutil import TimeDomain from apache_beam.transforms.trigger import AccumulationMode from apache_beam.transforms.trigger import AfterCount +from apache_beam.transforms.trigger import Repeatedly from apache_beam.transforms.userstate import BagStateSpec from apache_beam.transforms.userstate import CombiningValueStateSpec from apache_beam.transforms.userstate import TimerSpec @@ -678,7 +679,7 @@ def restore_timestamps(element): # accept only standard coders. ungrouped._windowing = Windowing( window.GlobalWindows(), - triggerfn=AfterCount(1), + triggerfn=Repeatedly(AfterCount(1)), accumulation_mode=AccumulationMode.DISCARDING, timestamp_combiner=TimestampCombiner.OUTPUT_AT_EARLIEST) result = (