Hacktoberfest 2026: le issue che i maintainer hanno segnato per ottobre, aperte e adatte ai principianti. Sfoglia le issue Hacktoberfest

Task supervisor raises TypeError when a task log event has a key named target or msg, orphaning the task process and recording a successful task as failed

Aperta Adatta ai principianti
  • Qualcosa non va?
#75,017 1 commento 0 reazioni 0 assegnatari Vedi su GitHub

I maintainer di solito rispondono entro 1 giorno

@liwenjie200543 ci sta già lavorando.

Dal 19/10/2026.

  • #75029 di @liwenjie200543 — aperta

Valutazione

Difficoltà
2/5
Tempo stimato
1-3 ore
Idoneità per principianti
78/100
Tipo di issue
Bug
Chiarezza
Specificata chiaramente
Stato di attività
Attiva
Stack tecnologico
python
Ambito
backend

Direzione di ricerca

Leggi process_log_messages_from_subprocess e _log_to_target in task-sdk/src/airflow/sdk/execution_time/supervisor.py, poi esamina test_process_log_messages_from_subprocess. Aggiungi un test per un evento che contenga sia target sia msg e verifica che l’inoltro si completi senza generare eccezioni e preservi l’evento. Il lavoro è concluso quando il test di regressione passa e l’elaborazione dei log del task continua.

Scritto dal modello di indicizzazione a partire dal testo della issue.

Descrizione

Under which category would you file this issue?

Task SDK

Apache Airflow version

main (development), reproduced at d2c41b8c2027fcba5e632f00f5f80d86ec767981. Also 3.3.2 with apache-airflow-task-sdk 1.3.2 and 3.4.0b1 with task-sdk 1.4.0b1, which carry the same helper. 3.3.1 and task-sdk 1.3.1 are not affected.

What happened and how to reproduce it?

process_log_messages_from_subprocess in task-sdk/src/airflow/sdk/execution_time/supervisor.py (lines 2909-2921) pops level, event and exception from each JSON log event coming from the task process and spreads every other key into the helper _log_to_target(target, level, msg, **event) (line 2913, defined at line 2916). A log event that carries a key named target or msg collides with the helper's own parameter names and raises:

TypeError: _log_to_target() got multiple values for argument 'target'

Nothing on the way out catches it: make_buffered_socket_reader.cb catches only StopIteration, _service_subprocess only BrokenPipeError and ConnectionResetError, ActivitySubprocess.wait has only a finally that closes the selector, _PythonCoordinator.execute_task has no handler and supervise_task catches only TaskLaunchError and httpx request errors. So wait() raises before update_task_state_if_needed() and _upload_logs(), the heartbeat loop stops, and the child, which runs in its own process group, is not killed.

An ordinary log call in a task is enough:

structlog.get_logger().info("deploying", target="prod")

Airflow itself supplies a second trigger. task_runner.py line 1239 logs log.debug("Using serialized startup message from environment", msg=msg) in the run_as_user re-exec path, and debug events are forwarded to the supervisor when [logging] logging_level = DEBUG. So every impersonated task in a DEBUG-logging deployment should hit this with no Dag author involvement (the debug forwarding of a msg= key was reproduced; the sudo re-exec itself was not run).

Observed on the checkout, driving supervise_task with a real forked child and a mock API transport that records every request:

info("deploying", target="prod")   -> supervise_task raised TypeError after 0.35 s
   requests sent: PATCH .../run, one PUT .../heartbeat; no state PATCH, no log upload
   child not killed: ran user code to the end, then its terminal message failed with
   ConnectionResetError; pid still present 9 s after the supervisor raised
info("deploying", msg="prod")      -> same, "multiple values for argument 'msg'"
debug("deploying", msg="prod")     -> rc 0 at logging_level=INFO, TypeError at DEBUG
info("deploying", environment="prod") -> rc 0, PATCH .../state {"state": "success"}

Under the standard executors the task instance does not stay running until the heartbeat timeout: the LocalExecutor worker reports the workload's failure state and the Celery wrapper re-raises, so the scheduler's process_executor_events sees a failed event for a running task instance, logs a state mismatch, counts tasks.killed_externally and calls handle_failure (read from the code, the executor and scheduler were not run). The result is that a task whose user code succeeds is recorded as failed or up for retry, a retry can start while the orphaned process is still running user code, and the task's final logs are never uploaded.

The same generator serves the Dag processor: DagFileProcessorProcess registers it through WatchedSubprocess._register_pipe_readers, and DagFileProcessorManager._service_processor_sockets (airflow-core/src/airflow/dag_processing/manager.py lines 636-639) catches only the same two socket errors while DagProcessorJobRunner._execute logs and re-raises. A Dag file with structlog.get_logger().info("parse", target="prod") at module level, run under DagFileProcessorProcess.start, raised the same TypeError from the processor's selector callback; from the code, that stops the Dag processor job and does so again on every restart.

The helper was added by #69523 (6d9cc2ed37) and backported by #72888 (3b6f63eef9). Before it, the code called target.log(level, msg, **event) directly, where only level and event could collide and both are popped.

How to reproduce

  1. A Dag with a Python task whose callable runs structlog.get_logger().info("deploying", target="prod") and then sleeps a few seconds.
  2. Run it under the task SDK supervisor (any executor).
  3. The supervisor logs the TypeError above and exits without sending a terminal state; the task process keeps running; the task instance is then failed by the scheduler's state-mismatch handling.

For the Dag processor: put the same log call at module level in a Dag file and watch the Dag processor stop.

What you think should happen instead?

A log event's keys should never be able to collide with the forwarding helper's parameters. Making them positional-only removes both collisions:

def _log_to_target(target, level, msg, /, **event):

Alternatively pass the event as one dict and call target.log(level, msg, **event) inside. The existing test_process_log_messages_from_subprocess sends only level and event; a case with {"level": "info", "event": "x", "target": "y", "msg": "z"} would cover this.

Operating System

Linux

Deployment

Other

Apache Airflow Provider(s)

No response

Versions of Apache Airflow Providers

Not relevant.

Official Helm Chart version

No response

Kubernetes Version

Not Applicable

Helm Chart configuration

Not Applicable

Docker Image customizations

Not Applicable

Anything else?

With [logging] logging_level = DEBUG and run_as_user set, the startup debug line in task_runner.py (line 1239) carries msg= and should trigger the same crash for every impersonated task.

Are you willing to submit PR?
  • Yes I am willing to submit a PR!
Code of Conduct
Lingua principale
Python
Stelle
47.3k
Fork
18.1k
Merge medio
23h 27m
PR unite (30g)
629

Preparare l'ambiente

Apri in Codespaces

Avvia il container di sviluppo del progetto nel browser, con il tuo account GitHub.

Come iniziare

  1. Leggi tutta la issue e poi la guida ai contributi del progetto.
  2. Commenta sulla issue per dire che te ne occupi tu — evita che due persone facciano lo stesso lavoro.
  3. Fai un fork del repository e lavora su un branch.
  4. Apri una pull request che faccia riferimento al numero della issue.

Altre issue di apache/airflow

Tutte le issue di apache/airflow

Issue simili

Altre issue su Python

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.