Biao Geng created FLINK-41271:
---------------------------------
Summary: 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
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)
