Eliaaazzz opened a new pull request, #39746:
URL: https://github.com/apache/beam/pull/39746
Bounds the deduplication state of the `Watch` transform by event time.
Addresses the garbage collection half of #18459, and the TODO left on
`PollingGrowthState.getCompleted`:
> Timestamps are preserved to allow garbage-collecting this state in the
future, e.g. dropping elements from "completed" and from
`computeNeverSeenBeforeResults()` if their timestamp is more than X behind the
watermark. As of writing, we don't do this [...] TODO: do this.
### What changes
`Watch` remembers the key of every output it has emitted, so the restriction
of an input that is watched indefinitely grows without bound.
`Growth.withTimestampCursor()` adds a cursor to `PollingGrowthState`, the
greatest timestamp emitted for that input. A key is retired from `completed`
once the cursor has moved more than the allowed lateness past it, so the set
holds a trailing window. Deduplication still goes by output key, and a key at
the retention floor is kept, so an output that arrives later at the same
timestamp is still deduplicated rather than emitted twice.
An output whose timestamp is below the floor is taken as already seen and is
dropped, since the key that would prove it seen has been retired. That suits a
`PollFn` whose outputs arrive in roughly non-decreasing timestamp order, such
as one that lists files by last modified time. `withTimestampCursor(Duration
allowedLateness)` widens the window for a source that reports outputs further
out of order.
The option is off by default and the transform behaves exactly as before
without it.
### Compatibility
A restriction without a cursor encodes under the existing tag and keeps the
pre-cursor bytes, so an in-flight pipeline can be updated onto this version. A
restriction written before the cursor existed also retires correctly on its
first claim, because `completed` has always recorded each key's own output
timestamp rather than a poll time.
### Known tradeoff
An output that a `PollFn` reports again with a later timestamp after its key
was retired is emitted a second time. This is documented on the option, and is
the reason the option is opt-in.
### Testing
`WatchTest`, 28 tests, all passing:
- `:sdks:java:core:test --tests "org.apache.beam.sdk.transforms.WatchTest"`,
20 tests
- `:runners:direct-java:needsRunnerTests --tests
"org.apache.beam.sdk.transforms.WatchTest"`, 8 tests
New coverage: cursor state coder round trip and pre-cursor byte format, key
retirement at the cursor, retention widened by allowed lateness, claim rejected
below the retention floor, outputs below the floor dropped while an output at
the floor is emitted, end to end exactly once delivery with the cursor enabled,
and rejection of a negative allowed lateness.
Each new test was checked against a mutated implementation to confirm it
fails when the behaviour it covers is removed:
| Mutation | Failing test |
| --- | --- |
| Drop outputs at the floor rather than below it |
`testPollingGrowthTrackerDropsOutputsBehindCursor` |
| Skip retirement in `trySplit` |
`testPollingGrowthTrackerRetiresCompletedBehindCursor`,
`testPollingGrowthTrackerAllowedLatenessRetainsCompleted` |
| Remove the retention floor check in `tryClaim` |
`testPollingGrowthTrackerRejectsClaimBehindCursor` |
| Remove the retention floor filter entirely |
`testMultiplePollsWithTimestampCursor` |
`checkstyleMain`, `checkstyleTest`, `spotlessJavaCheck`, `javadoc` and
`spotbugsMain` are clean on `:sdks:java:core`.
### Relationship to the Python SDK
The same mechanism is under review for the Python `Watch` transform in
#39461. The Java API folds the two Python knobs, `timestamp_cursor` and
`allowed_lateness`, into `withTimestampCursor()` and
`withTimestampCursor(Duration)`, so an allowed lateness cannot be set and then
silently ignored while the cursor is off. Happy to match the Python surface
exactly instead if you would prefer the two SDKs to read the same.
------------------------
Thank you for your contribution! Follow this checklist to help us
incorporate your contribution quickly and easily:
- [ ] [**Choose
reviewer(s)**](https://beam.apache.org/contribute/#make-your-change) and
mention them in a comment (`R: @username`).
- [ ] Mention the appropriate issue in your description (for example:
`addresses #123`), if applicable. This will automatically add a link to the
pull request in the issue. If you would like the issue to automatically close
on merging the pull request, comment `fixes #<ISSUE NUMBER>` instead.
- [ ] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual
Contributor License Agreement](https://cla.apache.org/).
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).
To check the build health, please visit
[https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md](https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md)
--
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]
