
**Apache Airflow version**: 2.0.0b3


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

**Environment**: MacOS with Docker

- **Cloud provider or hardware configuration**:
- **OS** (e.g. from /etc/os-release):
- **Kernel** (e.g. `uname -a`):
- **Install tools**:
- **Others**:

**What happened**:

```
Traceback (most recent call last):
  File "/usr/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 984, in _run_raw_task
    result = task_copy.execute(context=context)
  File "/usr/local/airflow/plugins/custom_trigger_operator.py", line 138, in execute
    dag_run.execution_date,
UnboundLocalError: local variable 'dag_run' referenced before assignment
```

**What you expected to happen**:

The dag_run object doesn't exist in https://github.com/apache/airflow/blob/94ba200d42403d1067a63e2e6b158d893c2e8b5a/airflow/operators/dagrun_operator.py#L161
as the DagRunAlreadyExists exception is raised

**How to reproduce it**:

```
first_task = DummyOperator(...)

trigger_task = TriggerDagRunOperator(...
reset_dag_run=True,
wait_for_completion=True,
trigger_dag_id='my_dag_to_trigger'
...)

```
Run the DAG above once. It success. Clear the task first_task to run a second time the DAG and so trigger again my_dag_to_trigger, you will get the error.

**Anything else we need to know**:

Nope

airflow/operators/dagrun_operator.py
"""
    Triggers a DAG run for a specified ``dag_id``
    :param trigger_dag_id: the dag_id to trigger (templated)
    :type trigger_dag_id: str
    :param conf: Configuration for the DAG run
    :type conf: dict
    :param execution_date: Execution date for the dag (templated)
    :type execution_date: str or datetime.datetime
    :param reset_dag_run: Whether or not clear existing dag run if already exists.
        This is useful when backfill or rerun an existing dag run.
        When reset_dag_run=False and dag run exists, DagRunAlreadyExists will be raised.
        When reset_dag_run=True and dag run exists, existing dag run will be cleared to rerun.
    :type reset_dag_run: bool
    :param wait_for_completion: Whether or not wait for dag run completion. (default: False)
    :type wait_for_completion: bool
    :param poke_interval: Poke interval to check dag run status when wait_for_completion=True.
        (default: 60)
    :type poke_interval: int
    :param allowed_states: list of allowed states, default is ``['success']``
    :type allowed_states: list
    :param failed_states: list of failed or dis-allowed states, default is ``None``
    :type failed_states: list
    """
    template_fields = ("trigger_dag_id", "execution_date", "conf")
    ui_color = "#ffefeb"
    @property
    def operator_extra_links(self):