Call now() once so start and end have exactly the same base timestamp. (#36574)

Author: shunping

sdks/python/apache_beam/transforms/periodicsequence.py

@@ -337,8 +337,7 @@ def expand(self, pbegin):
     if self.rebase == RebaseMode.REBASE_ALL:
       duration = Timestamp.of(self.stop_ts) - Timestamp.of(self.start_ts)
       impulse_element = pbegin | beam.Impulse() | beam.Map(
-          lambda _:
-          [Timestamp.now(), Timestamp.now() + duration, self.interval])
+          lambda _: [now := Timestamp.now(), now + duration, self.interval])
     elif self.rebase == RebaseMode.REBASE_START:
       impulse_element = pbegin | beam.Impulse() | beam.Map(
           lambda _: [Timestamp.now(), self.stop_ts, self.interval])