**Apache Airflow version**:
2.0 / master

**Environment**:

breeze

**What happened**:

Using this DAG:

```py
from airflow.operators.bash import BashOperator
from airflow.operators.python import task
from airflow.models import DAG
from airflow.utils.task_group import TaskGroup

@task
def show():
    print("Cats are awesome!")

with DAG(
    "using_task_group",
    default_args={'owner': 'airflow'},
    start_date=days_ago(2),
    schedule_interval=None,
) as dag3:
    start_task = BashOperator(
        task_id="start_task",
        bash_command="echo start",
    )

    end_task = BashOperator(
        task_id="end_task",
        bash_command="echo end",
    )

    with TaskGroup(group_id="show_tasks") as tg1:
        previous_show = show()
        for _ in range(100):
            next_show = show()
            previous_show >> next_show
            previous_show = next_show
```
I get:
```
Broken DAG: [/files/dags/test.py] Traceback (most recent call last):
  File "/opt/airflow/airflow/models/baseoperator.py", line 410, in __init__
    task_group.add(self)
  File "/opt/airflow/airflow/utils/task_group.py", line 140, in add
    raise DuplicateTaskIdFound(f"Task id '{key}' has already been added to the DAG")
airflow.exceptions.DuplicateTaskIdFound: Task id 'show_tasks.show' has already been added to the DAG
```

If I remove the task group the task are generated as expected.

**What you expected to happen**:

I expect to be able to generate tasks dynamically using TaskGroup and task decoratos.

**How to reproduce it**:

Use the DAG from above.

**Anything else we need to know**:

N/A

tests/operators/test_python.py
short_op = ShortCircuitOperator(task_id='make_choice', dag=dag, python_callable=lambda: value)
        branch_1 = DummyOperator(task_id='branch_1', dag=dag)
        branch_1.set_upstream(short_op)
        branch_2 = DummyOperator(task_id='branch_2', dag=dag)
        branch_2.set_upstream(branch_1)
        upstream = DummyOperator(task_id='upstream', dag=dag)
        upstream.set_downstream(short_op)
        dag.clear()
        logging.error("Tasks %s", dag.tasks)
        dr = dag.create_dagrun(
            run_type=DagRunType.MANUAL,
            start_date=timezone.utcnow(),
            execution_date=DEFAULT_DATE,
            state=State.RUNNING,
        )
        upstream.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE)
        short_op.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE)
        tis = dr.get_task_instances()
        self.assertEqual(len(tis), 4)
        for ti in tis:
            if ti.task_id == 'make_choice':
                self.assertEqual(ti.state, State.SUCCESS)
            elif ti.task_id == 'upstream':
                self.assertEqual(ti.state, State.SUCCESS)
            elif ti.task_id == 'branch_1' or ti.task_id == 'branch_2':
                self.assertEqual(ti.state, State.SKIPPED)
            else:
                raise ValueError(f'Invalid task id {ti.task_id} found!')
        value = True
        dag.clear()