get_dagrun() function starts giving error in 2.0 when using inside cluster policy

[sudarshan2906]

Apache Airflow version: 2.0

• OS (e.g. from /etc/os-release):

PRETTY_NAME="Debian GNU/Linux 10 (buster)"
NAME="Debian GNU/Linux"
VERSION_ID="10"
VERSION="10 (buster)"
VERSION_CODENAME=buster
ID=debian

What happened:
I am using a task_instance_mutation_hook cluster policy and inside that using get_dagrun() function from task_instance object to get the dag run object for the task instance. This was working fine in 1.10.14 but it started giving following error in 2.0 due to which scheduler is not starting.

dag_run = task_instance.get_dagrun()
  File "/usr/local/lib/python3.7/site-packages/airflow/utils/session.py", line 65, in wrapper
    return func(*args, session=session, **kwargs)
  File "/usr/local/lib/python3.7/contextlib.py", line 119, in __exit__
    next(self.gen)
  File "/usr/local/lib/python3.7/site-packages/airflow/utils/session.py", line 32, in create_session
    session.commit()
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/orm/session.py", line 1042, in commit
    self.transaction.commit()
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/orm/session.py", line 504, in commit
    self._prepare_impl()
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/orm/session.py", line 472, in _prepare_impl
    self.session.dispatch.before_commit(self.session)
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/event/attr.py", line 322, in __call__
    fn(*args, **kw)
  File "/usr/local/lib/python3.7/site-packages/airflow/utils/sqlalchemy.py", line 217, in _validate_commit
    raise RuntimeError("UNEXPECTED COMMIT - THIS WILL BREAK HA LOCKS!")

I am using the dag run object to get the conf passed to the dag run and I am setting some properties of the task_instance according to it.

How to reproduce it:
Example of the cluster policy used:

def task_instance_mutation_hook(task_instance):
    dag_run = task_instance.get_dagrun()
    conf = dag_run.conf

[kaxil]

A workaround for this, for now, is to pass a session to get_dagrun method:

from airflow.settings import Session

session = Session()

def task_instance_mutation_hook(task_instance):
    dag_run = task_instance.get_dagrun(session=session)
    conf = dag_run.conf

[sudarshan2906]

@kaxil
I tried the workaround. But getting this error:

Traceback (most recent call last):
  File "/usr/local/bin/airflow", line 5, in <module>
    from airflow.__main__ import main
  File "/usr/local/lib/python3.7/site-packages/airflow/__init__.py", line 46, in <module>
    settings.initialize()
  File "/usr/local/lib/python3.7/site-packages/airflow/settings.py", line 430, in initialize
    import_local_settings()
  File "/usr/local/lib/python3.7/site-packages/airflow/settings.py", line 400, in import_local_settings
    import airflow_local_settings
  File "/root/airflow/config/airflow_local_settings.py", line 7, in <module>
    session = Session()
TypeError: 'NoneType' object is not callable

[kaxil]

Try with the following to delay import and creating a session in the mutation hook

def task_instance_mutation_hook(task_instance):
    from airflow.settings import Session
    session = Session()
    dag_run = task_instance.get_dagrun(session=session)
    conf = dag_run.conf

[kaxil]

@sudarshan2906 Did it work?

[sudarshan2906]

@kaxil Sorry for the late reply.
Yes this worked :)

[nicolasge]

@sudarshan2906 does the code still work for you in Airflow 2.10.1? I tried the same code in my environment but it failed