<!--

Welcome to Apache Airflow!  For a smooth issue process, try to answer the following questions.
Don't worry if they're not all applicable; just try to include what you can :-)

If you need to include code snippets or logs, please put them in fenced code
blocks.  If they're super-long, please use the details tag like
<details><summary>super-long log</summary> lots of stuff </details>

Please delete these comment blocks before submitting the issue.

-->

<!--

IMPORTANT!!!

PLEASE CHECK "SIMILAR TO X EXISTING ISSUES" OPTION IF VISIBLE
NEXT TO "SUBMIT NEW ISSUE" BUTTON!!!

PLEASE CHECK IF THIS ISSUE HAS BEEN REPORTED PREVIOUSLY USING SEARCH!!!

Please complete the next sections or the issue will be closed.
These questions are the first thing we need to know to understand the context.

-->

**Apache Airflow version**: 1.10.12 (also tested with 1.10.15, 2.0.2 but less extensively)


**Kubernetes version (if you are using kubernetes)** (use `kubectl version`): None

**Environment**: Linux / Docker

- **Cloud provider or hardware configuration**: None
- **OS** (e.g. from /etc/os-release): Red Hat Enterprise Linux Server 7.9 (Maipo)
- **Kernel** (e.g. `uname -a`): Linux d7b9410c0f25 4.19.104-microsoft-standard #1 SMP Wed Feb 19 06:37:35 UTC 2020 x86_64 x86_64 x86_64 GNU/Linux 
- **Install tools**: 
- **Others**: Tested on both real Linux (Red Hat) and a docker inside a windows machine.

**What happened**: Custom `task_instance_mutation_hook` not called by scheduler even though airflow loads the `airflow_local_settings` module.

**What you expected to happen**:
`task_instance_mutation_hook` called before every task instance run. 
I think the way `airflow.models.dagrun` loads `task_instance_mutation_hook` from `airflow_local_settings` does not work when `airflow_local_settings` imports `airflow.models.taskinstance` or `airflow.models.dagrun`.

**How to reproduce it**:

1. Added `airflow_local_settings.py` to \{AIRFLOW_HOME\}\config
```python
import logging
from airflow.models.taskinstance import TaskInstance

def task_instance_mutation_hook(ti: TaskInstance):
    logging.getLogger("").warning("HERE IN  task_instance_mutation_hook log")
    print("HERE IN  task_instance_mutation_hook")
    ti.queue = "X"
```

2. See output `[2021-05-06 11:13:04,076] {settings.py:392} INFO - Loaded airflow_local_settings from /usr/local/airflow/config/airflow_local_settings.py.`
3. function is never called - log/print is not written and queue does not update.
4. Additionally, example code to reproduce code issue:

```python
import airflow
import airflow.models.dagrun

import inspect
print(inspect.getfile(airflow.settings.task_instance_mutation_hook))
print(inspect.getfile(airflow.models.dagrun.task_instance_mutation_hook))
```
outputs 
```
/usr/local/airflow/config/airflow_local_settings.py
/opt/bb/lib/python3.7/site-packages/airflow/settings.py
```

5. when removing `from airflow.models.taskinstance import TaskInstance` from airflow_local_settings.py everything works as expected.

**Anything else we need to know**:

BTW, do the logs printed from `task_instance_mutation_hook` go anywhere? Even after I remove the import and the queue is update, I can't see anything in the logs files or in the scheduler console. 

airflow/models/dagrun.py
count += (
                session.query(TI)
                .filter(
                    TI.dag_id == self.dag_id,
                    TI.execution_date == self.execution_date,
                    TI.task_id.in_(schedulable_ti_ids),
                )
                .update({TI.state: State.SCHEDULED}, synchronize_session=False)
            )
        
        if dummy_ti_ids:
            count += (
                session.query(TI)
                .filter(
                    TI.dag_id == self.dag_id,
                    TI.execution_date == self.execution_date,
                    TI.task_id.in_(dummy_ti_ids),
                )
                .update(
                    {
                        TI.state: State.SUCCESS,
                        TI.start_date: timezone.utcnow(),
                        TI.end_date: timezone.utcnow(),
                        TI.duration: 0,
                    },
                    synchronize_session=False,
                )
            )
        return count