[ChickenBenny]

closes: #75557

supervise_task fails a task whose runtime cannot launch, with the reason in its task log, only when the coordinator raises TaskLaunchError. SubprocessCoordinator raised it only for a native Dag file that is missing or cannot run. Any other failure before the runtime connected reached the executor before the task instance was started: a @task.stub artifact that cannot be found, a runtime that cannot be executed (java not installed, wrong architecture), an unsupported supervisor schema version, or a runtime that exits or times out before connecting. The try then had no task log, the task was failed through the scheduler's "state mismatch" path, and the runtime's output, already read by _accept_connections, was dropped. A JVM that printed Error: Could not find or load main class ... left that line in no log at all.

Rather than catching each of these where it is raised, SubprocessCoordinator now raises any failure before the task instance is started as TaskLaunchError, through one context manager, _raise_as_launch_error, around two blocks:

• in execute_task, resolving the Dag bundle and building the command;
• in _PopenActivitySubprocess.start, starting the runtime until it connects.

_on_child_started, which starts the task instance, and wait() stay outside, because the supervisor starts the task instance itself when it handles a TaskLaunchError. A TaskLaunchError raised inside, such as a routing mistake that is not retryable, is kept as it is. The blocks are only indented, so the diff reads best with whitespace ignored.

The runtime's output does not go into the error. _accept_connections writes what the runtime printed before it failed to connect to the task log, one event per line under task.stdout and task.stderr, through the same forwarder a connected runtime's output goes through, so each line keeps its own level and timestamp. It keeps the last 16 KiB of each stream, enough for a JVM or Node stack trace, and says so when it drops earlier output. Output still buffered in the sockets is read without blocking, because a runtime can exit before its last lines are read, and a runtime that keeps printing while it never connects is read no further than a fixed budget. Reusing that forwarder is why WatchedSubprocess._target_loggers is split out of _get_target_loggers: the loggers to forward to have to be resolved before the subprocess object exists.

Most of the new failures are retryable, like a native Dag file that is missing or cannot run: an artifact can appear once a bundle finishes syncing, and a runtime that timed out on a busy worker can start on the next try. A supervisor schema version this worker's Task SDK does not know is not retryable, since retrying cannot teach it one. Catching Exception also reports a coordinator bug as a launch failure, without a retry; its traceback is then in the task log rather than in no log.

TaskLaunchError said a coordinator raises it before it starts the runtime, which no longer holds: a runtime that exits or times out before connecting is already past Popen. It and BaseCoordinator.execute_task now state the contract custom coordinators implement against as the task instance not having been started, which is what the supervisor does itself when it handles the error. A coordinator that knows whether retrying helps should say so rather than leave it to the exception type:

def _build_execute_task_command(self, *, what: TaskInstance) -> tuple[list[str], str | None]:
    if (jar := self._find_jar(what.dag_id)) is None:
        raise TaskLaunchError(f"no JAR for {what.dag_id!r}", retryable=False)
    return ["java", "-jar", str(jar)], _SCHEMA_VERSION

[jason810496]

Just made some changes myself to get this PR in 3.4.0 RC.

[uranusjr]

Also in oordinator.py, TaskLaunchError says "before it starts the runtime", and execute_task says "Raise it before the runtime starts". This PR raises it after Popen (the runtime exited or timed out before connecting).

The real invariant, which _raise_as_launch_error's own docstring states, is "before the task instance is started". Update both docstrings. Custom coordinators implement against this contract.

[jason810496]

Thanks for the review.

[github-actions]

Backport successfully created: v3-4-test

Note: As of Merging PRs targeted for Airflow 3.X
the committer who merges the PR is responsible for backporting the PRs that are bug fixes (generally speaking) to the maintenance branches.

In matter of doubt please ask in #release-management Slack channel.

Status	Branch	Result
✅	v3-4-test