[Bug]: PeriodicImpulse and MatchContinuously default start_timestamp is the module import time #40564

What happened?

PeriodicImpulse and MatchContinuously use Timestamp.now() as the default value of start_timestamp:

https://github.com/apache/beam/blob/d9e99ad211f/sdks/python/apache_beam/transforms/periodicsequence.py#L295
https://github.com/apache/beam/blob/d9e99ad211f/sdks/python/apache_beam/io/fileio.py#L395

Python evaluates a default argument once, when the function is defined, so the default start is the time apache_beam was imported, not the time the transform is created. If the transform is created some time after the import (a long-running launcher process, a notebook kernel, slow pipeline setup), the impulse starts in the past and ImpulseSeqGenDoFn emits a burst of catch-up elements with timestamps from before the pipeline was built.

import time
import apache_beam as beam
from apache_beam.transforms.periodicsequence import PeriodicImpulse
from apache_beam.utils.timestamp import Timestamp

time.sleep(10)  # e.g. an idle notebook kernel or a long-running launcher

built_at = Timestamp.now()
with beam.Pipeline() as p:
  _ = (
      p
      | PeriodicImpulse(stop_timestamp=built_at + 3, fire_interval=1)
      | beam.Map(lambda ts: print('element', ts, 'seconds before construction:',
                                  round(float(built_at - ts), 1))))

Output (14 elements, 10 of them before the pipeline was built):

element 1791923872.515695 seconds before construction: 10.0
element 1791923873.515695 seconds before construction: 9.0
...
element 1791923885.515695 seconds before construction: -3.0

Expected: the impulse starts when the transform is created, so 3 elements (built_at, built_at + 1, built_at + 2).

The gap grows with the age of the process. A service that has been running for a day and builds a PeriodicImpulse(fire_interval=60) pipeline emits about 1,440 elements as soon as it starts. With apply_windowing=True, each of those elements also lands in a window that closed before the pipeline was built.

MatchContinuously has the same default, and ReadAllFromTextContinuously / ReadAllFromAvroContinuously use it when start_timestamp is not given.

Reproduced on master at d9e99ad (Python 3.12, DirectRunner). The default has been there since PeriodicImpulse was added.

Issue Priority

Priority: 2 (default / most bugs should be filed as P2)

Issue Components

• Component: Python SDK
• Component: Java SDK
• Component: Go SDK
• Component: Typescript SDK
• Component: IO connector
• Component: Beam YAML
• Component: Beam examples
• Component: Beam playground
• Component: Beam katas
• Component: Website
• Component: Infrastructure
• Component: Spark Runner
• Component: Flink Runner
• Component: Samza Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner