knockknock10 opened a new pull request, #41285: URL: https://github.com/apache/beam/pull/41285
This PR addresses https://github.com/apache/beam/issues/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. The implementation records exceptions at the consumer boundary, applies the selected policy during bundle completion, and resets the recorded exception state 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. ### Validation Executed: ```bash ./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`: 26 passed. - `PCollectionConsumerRegistryTest`: 15 passed. - Total: **41 passed, 0 failures, 0 errors, 0 skipped.** The branch was rebased onto the latest Apache Beam `master` before this validation. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
