[knockknock10]

This PR addresses #34419.

What problem does this solve?

When two DoFns are fused, an exception thrown by a downstream DoFn can propagate through the upstream DoFn's output() call. If the upstream DoFn catches that exception, the bundle may otherwise complete successfully even though downstream processing failed.

This change introduces configurable handling for exceptions thrown across fused DoFn boundaries in the Java Fn harness.

Implementation

Adds the fused_dofn_exception_handling experiment with three modes:

• legacy: Preserves the existing exception propagation behavior.
• warn: Preserves exception propagation, but logs a warning if a downstream exception was swallowed and the bundle would otherwise complete successfully.
• fail: Records downstream exceptions, prevents subsequent outputs to the affected downstream transform, and fails the bundle if it would otherwise complete successfully.

Exceptions are recorded at the consumer boundary. The selected policy is applied during bundle processing and completion, and the recorded exception state is reset when the bundle processor is reset.

Tests

Adds registry-level and bundle-level regression tests covering:

• Recording downstream consumer exceptions.
• Preventing subsequent outputs to an affected downstream transform in fail mode.
• Failing bundle completion when a downstream exception is swallowed.
• Warning when a swallowed exception does not otherwise fail the bundle.
• Preserving legacy behavior.
• Resetting exception state between bundles.
• Continuing delivery to healthy sibling consumers.
• Recording and surfacing Error subclasses.
• Preventing @FinishBundle execution after a swallowed processing failure.

Validation

Executed locally using Temurin JDK 21:

./gradlew :sdks:java:harness:test \
  --tests org.apache.beam.fn.harness.control.ProcessBundleHandlerTest \
  --tests org.apache.beam.fn.harness.data.PCollectionConsumerRegistryTest \
  --no-build-cache

Results:

• ProcessBundleHandlerTest and PCollectionConsumerRegistryTest:
• Build successful.

The branch was rebased onto Apache Beam master before validation.

[Copilot]

Copilot was unable to review this pull request because the user who requested the review has insufficient remaining quota. Try requesting a review at a lower effort level.

[github-actions]

📋 Review Summary

This PR adds the fused_dofn_exception_handling experiment (legacy, warn, fail) to detect and handle exceptions thrown by downstream consumers that are caught and swallowed by upstream fused DoFns. While the core tracking in PCollectionConsumerRegistry and ProcessBundleHandler is well-structured, there are a few correctness issues regarding bundle failure timing (running @FinishBundle after a downstream failure), sibling consumer blocking in multiplexing receivers, and unchecked Error handling.

🔍 General Feedback

• Bundle lifecycle on failure: In ProcessBundleHandler, failBundleIfDownstreamFailed() is currently invoked after finishFunctionRegistry runs. In fail mode, if a downstream transform failed during element processing, running @FinishBundle on the broken transform (which can flush invalid partial state or throw a secondary masking exception) should be avoided.
• Multiplexing consumer isolation: In MultiplexingMetricTrackingFnDataReceiver, calling failFastIfDownstreamFailed inside the consumer loop aborts the entire loop on subsequent elements, preventing healthy sibling consumers of the same PCollection from receiving elements before the exception is rethrown to the upstream DoFn.
• Catching Throwable:User code can throw Error subclasses (e.g., AssertionError) and catch Throwable or Error upstream; tracking Exception only misses these cases.

---

💡 Tip: Comment /gemini review on this pull request to trigger a new Gemini code review.

[github-actions]

Assigning reviewers:

R: @chamikaramj for label java.
Backup reviewer: ahmedabu98 (if the main reviewer is unresponsive, you can escalate to them by tagging their username)

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