Dag Processor Error

[aa-matthias]

Apache Airflow version

3.0.0

If "Other Airflow 2 version" selected, which one?

No response

What happened?

When running DAG processor these errors are shown in stderr - DAGs are not parsed successfully.

Note: Same DAG seems to parse successful when running airflow dags reserialize

--- Supervised process Last chance exception handler ---
Traceback (most recent call last):
  File "./virtualenvs/airflow/lib/python3.12/site-packages/airflow/sdk/execution_time/supervisor.py", line 368, in _fork_main
    target()
  File "./virtualenvs/airflow/lib/python3.12/site-packages/airflow/dag_processing/processor.py", line 82, in _parse_file_entrypoint
    msg = comms_decoder.get_message()
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "./virtualenvs/airflow/lib/python3.12/site-packages/airflow/sdk/execution_time/task_runner.py", line 632, in get_message
    msg = self.decoder.validate_json(line)
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "./virtualenvs/airflow/lib/python3.12/site-packages/pydantic/type_adapter.py", line 468, in validate_json
    return self.validator.validate_json(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "./virtualenvs/airflow/lib/python3.12/site-packages/pydantic/_internal/_mock_val_ser.py", line 100, in __getattr__
    raise PydanticUserError(self._error_message, code=self._code)
pydantic.errors.PydanticUserError: `TypeAdapter[typing.Annotated[typing.Union[ForwardRef('DagFileParseRequest'), airflow.sdk.execution_time.comms.ConnectionResult, airflow.sdk.execution_time.comms.VariableResult, airflow.sdk.execution_time.comms.ErrorResponse, airflow.sdk.execution_time.comms.OKResponse], FieldInfo(annotation=NoneType, required=True, discriminator='type')]]` is not fully defined; you should define `typing.Annotated[typing.Union[ForwardRef('DagFileParseRequest'), airflow.sdk.execution_time.comms.ConnectionResult, airflow.sdk.execution_time.comms.VariableResult, airflow.sdk.execution_time.comms.ErrorResponse, airflow.sdk.execution_time.comms.OKResponse], FieldInfo(annotation=NoneType, required=True, discriminator='type')]` and all referenced types, then call `.rebuild()` on the instance.

For further information visit https://errors.pydantic.dev/2.11/u/class-not-fully-defined

What you think should happen instead?

No response

How to reproduce

upgraded from 2.10.5 - migrated db and config, adapted dags, added new supervisor program for dag-processor

Operating System

Ubuntu 24.04.2 LTS

Versions of Apache Airflow Providers

No response

Deployment

Virtualenv installation

Deployment details

No response

Anything else?

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[eladkal]

Can you share dag example that reproduce this error?

[aa-matthias]

from datetime import datetime, timedelta

import pendulum
from airflow.providers.standard.operators.bash import BashOperator
from airflow.sdk import DAG

local_tz = pendulum.timezone("Europe/Berlin")

default_args = {
    "owner": "person1",
    "depends_on_past": False,
    "start_date": datetime(2019, 5, 31, tzinfo=local_tz),
    "email": ["alerting@company.com"],
    "email_on_failure": True,
    "email_on_retry": False,
    "retries": 1,
    "retry_delay": timedelta(minutes=5),
    # 'queue': 'bash_queue',
    "pool": "abb",
    # 'priority_weight': 10,
    # 'end_date': datetime(2016, 1, 1),
}

dag = DAG(
    dag_id="run_legacy_tasks",
    catchup=False,
    max_active_runs=1,
    default_args=default_args,
    schedule="0 6 * * *",
    tags=["one", "two", "three"],
    doc_md=__doc__,
)

t1 = BashOperator(
    task_id="job1",
    bash_command="/srv/airflow/legacy/job1/script.sh ",
    dag=dag,
)

[aa-matthias]

I don't think the actual dag code is causing this issue - i get it with the dag from the docs as well 🤔

 import datetime

 from airflow.sdk import DAG
 from airflow.providers.standard.operators.empty import EmptyOperator

 my_dag = DAG(
     dag_id="my_dag_name",
     start_date=datetime.datetime(2021, 1, 1),
     schedule="@daily",
 )
 EmptyOperator(task_id="task", dag=my_dag)

[jedcunningham]

What version of the apache-airflow-task-sdk do you have?

[aa-matthias]

1.0.0 with airflow 3.0.0 and 1.0.1 with airflow 3.0.1 (tried both in separate virtualenvs)

[aa-matthias]

is there anything else i can provide (logs, env, config)?