[Java] Bound the Watch deduplication state with a timestamp cursor

[Eliaaazzz]

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) 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.

See the Contributor Guide for more tips on how to make review process smoother.

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 57.13%. Comparing base (a0e2714) to head (5c603c7).

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #39746      +/-   ##
============================================
+ Coverage     54.97%   57.13%   +2.16%     
- Complexity     1722     3642    +1920     
============================================
  Files          1071     1191     +120     
  Lines        170836   192731   +21895     
  Branches       1262     3816    +2554     
============================================
+ Hits          93910   110120   +16210     
- Misses        74656    79046    +4390     
- Partials       2270     3565    +1295     

Flag	Coverage Δ
java	72.13% <ø> (+4.45%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[github-actions]

Assigning reviewers:

R: @chamikaramj for label java.

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).