<!--

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.

-->

**Description**

When using decorators to define dags, e.g. dag_1.py:

```python
from airflow.decorators import dag, task
from airflow.utils.dates import days_ago

DEFAULT_ARGS = {
    "owner": "airflow",
}


@task
def some_task():
    pass


@dag(
    default_args=DEFAULT_ARGS,
    schedule_interval=None,
    start_date=days_ago(2),
)
def my_dag():
    some_task()


DAG_1 = my_dag()

```

and

dag_2.py:

```python
from airflow.decorators import dag, task
from airflow.utils.dates import days_ago

DEFAULT_ARGS = {
    "owner": "airflow",
}


@task
def some_other_task():
    pass


@dag(
    default_args=DEFAULT_ARGS,
    schedule_interval=None,
    start_date=days_ago(2),
)
def my_dag():
    some_other_task()


DAG_2 = my_dag()

```

We have two different dags which have been written in isolation, but by sheer bad luck both define `my_dag()`. This seems fine for each file in isolation, but on the airflow UI, we only end up seeing one entry for `my_dag`, where it has picked up `dag_1.py` and ignored `dag_2.py`.

**Use case / motivation**

We currently end up with only one DAG showing up on the UI, and no indication as to why the other one hasn't appeared.

Suggestion: popup similar to 'DAG import error' to highlight what needs changing in one of the DAG files in order for both to show up ("DAG import error: duplicate dag names found - please review {duplicate files} and ensure all dag definitions are unique"?)

**Are you willing to submit a PR?**

No time to spare on this at present

**Related Issues**

I haven't found any related issues with the search function.
tests/models/test_dagbag.py
"""
        Test that Serialized DAG is updated in DagBag when it is updated in
        Serialized DAG table after 'min_serialized_dag_fetch_interval' seconds are passed.
        """
        with freeze_time(tz.datetime(2020, 1, 5, 0, 0, 0)):
            example_bash_op_dag = DagBag(include_examples=True).dags.get("example_bash_operator")
            SerializedDagModel.write_dag(dag=example_bash_op_dag)
            dag_bag = DagBag(read_dags_from_db=True)
            ser_dag_1 = dag_bag.get_dag("example_bash_operator")
            ser_dag_1_update_time = dag_bag.dags_last_fetched["example_bash_operator"]
            assert example_bash_op_dag.tags == ser_dag_1.tags
            assert ser_dag_1_update_time == tz.datetime(2020, 1, 5, 0, 0, 0)
        
        
        with freeze_time(tz.datetime(2020, 1, 5, 0, 0, 4)):
            with assert_queries_count(0):
                assert dag_bag.get_dag("example_bash_operator").tags == ["example", "example2"]
        
        with freeze_time(tz.datetime(2020, 1, 5, 0, 0, 6)):
            example_bash_op_dag.tags += ["new_tag"]
            SerializedDagModel.write_dag(dag=example_bash_op_dag)
        
        
        with freeze_time(tz.datetime(2020, 1, 5, 0, 0, 8)):
            with assert_queries_count(2):
                updated_ser_dag_1 = dag_bag.get_dag("example_bash_operator")
                updated_ser_dag_1_update_time = dag_bag.dags_last_fetched["example_bash_operator"]
        assert set(updated_ser_dag_1.tags) == {"example", "example2", "new_tag"}
        assert updated_ser_dag_1_update_time > ser_dag_1_update_time
    def test_collect_dags_from_db(self):