Forward termination signals from supervisor to task subprocess

[andreahlert]

Fixes: #58936

Summary

When a Kubernetes worker pod receives SIGTERM (e.g. spot interruption, scaling down, rolling update), the signal is delivered to the supervisor process (PID 1 in the container). Previously, the supervisor had no signal handler installed and would exit with default behavior, leaving the task subprocess orphaned without ever calling the operator's on_kill() hook. This meant spawned resources (pods, subprocesses, etc.) were never cleaned up.

Root cause: The supervise() function starts the task subprocess and calls process.wait(), but never installs signal handlers for SIGTERM/SIGINT. The task subprocess does have a SIGTERM handler (registered in task_runner.py) that calls on_kill(), but the signal never reaches it because the supervisor process terminates first.

Fix: Install SIGTERM/SIGINT signal handlers in supervise() that forward the received signal to the task subprocess via os.kill(). The child's existing handler then calls on_kill() as expected, restoring the Airflow 2 behavior.

Signal flow after fix:

1. K8s sends SIGTERM to supervisor (PID 1)
2. Supervisor's new handler forwards SIGTERM to task subprocess
3. Task subprocess's existing _on_term handler calls operator.on_kill()
4. Operator cleans up resources (pods, subprocesses, etc.)
5. Subprocess exits, supervisor's wait() returns normally

Changes

• task-sdk/src/airflow/sdk/execution_time/supervisor.py: Added signal forwarding in supervise() function. Signal handlers are saved, installed before process.wait(), and restored in a finally block.
• task-sdk/tests/task_sdk/execution_time/test_supervisor.py: Added test that verifies SIGTERM forwarding from supervisor to subprocess triggers the operator's on_kill() hook.

Test plan

• New test test_on_kill_hook_called_when_supervisor_receives_sigterm verifies the signal forwarding chain
• Existing test_on_kill_hook_called_when_sigkilled still passes (no regression)
• Existing signal-related tests (test_kill_escalation_path, test_exit_by_signal) still pass

[SameerMesiah97]

Looks fine but needs slightly more polish. It seems like CI might have flaked but that should hopefully not happen when you force push again.

[SameerMesiah97]

Are you swallowing the exception here because you anticipate that the child worker has been killed before invoking os.kill? If that is the case, I would add a comment here explaining that. Like this:

# Child process may have already exited during shutdown races.

This is more a nit but silent exception swallowing tends to raise eyebrows for readers who might have partial context.

[SameerMesiah97]

Your new implementation handles both SIGTERM and SIGINT, but you appear to be testing only SIGTERM here. Is this because you only anticipate K8s to send SIGTERM? I would suggest explaining that here in a comment so that a casual reader does not assume this is a gap. Not a blocking suggestion.

[SameerMesiah97]

I get why the loop needs to run “long enough” so the subprocess is alive when the signal is delivered, but 1000 iterations at 1s each feels a bit overkill. If for some reason the subprocess doesn’t get terminated as expected, this could run for ~15 minutes and materially stall CI. Is there any reason this can’t be reduced to a smaller number (e.g. 30–60 iterations) while still leaving plenty of headroom for signal delivery?