[Bug]: Bounded Java WriteFiles finalization materializes all FileResults as a global side input

🤖 This report and proposed patch were prepared with an AI coding agent from a production investigation. Identifiers and application-specific details are intentionally omitted; the Dataflow job and support-case identifiers can be shared privately with the Google Dataflow team.

### What happened?

A bounded Java pipeline using Apache Beam 2.64.0 on Google Cloud Dataflow took multiple hours to finalize several otherwise-completed unwindowed `TextIO.write()` transforms using runner-determined sharding.

The distinguishing factor was not side-input byte size. It was the number of temporary-file metadata records entering `WriteFiles/GatherTempFileResults`.

| Output | `FileResult` count | Observed status |
|---|---:|---|
| A | 5,119 | completed |
| B | 23,633 | completed |
| C | 47,285 | completed |
| D | 89,677 | still finalizing |
| E | 97,106 | still finalizing |

One corresponding internal side-input materialization was approximately 18 MiB. The two largest finalizers were not deadlocked: they advanced near-linearly at roughly 200–225 `FileResult` records per minute, making finalization take many hours.

Google Cloud support's backend analysis reported that the job was spending substantial time reading and retrying GCS files. Repeated worker stacks during this phase included:

```text
GoogleCloudStorageReadChannel
-> IsmReaderImpl
-> IsmSideInputReader
-> IterableLikeCoder.registerByteSizeObserver
-> OutputObjectAndByteCounter
```

This was not an application-created `PCollectionView`. It was the internal side input introduced by bounded, unwindowed `WriteFiles`.

We have not proved whether the dominant defect is the SDK's use of a global side input, Dataflow's access pattern for a high-cardinality ISM iterable, or their interaction. The proposed SDK patch removes that interaction. It retains the existing one-call rename and cleanup protocol while making runner-determined positional shard assignment deterministic.

### Relevant SDK path

Both fixed sharding and bounded runner-determined sharding feed their `FileResult`s into `GatherTempFileResults`:

https://github.com/apache/beam/blob/v2.64.0/sdks/java/core/src/main/java/org/apache/beam/sdk/io/WriteFiles.java#L469-L490

For unwindowed writes, `GatherResults` reshuffles the records, creates `View.asIterable()`, reifies the view in the global window, and copies the iterable into a `List`:

https://github.com/apache/beam/blob/v2.64.0/sdks/java/core/src/main/java/org/apache/beam/sdk/io/WriteFiles.java#L527-L570

`FinalizeFn` then copies that list again before finalizing all destinations and moving the files:

https://github.com/apache/beam/blob/v2.64.0/sdks/java/core/src/main/java/org/apache/beam/sdk/io/WriteFiles.java#L1335-L1384

The same global side-input path remains on master as of `7eef1304b19c5a47c12354476c4319df5183908d`:

https://github.com/apache/beam/blob/7eef1304b19c5a47c12354476c4319df5183908d/sdks/java/core/src/main/java/org/apache/beam/sdk/io/WriteFiles.java#L534-L577

PR #26289 introduced `View.asIterable()` in place of `View.asList()` to reduce the cost of small writes, while preserving a downstream `ListCoder` for update compatibility:

https://github.com/apache/beam/pull/26289

This report concerns the high-cardinality failure mode that remains in that path.

Related prior art: #18629 describes the same high-cardinality ISM side-input pathology in BigQuery batch-load finalization. It concerns a different transform, but the finely sharded temporary-file metadata and `IsmReader` / coder-size stack are closely related.

### Reproduction shape

The production trigger can be approximated with a bounded pipeline that induces tens of thousands of runner bundles and then performs an unwindowed write without explicit sharding:

```java
PCollection records =
pipeline
.apply(GenerateSequence.from(0).to(NUM_RECORDS))
.apply(/* produce fixed-size records */)
.apply(Reshuffle.viaRandomKey());

records.apply(TextIO.write().to("gs://BUCKET/reproduction/output"));
// Intentionally no withNumShards() and no withWindowedWrites().
```

