[
https://issues.apache.org/jira/browse/FLINK-41271?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18142808#comment-18142808
]
Dian Fu edited comment on FLINK-41271 at 11/11/26, 11:12 AM:
-------------------------------------------------------------
Fixed in:
- master via e790cfb25215448268df2cdbab7cbcd556c21acb
- release-2.4 via 762a32604fc5d872d241a6689190f9681122e432
- release-2.3 via 9710d9652b006044cd4cc860050c227574d37f43
- release-2.2 via 51270ca488143f37b7f9b6f3ce75e6f1845701a2
- release-1.20 via 922ac2741ad5985f97d3ca5b029b642facd7ded0
was (Author: dianfu):
Fixed in:
- master via e790cfb25215448268df2cdbab7cbcd556c21acb
- release-2.4 via 762a32604fc5d872d241a6689190f9681122e432
> PyFlink tasks remain RUNNING after a Beam data RPC failure
> ----------------------------------------------------------
>
> Key: FLINK-41271
> URL: https://issues.apache.org/jira/browse/FLINK-41271
> Project: Flink
> Issue Type: Bug
> Components: API / Python
> Environment: Flink 2.5-SNAPSHOT (local baseline
> {{{}8cab5daa02e5{}}}), Beam 2.75.0, Java 17.0.16, Python 3.11.13. Standalone
> cluster with one slot, parallelism 1 and Python process mode.
> Reporter: Biao Geng
> Assignee: Biao Geng
> Priority: Major
> Labels: pull-request-available
> Attachments: pyflink-data-rpc-reproducer.zip
>
>
> A PyFlink process-mode task can remain RUNNING after its Beam data RPC has
> terminated. The bundle-flush thread continues waiting for output completion,
> preventing the task from reporting a failure, so a configured restart
> strategy cannot take effect.
> This was reproduced with a real direct-memory allocation failure in the JVM's
> shaded Netty receive path. The Python worker remains alive; the test does not
> kill it or manually throw an OutOfMemoryError.
>
> *Reproduction*
> Commands, fixture and logs are attached.
> # Run Java source → Python map → Java sink. Confirm that a small warmup
> response reaches the sink.
> # With {{{}-XX:MaxDirectMemorySize=128m{}}}, retain direct buffers in the
> TaskManager until approximately 512 KiB remains.
> # Return a 16 MiB string from Python. This triggers a real direct OOM in the
> Netty receive path and terminates the data RPC.
> # For shared-worker recovery, add an independent keeper branch sharing the
> worker. A test-only watchdog fails the target once while the keeper stays
> running; this is necessary to test recovery independently of the missing
> automatic failure detection.
> *Actual behavior*
> * The task remains RUNNING for the 67.8-second observation window. Both
> thread dumps show {{ActiveBundle.close}} blocked in
> {{{}BeamFnDataInboundObserver.awaitCompletion{}}}.
> * With the keeper, target retries reuse the same closed data channel and
> fail with {{call already cancelled}} / {{{}Channel closed prematurely{}}}.
> The keeper stays at attempt 0. Direct memory has fallen to 77.8 MiB, with no
> further pressure injection.
> * Without the keeper, the forced-failure control creates a new worker and
> recovers.
> *Expected behavior*
> A terminal data RPC failure should fail all affected Python tasks and release
> their blocked waits. With retries enabled, they should recover using a
> functioning worker and data channel.
> *Fix and validation*
> The Flink-only fix observes RPC termination through Beam’s existing
> server-factory extension point and notifies all affected runners.
> * No-pressure control: complete 16 MiB result and FINISHED.
> * OOM with restart disabled: automatic FAILED and bundle-flush thread exits.
> * Recovery without a keeper succeeds. Shared-worker recovery succeeds in 3/3
> runs: both tasks restart at attempt 1 with a new worker/channel and FINISH.
> All fix-validation runs disable the watchdog.
> * 21 targeted tests, Spotless and Checkstyle pass.
> The fix releases the failed worker by restarting all affected tasks,
> including the keeper.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
