Repository navigation
[Bug]: Inline exception propagation with fused DoFns can lead to unexpected behavior #34419
Description
What happened?
Consider a pipeline where two steps A and B are fused and A -> B.
Runners such as Dataflow or the Fnapi harness wire up output(...) calls from A directly to processElement(...) calls to B. If B throws an exception during processing this exception then is just propagated up the stack and inside of A. If A doesn't catch this exception, it is caught at the runner level and results in a failed bundle.
However by raising the exception to A possible issues arise:
- the exception may interrupt A processing unexpectedly leaving it in some invalid state if it had bad error handling. Example could be not decrementing some static semaphore held in a final block.
- be caught within A if A itself has a try catch block for error handling. However the user may not realize that they are catching and supressing errors from B with this handling and it could lead to data loss as the processing of the input to B didn't complete.
Similar issues can occur if B itself outputs to a sink (such as fnapi data output or dataflow commit) which encounters an encoding error.
It seems like it would be preferable to catch such errors between DoFn boundaries instead of propagating them directly. Once detected, the process of the bundle could be cleanly stopped and the exceptions could be surfaced as failures to process the bundle.
This issue has been observed within Java execution on Dataflow but appears to be a common problem for the Java SDK and likely other sdks in which exceptions are thrown.
Issue Priority
Priority: 2 (default / most bugs should be filed as P2)
Issue Components
- Component: Python SDK
- Component: Java SDK
- Component: Go SDK
- Component: Typescript SDK
- Component: IO connector
- Component: Beam YAML
- Component: Beam examples
- Component: Beam playground
- Component: Beam katas
- Component: Website
- Component: Infrastructure
- Component: Spark Runner
- Component: Flink Runner
- Component: Samza Runner
- Component: Twister2 Runner
- Component: Hazelcast Jet Runner
- Component: Google Cloud Dataflow Runner
Activity
scwhittle commentedon Mar 25, 2025
Adding as 3.0.0 release milestone as this is potentially breaking behavior because existing pipelines may work due to relying on inline exception propagation.
github-actions commentedon Sep 2, 2025
This issue has been marked as stale due to 150 days of inactivity. It will be closed in 30 days if no further activity occurs. If you think that’s incorrect or this issue still needs to be addressed, please simply write any comment. If closed, you can reopen the issue at any time. Thank you for your contributions.
scwhittle commentedon Sep 2, 2025
Still a feature request for 3.0, pinging to remove staleness
Abacn commentedon Sep 3, 2025
This Issue is marked as Beam 3 - Milestone 1: No Code and ML focus. Please help us burning down the open blocker tasks.
If this issue is solved, close it. If it is obsolete, close as not planned. If it is no longer necessary for Beam 3, move it out of the Milestone.
If the work is pending, please be reminded that Milestone 1 is targeted for completion by Dec 30, 2025
If this issue involves multiple milestones (e.g. remove a public API should first mark it as deprecated, scoped in Milestone 1; then finally remove it in Beam 3 release candidate, scoped in Milestone 3) and Milestone 1 portion has completed, consider move the Issue to Milestone 3.
scwhittle commentedon Sep 9, 2025
I think that perhaps we could introduce this before beam3 behind a pipeline option and then change the option default for beam 3.
Currently
outputmethod from a dofn is wired up to the next dofn process element method via PCollectionConsumerRegistry here. I think that we could just modify where it currently logs and rethrows here to additionally save the exception, not delegate future outputs and prevent the bundle from finishing successfully. If it has a way to truncate the input so the bundle fails faster that would also be good.Abacn commentedon Nov 21, 2025
This sounds a proposed breaking change ideally making the switch between Beam 2->3. Currently stay at milestone 1 to track progress of
We can move to milestone 3 when milestone 1 closes.
github-actions commentedon Apr 21, 2026
This issue has been marked as stale due to 150 days of inactivity. It will be closed in 30 days if no further activity occurs. If you think that’s incorrect or this issue still needs to be addressed, please simply write any comment. If closed, you can reopen the issue at any time. Thank you for your contributions.
knockknock10 commentedon Nov 12, 2026
Hi @scwhittle,
If u don't mind 🙂
I'd be interested in helping investigate this issue.
I went through the discussion around exception propagation at fused DoFn boundaries. The case that particularly concerns me is when the upstream DoFn catches an exception thrown by the downstream DoFn, making the bundle appear to have continued successfully even though downstream processing failed.
I'd like to start with a focused Java harness regression test for a fused A → B pipeline where B throws and A has its own exception handling.
From there, I'd be interested in exploring how the exception can be recorded at the consumer boundary and surfaced as a bundle failure without allowing subsequent outputs to continue incorrectly.
Is this still part of the Beam 3.0 plan, and would a test-first contribution be useful at this stage?
scwhittle commentedon Nov 12, 2026
Hi, yes that is the case I am concerned about as well. I think that it was decided to not pursue Beam 3.0 as a big bundling of improvements/features and instead to make progress gradually and to instead announce such changes to behavior.
So yes, you'd be welcome to help contribute here. Writing a test sounds like a good way to start.
As for how to capture, I was thinking something along the lines of wiring output from A to go through a reciever that has a try-catch block around forwarding to B. Then if B throws an exception the receiver could note it. We could have various modes controlled by options on what the receiver does then:
knockknock10 commentedon Nov 12, 2026
Thanks, Sam! did I pronounce ur name well? sorry if not 🙂!
I went through the issue and started looking into how the exception is passed between fused DoFns.
From what I understand, when DoFn A sends an element to DoFn B and B throws an exception, that exception travels back to A. If A catches it, the bundle might still complete successfully even though B failed. I think we should first reproduce this properly with a test.
I'll start by checking the existing Java harness tests and add a regression test with two fused DoFns, where A catches the exception from B and continues processing. I'll also check the case where A doesn't catch the exception to make sure the existing behaviour is covered.
Once we have the test reproducing the issue, I'll look into the receiver wrapper approach you suggested and how we can track such failures at the bundle level. I'll also check how this works with bundle completion and processor reuse before deciding where the change should go.
I'll start with the tests and share what I find here.
knockknock10 commentedon Nov 13, 2026
Hi @scwhittle,
I've opened PR #41285 with the Java harness implementation.
It follows the three modes you suggested (
legacy,warn, andfail), controlled through thefused_dofn_exception_handlingexperiment. The default remainslegacy, so existing behavior is unchanged unless the experiment is enabled.I've added regression tests for the swallowed-exception scenario and the related bundle handling cases. Both targeted test classes pass locally (43 tests).
Could you take a look when you get a chance? I'd appreciate your feedback.