The exact `FileResult` count is runner-dependent, so the meaningful reproduction condition is the cardinality entering `GatherTempFileResults`, not only the number of input records.

### Expected behavior

Finalization necessarily performs work proportional to the number of temporary files, but an approximately 18 MiB metadata collection should not take hours to consume. The gather should use a normal shuffle or another bulk main-input path rather than a remote iterable side-input access pattern with very high per-element overhead.

### Proposed narrow SDK patch

PR: #39372

The patch replaces the bounded, unwindowed gather transport and adds deterministic ordering at the runner-determined shard-assignment boundary:

1. Gather `FileResult`s into bundle-sized lists.
2. Add one explicitly coded empty-list marker so empty writes still reach finalization.
3. Flatten and key those lists by `Void`.
4. Use `GroupByKey` as a main-input shuffle.
5. Flatten the grouped bundle lists back into the existing `PCollection>>` contract.
6. Use `TimestampCombiner.EARLIEST` internally to preserve the old minimum element timestamp, then restore the global-default windowing strategy.
7. Inside `FileBasedSink.WriteOperation.finalizeDestination`, sort unwindowed runner-determined results by unique temporary filename immediately before positional shard assignment.

The patch retains `ListCoder`, `FinalizeFn`'s defensive copy, and the existing single-call rename and cleanup protocol. It preserves:

- empty unwindowed writes producing at least one shard unless `skipIfEmpty` is enabled;
- missing empty-shard creation for fixed sharding;
- runner-determined shard count and final filename format;
- dynamic destinations and filename-policy side inputs;
- retry-safe rename options;
- output-filename visibility after finalization;
- the minimum output-filename element timestamp and the global-default public windowing strategy; and
- whole-temporary-directory cleanup after all renames.

The deliberate semantic change is that the previously arbitrary content-to-shard ordering now follows lexicographic temporary-filename order rather than materialization order. Fixed-shard assignment and the windowed/unbounded write path are unchanged.

The public windowing-strategy reset is important: without it, the internal `EARLIEST` combiner would leak through `getPerDestinationOutputFilenames()`, and a downstream aggregation applied directly to that collection could emit at `TIMESTAMP_MIN_VALUE` rather than at the end of the global window.

A structural regression test asserts that the unwindowed `GatherTempFileResults` subtree no longer contains `View.CreatePCollectionView`. A finalization unit test runs two permutations of the same `FileResult` set through the real shard-assignment path and verifies identical temp-file-to-final-shard mappings. Focused tests also preserve the output element timestamp and public windowing strategy. The full DirectRunner `WriteFilesTest` class exercises empty writes, fixed sharding, missing shards, dynamic destinations, filenames, and cleanup.

### Deliberate limitation

This is a narrow mitigation, not a complete finalization redesign. It still produces one O(N) list and uses one logical finalizer. Truly memory-bounded, parallel finalization would require a two-phase design because shard counts and final names must be established before parallel renames, missing empty shards must be generated exactly once, and the shared unwindowed temporary directory cannot be deleted until every rename chunk completes.

A Dataflow-side improvement to sequential ISM reads, range prefetch, or coder byte-size observation may also be appropriate to protect pipelines on released SDK versions. Those runner-side and chunked-finalization questions remain open, so PR #39372 addresses this issue rather than auto-closing it.

### Questions for maintainers

1. Is the global side-input gather intentional for any reason beyond ensuring an empty write still invokes finalization?
2. Is the empty-marker plus main-input `GroupByKey` approach, with deterministic ordering at the runner-determined shard-assignment boundary, acceptable as a narrow first fix?
3. Does the Dataflow team recognize this `IsmReaderImpl` / `IterableLikeCoder.registerByteSizeObserver` stack as a runner-side read-amplification pattern?
4. Should chunked/parallel finalization and a Dataflow-side mitigation be tracked as separate follow-ups?

### Issue Priority

Priority: 2 (default / most bugs should be filed as P2)

### Issue Components

- [x] Component: Java SDK
- [x] Component: IO connector
- [x] Component: Google Cloud Dataflow Runner