[goutamadwant]

Addresses the remaining Java pending-result scalability item in #18459.

Watch currently keeps every distinct unseen output from a poll in checkpointed pending restriction state. A very large poll can therefore create an oversized state value.

What changes

• Adds the opt-in Watch.Growth.withMaxPendingResults(int) API.
• For finite, repeatable incomplete polls, keeps at most the configured number of oldest unseen outputs as pending work in each round.
• Holds the watermark at or before the earliest omitted output and continues polling until omitted work can be recovered.
• Defers a configured termination condition while a poll is truncated, avoiding silent loss of omitted outputs.
• Preserves the existing first-output behavior for duplicate output keys.
• Never limits PollResult.complete(), because its contract promises that the polling function will not be called again.

Compatibility and scope

The option is disabled by default, so existing pipelines retain their current behavior. It does not change the restriction coder or persisted state format.

When enabled, the PollFn must repeat omitted outputs in subsequent polls and must eventually return a finite set that can drain below the limit. A function that continuously returns more distinct unseen outputs than the limit can keep polling indefinitely; the API documentation and runtime logging make this behavior explicit.

The option bounds checkpointed pending outputs. It does not bound the list constructed by the polling function, transient hashes used during selection, or completed keys sharing a retained timestamp. It can be combined with withTimestampCursor() to time-bound completed-key state when output timestamps advance past the retention floor.

This PR covers the Java SDK. Equivalent Python behavior can be tracked separately.

Validation

• :sdks:java:core:test --tests 'org.apache.beam.sdk.transforms.WatchTest' --no-build-cache --rerun-tasks - 30 tests passed
• :runners:direct-java:needsRunnerTests --tests 'org.apache.beam.sdk.transforms.WatchTest.testManyResultsWithPendingLimit' --no-build-cache --rerun-tasks - passed; recovered 300 outputs over three bounded rounds
• :sdks:java:core:checkstyleMain
• :sdks:java:core:checkstyleTest
• :sdks:java:core:spotlessJavaCheck
• :sdks:java:core:javadoc
• :sdks:java:core:spotbugsMain

---

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

• Mention the appropriate issue in your description.
• Update CHANGES.md with noteworthy changes.
• If this contribution is large, file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on making the review process smoother.

To check the build health, see BUILD_STATUS.md. See CI.md and the workflows README for GitHub Actions details and workflow triggers.

[github-actions]

Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment assign set of reviewers

[goutamadwant]

retest this please

[github-actions]

Assigning reviewers:

R: @chamikaramj for label java.

This pull request likely touches a core component ("core" label). Please review with scrutiny.

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

[github-actions]

Reminder, please take a look at this pr: @chamikaramj

[github-actions]

Assigning new set of reviewers because Pr has gone too long without review. If you would like to opt out of this review, comment assign to next reviewer:

R: @kennknowles for label java.

This pull request likely touches a core component ("core" label). Please review with scrutiny.

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)

[kennknowles]

This is complex code that I don't know very well, so review will take some time. There may be someone better to review, but I am not sure.

[kennknowles]

This should be @Nullable Integer or otherwise have a way of explicitly indicating it is unset.

[github-actions]

Reminder, please take a look at this pr: @kennknowles @chamikaramj

[derrickaw]

Hi @goutamadwant, friendly ping to resolve merge conflicts and address one comment so far. Thanks

[goutamadwant]

done @derrickaw
@kennknowles changed it to @Nullable Integer so null clears the limit. let me know if this looks good now! thanks