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

[ntopousis]

🤖 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:

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 FileResults 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:

beam/sdks/java/core/src/main/java/org/apache/beam/sdk/io/WriteFiles.java

Lines 534 to 577 in 7eef130

	private class GatherResults<ResultT>
	extends PTransform<PCollection<ResultT>, PCollection<List<ResultT>>> {
	private final Coder<ResultT> resultCoder;
	private GatherResults(Coder<ResultT> resultCoder) {
	this.resultCoder = resultCoder;
	}
	@Override
	public PCollection<List<ResultT>> expand(PCollection<ResultT> input) {
	if (getWindowedWrites()) {
	// Reshuffle the results to make them stable against retries.
	// Use a single void key to maximize size of bundles for finalization.
	return input
	.apply("Add void key", WithKeys.of((Void) null))
	.apply("Reshuffle", Reshuffle.of())
	.apply("Drop key", Values.create())
	.apply("Gather bundles", ParDo.of(new GatherBundlesPerWindowFn<>()))
	.setCoder(ListCoder.of(resultCoder))
	// Reshuffle one more time to stabilize the contents of the bundle lists to finalize.
	.apply(Reshuffle.viaRandomKey());
	} else {
	// Pass results via a side input rather than reshuffle, because we need to get an empty
	// iterable to finalize if there are no results.
	return input
	.getPipeline()
	.apply(
	"AsPossiblyEmptyList",
	Reify.viewInGlobalWindow(
	// Insert a reshuffle before taking the view to consolidate the (typically)
	// one-output-per-bundle writes.
	// This avoids producing a huge number of tiny files in the case that side
	// inputs are materialized to disk bundle-by-bundle.
	input.apply("Consolidate", Reshuffle.viaRandomKey()).apply(View.asIterable()),
	IterableCoder.of(resultCoder)))
	// View.asIterable() can be (significantly) cheaper than View.asList(), as it does not
	// create a backing indexable view, but we must return a list to maintain update
	// compatibility for consumers that are shared between this path and the streaming one.
	.apply(
	"IterableToList",
	MapElements.via(
	new SimpleFunction<Iterable<ResultT>, List<ResultT>>(
	x -> ImmutableList.copyOf(x)) {}))
	.setCoder(ListCoder.of(resultCoder));

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:

#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:

PCollection<String> 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 FileResults 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<List<FileResult<...>>> 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

• Component: Java SDK
• Component: IO connector
• Component: Google Cloud Dataflow Runner

[ntopousis]

.take-issue