Merge pull request apache#11066 from angoenka/fix_reshuffle

[BEAM-9465] Fire repeatedly in reshuffle
jto · Mar 7, 2020 · 23165a6 · 23165a6
2 parents a8e60bc + 0a2cee9
commit 23165a6
Showing 1 changed file with 2 additions and 1 deletion.
diff --git 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 = (