[
https://issues.apache.org/jira/browse/FLINK-41206?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18134735#comment-18134735
]
Biao Geng commented on FLINK-41206:
-----------------------------------
Hi [~sangkyoonnam], thanks for your interest. I am still working on this one.
For the issue, adding `tee` can help in the lauch stage but when the job is
running, it cannot report the correct OOM cause either, which makes it less
helpful for the real job in production. That's why I did not directly submit a
PR.
> 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
> Priority: Major
> Labels: pull-request-available
>
> 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. One possible solution
> 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.
> 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)
