[
https://issues.apache.org/jira/browse/FLINK-41271?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Dian Fu reassigned FLINK-41271:
-------------------------------
Assignee: Biao Geng
> 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)
