[darshan-sj]

[Java SDK - FileIO] Add withEvictWritersWhenFull option to WriteFiles

Summary & Motivation

When WriteFiles is configured with runner-determined sharding (WriteUnshardedBundlesToTempFiles), each bundle caps the number of concurrently open writers at maxNumWritersPerBundle (DEFAULT_MAX_NUM_WRITERS_PER_BUNDLE = 20) to bound worker memory usage (for example, each GCS writer allocates up to a 64 MB upload buffer by default).

Currently, when a bundle encounters more distinct writer keys (window, pane, destination) than maxNumWritersPerBundle, WriteFiles only offers two choices:

1. Default (Spill to Shuffle): Elements belonging to any destination beyond the first maxNumWritersPerBundle writers in the bundle are emitted to unwrittenRecordsTag and routed through an extra GroupByKey shuffle stage (GroupUnwritten -> WriteUnwritten -> DropShardNum -> Flatten).
2. withNoSpilling(): Eliminates the GroupUnwritten shuffle stage by setting maxNumWritersPerBundle = -1, which allows an unbounded number of concurrent writers per bundle and risks worker OOMs when writing to many dynamic destinations.

In pipelines where the input PCollection within a bundle is already grouped or ordered by writer key (e.g., upstream GroupByKey / Reshuffle by destination or table), a writer for an earlier destination is completely finished once the bundle moves on to the next destination. Keeping those 20 finished writers open until @FinishBundle forces all subsequent destinations in the bundle to spill over a second GroupByKey shuffle while holding 20 idle writer buffers in memory.

This PR introduces WriteFiles.withEvictWritersWhenFull() / WriteFiles.withEvictWritersWhenFull(boolean). When enabled, reaching maxNumWritersPerBundle evicts the oldest open writer in the bundle in FIFO order (flushing and closing its temporary file and releasing its buffer) and opens a new writer inline, completely avoiding the GroupUnwritten shuffle spill stage while strictly bounding concurrent writer memory to maxNumWritersPerBundle.

---

Changes Made

1. New Builder API in WriteFiles:

• Added getEvictWritersWhenFull() (defaulting to false), withEvictWritersWhenFull(), and withEvictWritersWhenFull(boolean evictWritersWhenFull).
• Added a <p><b>Warning:</b> in the Javadocs explaining that this option should only be used when the input PCollection elements within a bundle are grouped or ordered by writer keys (destination/window/pane), and warning that randomly ordered inputs across more destinations than maxNumWritersPerBundle will repeatedly close and reopen writers, producing many small files.

2. Omit Shuffle Spill Stage in WriteUnshardedBundlesToTempFiles.expand:

• Updated the expansion check to if (getMaxNumWritersPerBundle() < 0 || getEvictWritersWhenFull()) so that when writer eviction is enabled, the GroupUnwritten (GroupByKey), WriteUnwritten, DropShardNum, and Flatten transforms are omitted from the pipeline graph.

3. Changed the WriteUnshardedBundlesToTempFiles - removed the validation on maxNumWritersPerBundle and allowed it to be set to more than 20 (DEFAULT value).

---

Testing (WriteFilesTest)

• Added unit tests covering all the lines and branches changed.

---

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

• 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

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.

[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

[bharadwaj-aditya]

LGTM

[damccorm]

Thanks - rerunning a failed precommit (I think it was just a flake) before merging

[github-actions]

Assigning reviewers:

R: @Abacn 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).