[
https://issues.apache.org/jira/browse/FLINK-41206?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18134703#comment-18134703
]
Sangkyoon Nam commented on FLINK-41206:
---------------------------------------
I'd like to take this if you aren't already preparing a PR. I checked master
({{ebfb1b0c945}}), and {{pyflink-udf-runner.sh}} still pipes Python through
{{tee}}. I'd integrate your proposed approach onto master and leave the
callback location for review.
I haven't run your reproduction yet. I'd use it to validate the change on
master and check for a TaskManager INFO exit record with the worker ID, Python
PID and {{exitCode=137}}.
Please share the prototype branch or patch if you have one I can build on. If
the approach looks good, could a committer assign this to me?
> 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
>
> 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)
