raams95 opened a new pull request, #40565:
URL: https://github.com/apache/beam/pull/40565
Fixes #40564
`PeriodicImpulse` and `MatchContinuously` declared
`start_timestamp=Timestamp.now()` as a default argument. Python evaluates that
once, when the module is imported, so the default start was the import time of
`apache_beam`, not the time the transform was created. When the transform is
created later (a long-running launcher process, a notebook kernel, slow
pipeline setup), `ImpulseSeqGenDoFn` starts in the past and emits a burst of
catch-up elements with timestamps from before the pipeline existed:
```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)
built_at = Timestamp.now()
with beam.Pipeline() as p:
_ = (
p
| PeriodicImpulse(stop_timestamp=built_at + 3, fire_interval=1)
| beam.Map(print))
```
On master this prints 14 elements, starting 10 seconds before `built_at`.
With this change it prints the 3 expected ones, `built_at`, `built_at + 1` and
`built_at + 2`.
The default is now `None`, and `__init__` resolves it to `Timestamp.now()`.
This matches what #35569 assumed when it added `rebase`: that a
`Timestamp.now()` start is "the timestamp when the PTransform is expanded (i.e.
the pipeline's construction time)". Explicit `start_timestamp` values and the
`rebase` modes behave as before. `MatchContinuously` had the same default, and
`ReadAllFromTextContinuously` / `ReadAllFromAvroContinuously` pass through to
it, so it gets the same fix.
Tests:
- `periodicsequence_test.test_default_start` read the default from
`inspect.signature`, so it tested the import-time value. It now creates the
transform with `Timestamp.now` patched to a fixed time and checks the pipeline
output starts there.
- `fileio_test.MatchContinuouslyTest.test_default_start_timestamp` does the
same for `MatchContinuously(has_deduplication=False)`.
Both fail before the change (the default start is the import time, which is
later than the given stop, so `initial_restriction` raises `AssertionError`)
and pass after it.
Testing:
- `pytest apache_beam/transforms/periodicsequence_test.py
apache_beam/io/fileio_test.py` (71 passed, 3 skipped)
- `pytest apache_beam/io/textio_test.py apache_beam/io/avroio_test.py -k
Continuously`
- `yapf --diff` and `ruff check` on the changed files
------------------------
Thank you for your contribution! Follow this checklist to help us
incorporate your contribution quickly and easily:
- [x] Mention the appropriate issue in your description (for example:
`addresses #123` or `fixes #456`).
- [x] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual
Contributor License Agreement](https://www.apache.org/licenses/icla.pdf). (Not
needed, small change.)
See the [Contributor Guide](https://beam.apache.org/contribute) for more
tips on [how to make review process
smoother](https://github.com/apache/beam/blob/master/CONTRIBUTING.md#make-the-reviewers-job-easier).
--
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]
