jason810496 opened a new pull request, #71075: URL: https://github.com/apache/airflow/pull/71075
- related: https://github.com/apache/airflow/pull/70805#discussion_r3690030860 ## Why The language-SDK subprocess coordinator left the task instance QUEUED until the runtime had connected back over the comm/logs sockets, so JVM boot, artifact discovery and the whole `task_startup_timeout` window were charged to `[scheduler] task_queued_timeout` instead of the task's runtime, and a runtime that never started looked like a task nothing had ever picked up. ## How The Python path reports the forked child's pid before handing it any work and does its Dag bundle materialization inside that window (`task_runner.parse()`). This brings the coordinator path in line: RUNNING is reported at the very top of `execute_task`, before `_build_execute_task_command`, so every bit of worker-side preparation happens with the task RUNNING. - The pid reported is this supervisor's, not the runtime's, because that is the only way the transition can precede the spawn: - the server stores the pid `task_instances.start` carried and 409s any heartbeat whose pid differs, killing the task (`task_instances.py:944-958`, `supervisor.py:1626-1639`); - a second `start` with a new pid is also 409 `invalid_state` (`task_instances.py:205-229`), so re-reporting the real pid later is not an option. - `_PopenActivitySubprocess.supervisor_pid()` is the single seam both the transition and every heartbeat read, so they cannot drift. `InProcessTestSupervisor` already reports `os.getpid()` this way. - Nothing changes on the wire. `ToSupervisor` has no "enter running" message and heartbeats are sent by the supervisor, so no language SDK needs a change. - The preparation window now heartbeats. Nothing else does until `wait()` starts monitoring, so a slow launch (a first-time bundle clone, a raised `task_startup_timeout`) would otherwise look like a zombie to `[scheduler] task_instance_heartbeat_timeout` and be reaped mid-startup. The Python path gets this for free because its supervisor is already monitoring while the child does the same work. - Since the task is RUNNING from the first call on, `execute_task` owns the terminal transition when it never gets a process to monitor: it reports FAILED, or UP_FOR_RETRY when the run context says so, and returns a non-zero-exit `ExecutionResult` rather than raising. That is what the Python path already does for a failed task, so the executor and scheduler take the ordinary failed-task path instead of the "killed externally" branch. If that report itself fails the exception escapes, because returning a state the server never recorded would put the run straight back in the RUNNING-until-reaped hole. - `execute_task` holds `_warm_shutdown_signals()` across the RUNNING window, as `_PythonCoordinator` does, so a SIGTERM in there cannot kill the supervisor and orphan a running task. ## What - `SubprocessCoordinator.execute_task`: report RUNNING before `_build_execute_task_command`; heartbeat through the launch via new `_heartbeat_until_monitored`; install the warm-shutdown handlers; report the terminal state via new `_finish_failed_startup` when the runtime never starts. - `ActivitySubprocess`: extract the RUNNING transition into `_report_running` so `start()` performs it and `_on_child_started` only ever consumes a `ti_context` (no nullable dual-purpose arg); add `_reported_pid` and use it for both the transition and heartbeats. - `_accept_connections`: raise the new `SubprocessStartupError` (carrying the runtime's exit code) instead of bare `TimeoutError` / `RuntimeError`, and write the runtime's pre-handshake stdout/stderr to the task log instead of discarding it with the drain buffers. Its own diagnostics go to the task logger too. - Document in the Java / Go / TypeScript pages that the `task_startup_timeout` wait happens with the task already running, and that the pid recorded on the task instance is the supervisor's. Two existing tests change: the `pytest.raises(TimeoutError)` / `pytest.raises(RuntimeError)` assertions around `_accept_connections` now expect `SubprocessStartupError`, and four others gain the required `ti_context=` kwarg. Neither accommodates a behavior regression; `test_running_reported_before_any_preparation` is the regression test that fails without this change. Not addressed, and unchanged by this PR: `_ResourceTracker.__exit__` terminates only the direct `Popen`, so a launcher that forks the real runtime and exits can orphan it when startup fails. A robust fix needs `start_new_session` plus a process-group kill, which changes signal delivery for every language runtime and belongs in its own PR. This PR reduces the exposure rather than adding to it: a startup failure used to leave the run QUEUED for the scheduler to requeue twice before failing (up to three launch attempts, so up to three orphans), and is now a single attempt followed by a terminal state. No newsfragment: the coordinator feature is itself unreleased on `main` (`airflow-core/newsfragments/68548.feature.rst` still pending), so nothing here is visible to a released version. ## Verification - `uv run --project task-sdk pytest task-sdk/tests/task_sdk/coordinators -q` — 137 passed. - `uv run --project task-sdk pytest task-sdk/tests/task_sdk/execution_time/test_supervisor.py task-sdk/tests/task_sdk/execution_time/test_coordinator.py -q` — 190 passed, 1 pre-existing failure (`test_remote_logging_conn_caches_connection_not_client`, needs the amazon provider; identical on a clean tree). - `uv run --project task-sdk pytest task-sdk/tests -q --ignore=task-sdk/tests/task_sdk/serde` — the only failures are the 47 that also fail on a clean checkout of this commit's parent (missing pandas / celery / providers-manager deps in the local venv). - `prek run --from-ref upstream/main --stage pre-commit` and `--stage manual` — both pass. - `prek run mypy-task-sdk --all-files` — passes. --- ##### Was generative AI tooling used to co-author this PR? - [x] Yes, with help of Claude Code (Opus 5) following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
