Schduler crash when delete a dag on UI.

[AndrewTsao]

Apache Airflow version

2.3.2 (latest released)

What happened

[2022-07-18 08:55:48,355] {dagbag.py:197} WARNING - Serialized DAG call_monitor no longer exists
[2022-07-18 08:55:48,356] {scheduler_job.py:1106} ERROR - Couldn't find dag call_monitor in DagBag/DB!
[2022-07-18 08:55:48,376] {scheduler_job.py:756} ERROR - Exception when executing SchedulerJob._run_scheduler_loop
Traceback (most recent call last):
  File "/home/xxxxx/.local/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 739, in _execute
    self._run_scheduler_loop()
  File "/home/xxxxx/.local/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 827, in _run_scheduler_loop
    num_queued_tis = self._do_scheduling(session)
  File "/home/xxxxx/.local/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 912, in _do_scheduling
    guard.commit()
  File "/home/xxxxx/.local/lib/python3.8/site-packages/airflow/utils/sqlalchemy.py", line 284, in commit
    self.session.commit()
  File "/home/xxxxx/.local/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 1423, in commit
    self._transaction.commit(_to_root=self.future)
  File "/home/xxxxx/.local/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 829, in commit
    self._prepare_impl()
  File "/home/xxxxx/.local/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 808, in _prepare_impl
    self.session.flush()
  File "/home/xxxxx/.local/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 3255, in flush
    self._flush(objects)
  File "/home/xxxxx/.local/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 3395, in _flush
    transaction.rollback(_capture_exception=True)
  File "/home/xxxxx/.local/lib/python3.8/site-packages/sqlalchemy/util/langhelpers.py", line 70, in __exit__
    compat.raise_(
  File "/home/xxxxx/.local/lib/python3.8/site-packages/sqlalchemy/util/compat.py", line 211, in raise_
    raise exception
  File "/home/xxxxx/.local/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 3355, in _flush
    flush_context.execute()
  File "/home/xxxxx/.local/lib/python3.8/site-packages/sqlalchemy/orm/unitofwork.py", line 453, in execute
    rec.execute(self)
  File "/home/xxxxx/.local/lib/python3.8/site-packages/sqlalchemy/orm/unitofwork.py", line 627, in execute
    util.preloaded.orm_persistence.save_obj(
  File "/home/xxxxx/.local/lib/python3.8/site-packages/sqlalchemy/orm/persistence.py", line 234, in save_obj
    _emit_update_statements(
  File "/home/xxxxx/.local/lib/python3.8/site-packages/sqlalchemy/orm/persistence.py", line 1032, in _emit_update_statements
    raise orm_exc.StaleDataError(
sqlalchemy.orm.exc.StaleDataError: UPDATE statement on table 'dag_run' expected to update 11 row(s); 0 were matched.
[2022-07-18 08:55:49,488] {process_utils.py:125} INFO - Sending Signals.SIGTERM to group 3261. PIDs of all processes in the group: [18519, 3261]
[2022-07-18 08:55:49,488] {process_utils.py:80} INFO - Sending the signal Signals.SIGTERM to group 3261
[2022-07-18 08:55:51,139] {process_utils.py:240} INFO - Waiting up to 5 seconds for processes to exit...
[2022-07-18 08:55:51,269] {process_utils.py:75} INFO - Process psutil.Process(pid=3261, status='terminated', exitcode=0, started='2022-07-14 16:23:21') (3261) terminated with exit code 0
[2022-07-18 08:55:51,271] {process_utils.py:75} INFO - Process psutil.Process(pid=18519, status='terminated', started='08:55:49') (18519) terminated with exit code None
[2022-07-18 08:55:51,272] {scheduler_job.py:768} INFO - Exited execute loop
Traceback (most recent call last):
  File "/home/xxxxx/.local/bin/airflow", line 8, in <module>
    sys.exit(main())
  File "/home/xxxxx/.local/lib/python3.8/site-packages/airflow/__main__.py", line 38, in main
    args.func(args)
  File "/home/xxxxx/.local/lib/python3.8/site-packages/airflow/cli/cli_parser.py", line 51, in command
    return func(*args, **kwargs)
  File "/home/xxxxx/.local/lib/python3.8/site-packages/airflow/utils/cli.py", line 99, in wrapper
    return f(*args, **kwargs)
  File "/home/xxxxx/.local/lib/python3.8/site-packages/airflow/cli/commands/scheduler_command.py", line 75, in scheduler
    _run_scheduler_job(args=args)
  File "/home/xxxxx/.local/lib/python3.8/site-packages/airflow/cli/commands/scheduler_command.py", line 46, in _run_scheduler_job
    job.run()
  File "/home/xxxxx/.local/lib/python3.8/site-packages/airflow/jobs/base_job.py", line 244, in run
    self._execute()
  File "/home/xxxxx/.local/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 739, in _execute
    self._run_scheduler_loop()
  File "/home/xxxxx/.local/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 827, in _run_scheduler_loop
    num_queued_tis = self._do_scheduling(session)
  File "/home/xxxxx/.local/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 912, in _do_scheduling
    guard.commit()
  File "/home/xxxxx/.local/lib/python3.8/site-packages/airflow/utils/sqlalchemy.py", line 284, in commit
    self.session.commit()
  File "/home/xxxxx/.local/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 1423, in commit
    self._transaction.commit(_to_root=self.future)
  File "/home/xxxxx/.local/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 829, in commit
    self._prepare_impl()
  File "/home/xxxxx/.local/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 808, in _prepare_impl
    self.session.flush()
  File "/home/xxxxx/.local/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 3255, in flush
    self._flush(objects)
  File "/home/xxxxx/.local/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 3395, in _flush
    transaction.rollback(_capture_exception=True)
  File "/home/xxxxx/.local/lib/python3.8/site-packages/sqlalchemy/util/langhelpers.py", line 70, in __exit__
    compat.raise_(
  File "/home/xxxxx/.local/lib/python3.8/site-packages/sqlalchemy/util/compat.py", line 211, in raise_
    raise exception
  File "/home/xxxxx/.local/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 3355, in _flush
    flush_context.execute()
  File "/home/xxxxx/.local/lib/python3.8/site-packages/sqlalchemy/orm/unitofwork.py", line 453, in execute
    rec.execute(self)
  File "/home/xxxxx/.local/lib/python3.8/site-packages/sqlalchemy/orm/unitofwork.py", line 627, in execute
    util.preloaded.orm_persistence.save_obj(
  File "/home/xxxxx/.local/lib/python3.8/site-packages/sqlalchemy/orm/persistence.py", line 234, in save_obj
    _emit_update_statements(
  File "/home/xxxxx/.local/lib/python3.8/site-packages/sqlalchemy/orm/persistence.py", line 1032, in _emit_update_statements
    raise orm_exc.StaleDataError(
sqlalchemy.orm.exc.StaleDataError: UPDATE statement on table 'dag_run' expected to update 11 row(s); 0 were matched.

What you think should happen instead

Scheduler can't crash by DAGs's operation like delete a DAG.

How to reproduce

1 change a DAG dag_id in the DAG script.
2 delete old DAG on UI.

Operating System

CentOS Linux release 7.9.2009 (Core)

Versions of Apache Airflow Providers

airflow-code-editor                      5.2.2
apache-airflow                           2.3.2
apache-airflow-providers-celery          3.0.0
apache-airflow-providers-ftp             3.0.0
apache-airflow-providers-http            3.0.0
apache-airflow-providers-imap            3.0.0
apache-airflow-providers-microsoft-psrp  2.0.0
apache-airflow-providers-microsoft-winrm 3.0.0
apache-airflow-providers-mysql           3.0.0
apache-airflow-providers-redis           3.0.0
apache-airflow-providers-samba           4.0.0
apache-airflow-providers-sftp            3.0.0
apache-airflow-providers-sqlite          3.0.0
apache-airflow-providers-ssh             3.0.0

Deployment

Virtualenv installation

Deployment details

No response

Anything else

Once.

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

[potiuk]

Converted it into discussion as this is kind of expected.

Until versioning is implemented, similar problems are expected. We are aware about some of the edge cases like this, but we want to handle them with https://cwiki.apache.org/confluence/display/AIRFLOW/AIP-36+DAG+Versioning - currently you are not supposed to change the dag_id and delete the old one. Airflow does not support versions of the dag when you change or rename them. If you want to rename the dag, Delete it first, copy the old DAG into new file changing the id and then remove the old file.

Currently this paragraph describe your DAG evolution: https://airflow.apache.org/docs/apache-airflow/stable/index.html

Workflows are expected to be mostly static or slowly changing. You can think of the structure of the tasks in your workflow as slightly more dynamic than a database structure would be. Airflow workflows are expected to look similar from a run to the next, this allows for clarity around unit of work and continuity.