
**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

tests/operators/test_dagrun_operator.py
#
#
#
import pathlib
import tempfile
from datetime import datetime
from unittest import TestCase
from airflow.exceptions import AirflowException, DagRunAlreadyExists
from airflow.models import DAG, DagBag, DagModel, DagRun, Log, TaskInstance
from airflow.models.serialized_dag import SerializedDagModel
from airflow.operators.dagrun_operator import TriggerDagRunOperator
from airflow.utils import timezone
from airflow.utils.session import create_session
from airflow.utils.state import State
DEFAULT_DATE = datetime(2019, 1, 1, tzinfo=timezone.utc)
TEST_DAG_ID = "testdag"
TRIGGERED_DAG_ID = "triggerdag"
DAG_SCRIPT = (
    "from datetime import datetime\n\n"
    "from airflow.models import DAG\n"
    "from airflow.operators.dummy_operator import DummyOperator\n\n"
    "dag = DAG(\n"
    'dag_id="{dag_id}", \n'
    'default_args={{"start_date": datetime(2019, 1, 1)}}, \n'
    "schedule_interval=None,\n"
    ")\n"
    'task = DummyOperator(task_id="test", dag=dag)'
).format(dag_id=TRIGGERED_DAG_ID)
class TestDagRunOperator(TestCase):