raams95 opened a new issue, #40564: URL: https://github.com/apache/beam/issues/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. ```python 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 d9e99ad211f (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 - [x] 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 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
