Show Deprecation warning on duplicate Task ids

[kaxil]

We already raise the warning at https://github.com/apache/airflow/blob/1.10.10/airflow/models/dag.py#L1318-L1328 but that wasn't enough.

We raise an error for Airflow 2.0 . PR where we did that: https://github.com/apache/airflow/pull/6549/files

Before:

airflow ❯ python -c "import airflow; from airflow.operators.dummy_operator import DummyOperator; dag = airflow.DAG(dag_id='test', start_date=airflow.utils.timezone.utcnow()); t1 = DummyOperator(dag=dag, task_id='t1'); t1_2 = DummyOperator(dag=dag, task_id='t1')"

After:

airflow ❯ python -c "import airflow; from airflow.operators.dummy_operator import DummyOperator; dag = airflow.DAG(dag_id='test', start_date=airflow.utils.timezone.utcnow()); t1 = DummyOperator(dag=dag, task_id='t1'); t1_2 = DummyOperator(dag=dag, task_id='t1')"
/opt/airflow/airflow/models/baseoperator.py:555: PendingDeprecationWarning: The requested task could not be added to the DAG because a task with task_id t1 is already in the DAG. Starting in Airflow 2.0, trying to overwrite a task will raise an exception.
  category=PendingDeprecationWarning)

---

Make sure to mark the boxes below before creating PR: [x]

• Description above provides context of the change
• Unit tests coverage for changes (not needed for documentation changes)
• Target Github ISSUE in description if exists
• Commits follow "How to write a good git commit message"
• Relevant documentation is updated including usage instructions.
• I will engage committers as explained in Contribution Workflow Example.

---

In case of fundamental code change, Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in UPDATING.md.
Read the Pull Request Guidelines for more information.

[jhtimmins]

Looks good overall. Basically two things worth noting.

1. A few naming ambiguities in the tests that could be clarified.
2. The tests are currently testing the content of warning message strings. This is making them brittle.

[jhtimmins]

Testing for the exact response message is likely to make this case brittle. Here are some possible alternatives:

1. Create a class constant on Dag that stores this string, and use it both in the implementation and the test.
2. Same as 1, but instead put it in a warnings.py file or similar.
3. Same as 1, but add some type of Warning or Alert mixin.
4. Create a child class of PendingDeprecationWarning that encompasses the message body, then just text for the Exception type.
5. Only modify the test and check for the presence of t1 and 2.0.

4 probably makes the most sense. Can add 5 to it for extra safety that string interpolation is working.

[kaxil]

The main reason we test for the exact same message is otherwise, even if someone changes the message in warnings.py the test would still pass.

[kaxil]

CI failures are unrelated. @potiuk any idea?

[potiuk]

Will be fixed by #8758

[ashb]

Do we want dag.task_dict[self.task_id] != self, or dag.task_dict[self.task_id] is not self.

The former use __eq__, the later checks for exactly the same object.

[kaxil]

Updated, I will update Airflow Master as we currently raise errors after == check

https://github.com/kaxil/airflow/blob/c3a46b9fec618489314a4ac8d4be289481e8edd3/airflow/models/baseoperator.py#L603-L606

[ashb]

Only cos it's related to this PR, not that it's a bug with this PR

And I think I've just found a problem in our example dags caused by us using != in master:

airflow/airflow/providers/google/cloud/example_dags/example_gcs.py

Lines 129 to 130 in 6e4f5fa

	delete_bucket_1 = GCSDeleteBucketOperator(task_id="delete_bucket", bucket_name=BUCKET_1)
	delete_bucket_2 = GCSDeleteBucketOperator(task_id="delete_bucket", bucket_name=BUCKET_2)

[tooptoop4]

@kaxil this is causing issue for me if i have same taskid in 2 different dags

[kaxil]

@kaxil this is causing issue for me if i have same taskid in 2 different dags

It should not show you warning for different dags, unless you are using context manager and have as dag: for both of your DAGs

[tooptoop4]

@kaxil raised #11354

[JeffryMAC]

@kaxil I use context manager and as dag for every DAG I have.
the "as dag" is not the dag_id. why it matters?

[kaxil]

@kaxil I use context manager and as dag for every DAG I have.
the "as dag" is not the dag_id. why it matters?

You are right, in theory it should not, although I roughly remember I had to change:

with DAG(dag_id='d1') as dag:
     t1 = BashOperator(...)

with DAG(dag_id='d2') as dag:
     t1 = BashOperator(...)

to

with DAG(dag_id='d1') as dag_1:
     t1 = BashOperator(...)

with DAG(dag_id='d2') as dag_2:
     t1 = BashOperator(...)

Can't remember why and when though