bgeng777 opened a new pull request, #29677: URL: https://github.com/apache/flink/pull/29677
## What is the purpose of the change Fix [FLINK-41271](https://issues.apache.org/jira/browse/FLINK-41271). A terminal Beam data RPC failure can leave PyFlink tasks RUNNING while waiting for bundle output. A co-located task can also retain the failed shared worker, causing subsequent target retries to reuse its closed channel. ## Brief change log - Observe data RPC termination through Beam's existing server-factory extension point and notify the affected Python runners. - Unblock result and bundle-completion waits, fail the affected tasks, and ignore callbacks during resource teardown. - Add tests for RPC termination, shared listeners, failure ordering and blocked waits. No Beam dependency changes are required. ## Verifying this change - 21 targeted tests passed in `PythonJobBundleFactoryTest`, `BeamPythonFunctionRunnerTest` and `AbstractExternalPythonFunctionOperatorTest`; Spotless and Checkstyle passed. - Standalone reproduction: a real Netty receive allocation fails after a 16 MiB Python response. Without the fix, the task remained RUNNING for the 67.8-second observation window. With the fix and restart disabled, it automatically becomes FAILED and the bundle-flush thread exits. The no-pressure control finishes normally. - With retries enabled and no test watchdog, recovery succeeds without a keeper and in 3/3 shared-worker runs. All affected tasks restart using a new worker and data channel, then finish. Full-repository `clean verify` was not completed locally. ## Does this pull request potentially affect one of the following parts: - Dependencies: no. - Public API: no. - Serializers: no. - Runtime per-record code paths: yes; Python runners check for data RPC failure. - Deployment or recovery: yes; tasks sharing a failed Python data RPC are failed so they can recover. - S3 file system connector: no. ## Documentation - Does this pull request introduce a new feature? No; this is a bug fix. - Documentation: not applicable. --- ##### Was generative AI tooling used to co-author this PR? - [X] Yes Generated-by: Codex (GPT-6) -- 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]
