Biao Geng created FLINK-41206:
---------------------------------
Summary: PyFlink loses Python worker exit status when the worker
process is killed
Key: FLINK-41206
URL: https://issues.apache.org/jira/browse/FLINK-41206
Project: Flink
Issue Type: Improvement
Components: API / Python
Reporter: Biao Geng
h3. Problem
When a Python worker is killed while the TaskManager remains running, Flink may
report only errors such as:
{code:java}
Failed to close remote bundle
CANCELLED: client cancelled
{code}
These messages do not clearly tell users that the Python worker process has
exited. The default TaskManager logs also do not record its exit code. Users
may therefore investigate a communication problem without realizing that the
Python process has died.
h3. Expected behavior
The default TaskManager log should explicitly record that the Python worker has
exited, including its worker ID, PID and exit code, without requiring DEBUG
logging.
This gives users a clear starting point: identify the worker that exited, then
use its PID and exit code to investigate the cause. For example, exit 137 is
consistent with SIGKILL and can help users correlate the failure with
OS/container OOM records. It does not, by itself, prove an OOM kill.
h3. Reproduction
# Start a local standalone session cluster at INFO level. Run a one-row
scalar-UDF job in process mode; have the UDF write {{os.getpid()}} and worker
ID to a marker file, then wait in a loop.
# Once the UDF is entered, send {{kill -9}} to that actual Python PID, leaving
the launcher/JM/TM alone.
# Inspect the task failure and TaskManager log.
The task reports a gRPC/bundle failure, but no explicit worker-exit record with
the Python PID and exit code. These tests deliberately send SIGKILL; they do
not induce an OOM.
h3. Why the exit information is lost
The TaskManager starts a Bash [launcher
script|https://github.com/apache/flink/blob/0addadb5caeead0aedaa780443cce89ae6ff0487/flink-python/pyflink/bin/pyflink-udf-runner.sh],
which then starts Python. Java directly manages the script's process, so the
exit status it receives is the script's.
The script uses {{tee}} to save Python's output. By default, it returns
{{{}tee{}}}'s exit status, so Python can fail while the script reports success.
I tried changing the script to return Python's status. This corrected startup
failures, but exposed another race during runtime failure: after the Python
worker dies, Flink starts cleanup and can terminate the script before it
forwards Python's status. Java then observes the script's termination code,
143, instead of Python's SIGKILL code, 137.
Printing an exit message from the script also does not reliably make it visible
in default TaskManager logs. The script's output is separate from Python's Beam
logging channel, and Beam discards that process output unless Beam's
ProcessManager DEBUG logging is enabled.
h3. Alternatives tested
||Attempt||Result||
|Preserve {{PIPESTATUS}} and print a shell hint|Startup SIGKILL changed exit 0
to 137; missing {{grpc}} changed 0 to 1. Runtime default logs still lacked the
diagnostic.|
|Enable ProcessManager DEBUG|Exposed the shell's 137 hint in MiniCluster JVM
stderr, but requires DEBUG.|
|Read the boot log once on bundle failure|Added startup identity information,
but missed the 137 hint in both runtime tests.|
|Enrich exceptions using a bounded 500 ms probe|Included 137 in an earlier
MiniCluster prototype; rejected from this change because of added
coordination/waiting.|
|Add an INFO callback, retaining Bash|Standalone: Python PID 67309 was killed;
the callback logged Bash PID 67307 / 143 after cleanup.|
|{{exec}} Python, retaining the INFO callback|Two standalone kills logged
actual Python PID / 137. Cancellation logged 143; 1,000 normal rows completed;
missing {{grpc}} retained its traceback and reported 1.|
h3. Proposed change
Replace Bash with Python while retaining output mirroring and the boot log:
{code:java}
log="$BOOT_LOG_DIR/flink-python-udf-boot.log"
exec > >(tee "${log}") 2>&1
exec "${python}" -m pyflink.fn_execution.beam.beam_boot "$@"
{code}
Add one neutral {{Process.onExit()}} INFO record per worker exit:
{code:java}
Process for worker 1-1 exited: pid=38483, exitCode=137
{code}
This removes the separate shell that can lose Python's status during cleanup.
No application-level waiting, polling, exit-classification flags or new
environment factory is proposed. The prototype uses a Flink-local Beam
ProcessManager copy; the hook's integration location remains a review decision.
h3. Validation and limits
The prototype used a formal Maven-packaged JAR, separate standalone
JM/TM/client JVMs and INFO logging. The same Java JAR logged Bash/143 with the
pipeline launcher and Python/137 with {{{}exec{}}}. Worker PID/parent checks
confirmed the process identity. Callback regressions covered exit 0, exit 23
and SIGKILL; module verification passed 370 unit tests and one integration
test. Launcher/boot checks covered cleanup races, log-write failure and delayed
log writing.
The combined prototype was tested on 2.4-SNAPSHOT. Unmodified master
{{0addadb5caee}} also lacks the default exit diagnostic; integrating the
prototype on that master remains outstanding.
Intentional changes: background {{tee}} makes immediate boot-log snapshots best
effort; its failures no longer determine Python's exit status; cleanup SIGTERM
reaches Python directly; every exit adds INFO. A successful job's worker
cleanup also logged 143, so nonzero exits must not automatically be classified
as crashes.
This change adds a log record; it does not guarantee 137 in the final Java
exception, identify OOM automatically, or fix bundle/data-channel deadlocks.
Windows handling is outside this first step. Logging also cannot be guaranteed
if the TaskManager itself is killed.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
