Schedulers crashing when upgrading to 2.10.4 from 2.8.2

[nkjmsg]

Hi,

I recently attempted to upgrade the Airflow version from 2.8.2 to 2.10.4. In my local machine, using Docker, I was able to upgrade to 2.9.2 from 2.8.2 successfully, then 2.10.4 from 2.9.2 thereafter. All containers were brought up just fine and I performed validations of DAG executions to ensure that everything was working as expected. I also had some of my colleagues test the new version (2.10.4) in local and they were able to complete their validations with some minor changes.

However, when I pushed the change upstream to our development and QA environment (self-managed via AWS EKS) the schedulers started crashing (going through the cycle of CrashLoopBackOff and restarting continuously) despite the build and deployment (CD via Bamboo with Helm) being successful and the UI reflecting the changes for 2.10.4. The error logs retrieved from the schedulers:

[2025-02-19T20:56:01.541+0000] {scheduler_job_runner.py:1016} ERROR - Exception when executing SchedulerJob._run_scheduler_loop 
Traceback (most recent call last):
...
[2025-02-19T20:56:01.543+0000] {kubernetes_executor.py:703} INFO - Shutting down Kubernetes executor
[2025-02-19T20:56:02.563+0000] {process_utils.py:132} INFO - Sending Signals.SIGTERM to group 35. PIDs of all processes in the group: [35]
[2025-02-19T20:56:02.564+0000] {process_utils.py:87} INFO - Sending the signal Signals.SIGTERM to group 35
[2025-02-19T20:56:02.736+0000] {process_utils.py:80} INFO - Process psutil.Process(pid=35, status='terminated', exitcode=0, started='20:56:00') (35) terminated with exit code 0
[2025-02-19T20:56:02.736+0000] {scheduler_job_runner.py:1029} INFO - Exited execute loop
Traceback (most recent call last):
  File "/home/airflow/.local/bin/airflow", line 8, in <module>
    sys.exit(main())
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/__main__.py", line 62, in main
    args.func(args)
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/cli/cli_config.py", line 49, in command
    return func(*args, **kwargs)
...
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/models/dag.py", line 4101, in dags_needing_dagruns
    if not dag_ready(dag_id, cond=ser_dag.dag.timetable.dataset_condition, statuses=statuses):
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/models/serialized_dag.py", line 235, in dag
    return SerializedDAG.from_dict(data)
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/serialization/serialized_objects.py", line 1785, in from_dict
    return cls.deserialize_dag(serialized_obj["dag"])
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/serialization/serialized_objects.py", line 1713, in deserialize_dag
    v = decode_timetable(v)
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/serialization/serialized_objects.py", line 339, in decode_timetable
    return timetable_class.deserialize(var[Encoding.VAR])
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/timetables/simple.py", line 181, in deserialize
    return cls(decode_dataset_condition(data["dataset_condition"]))
KeyError: 'dataset_condition'

We've updated the dependencies and checked for any incompatibility but were not able to locate any. I looked at the DAGs with datasets and timetables to see if that would be causing the issue, but everything seems to be referencing them in the correct context.

We eventually just restored the DB to the previous version (2.8.2) and re-ran the deployment to point to 2.9.3, which brought the environments back up.

I am seeking assistance and guidance on why the upgrade to 2.10.4 results in a failure when the same Dockerfile and requirements.txt work just fine in a local setting. I can provide further information on the setup if needed. Thank you.

[pingu1m]

Experiencing the same error.

Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/jobs/scheduler_job_runner.py", line 999, in _execute
    self._run_scheduler_loop()
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/jobs/scheduler_job_runner.py", line 1138, in _run_scheduler_loop
    num_queued_tis = self._do_scheduling(session)
                     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/jobs/scheduler_job_runner.py", line 1244, in _do_scheduling
    self._create_dagruns_for_dags(guard, session)
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/utils/retries.py", line 93, in wrapped_function
    for attempt in run_with_db_retries(max_retries=retries, logger=logger, **retry_kwargs):
                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.12/site-packages/tenacity/__init__.py", line 443, in __iter__
    do = self.iter(retry_state=retry_state)
         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.12/site-packages/tenacity/__init__.py", line 376, in iter
    result = action(retry_state)
             ^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.12/site-packages/tenacity/__init__.py", line 398, in <lambda>
    self._add_action_func(lambda rs: rs.outcome.result())
                                     ^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/concurrent/futures/_base.py", line 449, in result
    return self.__get_result()
           ^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/concurrent/futures/_base.py", line 401, in __get_result
    raise self._exception
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/utils/retries.py", line 102, in wrapped_function
    return func(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/jobs/scheduler_job_runner.py", line 1312, in _create_dagruns_for_dags
    query, dataset_triggered_dag_info = DagModel.dags_needing_dagruns(session)
                                        ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/models/dag.py", line 4101, in dags_needing_dagruns
    if not dag_ready(dag_id, cond=ser_dag.dag.timetable.dataset_condition, statuses=statuses):
                                  ^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/models/serialized_dag.py", line 235, in dag
    return SerializedDAG.from_dict(data)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/serialization/serialized_objects.py", line 1785, in from_dict
    return cls.deserialize_dag(serialized_obj["dag"])
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/serialization/serialized_objects.py", line 1713, in deserialize_dag
    v = decode_timetable(v)
        ^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/serialization/serialized_objects.py", line 339, in decode_timetable
    return timetable_class.deserialize(var[Encoding.VAR])
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/timetables/simple.py", line 181, in deserialize
    return cls(decode_dataset_condition(data["dataset_condition"]))
                                        ~~~~^^^^^^^^^^^^^^^^^^^^^
KeyError: 'dataset_condition'

[pingu1m]

My issue was fixed by itself after 30min. I was debugging when things started working. I believe it has to do with db sessions. Once things are reset it will work.

[nkjmsg]

Interesting, it lingered for 2+ days for me. I ended up restoring the DB from a backup, then downgraded to 2.9.3. That worked without issues. Did you have any DAGs running in your environment when the issue occurred?