[GitHub] [airflow] henriqueribeiro opened a new issue #20681: "already registered for DAG" when dynamically generate tasks

[GitBox <gi...@apache.org>]

henriqueribeiro opened a new issue #20681:
URL: https://github.com/apache/airflow/issues/20681


   ### Apache Airflow version
   
   2.1.4
   
   ### What happened
   
   Recently I started to use TaskFlow API in some of my dag files where the tasks are being dynamically generated and started to notice (a lot of) warning messages in the logs. The messages look like this:
   ```bash
   [2022-01-05 13:19:11,729] {baseoperator.py:1295} WARNING - Dependency <Task(_PythonDecoratedOperator): 2_num_group.run_2>, 2_num_group.agg already registered for DAG: test_warn_msg
   [2022-01-05 13:19:11,730] {baseoperator.py:1295} WARNING - Dependency <Task(_PythonDecoratedOperator): 2_num_group.agg>, 2_num_group.run_2 already registered for DAG: test_warn_msg
   [2022-01-05 13:19:11,730] {baseoperator.py:1295} WARNING - Dependency <Task(_PythonDecoratedOperator): 2_num_group.run_1>, 2_num_group.agg already registered for DAG: test_warn_msg
   [2022-01-05 13:19:11,731] {baseoperator.py:1295} WARNING - Dependency <Task(_PythonDecoratedOperator): 2_num_group.agg>, 2_num_group.run_1 already registered for DAG: test_warn_msg
   [2022-01-05 13:19:11,731] {baseoperator.py:1295} WARNING - Dependency <Task(_PythonDecoratedOperator): 2_num_group.run_2>, 2_num_group.agg already registered for DAG: test_warn_msg
   ```
   
   And these messages keep appearing even when the DAG is not running and also when it's paused. So I guess this happens at when parsing the DAG.
   
   ### What you expected to happen
   
   I'm not sure if I'm doing something wrong, but this messages should not appear.
   
   ### How to reproduce
   
   Here is an DAG example to reproduce this issue:
   
   ```python
   from airflow.decorators import dag, task
   from airflow.utils.dates import days_ago
   from datetime import timedelta
   from airflow.operators.dummy import DummyOperator
   from airflow.utils.task_group import TaskGroup
   
   
   NUMS = [1, 2]
   
   default_args = {
       "owner": "henrique",
       "depends_on_past": False,
       "email_on_failure": True,
       "email_on_retry": False,
       "retries": 3,
       "retry_delay": timedelta(minutes=15),
   }
   
   
   def print_id(num: int):
       print(num)
       return num
   
   
   def run_tests():
       results = []
       for i in NUMS:
           result = task(task_id=f"run_{i}")(print_id)(i)
           results.append(result)
   
       return results
   
   
   @task()
   def agg(results):
       print(results)
   
   
   @dag(
       "test_warn_msg",
       default_args=default_args,
       schedule_interval="@once",
       start_date=days_ago(1),
       max_active_runs=1,
   )
   def test_supervisor():
       task_start = DummyOperator(task_id="task_start")
       task_end = DummyOperator(task_id="task_end")
       groups = []
       for i in NUMS:
           with TaskGroup(group_id=f"{i}_num_group") as tg:
               results = run_tests()
               aggregation = agg(results)
   
               groups.append(tg)
   
       task_start >> groups >> task_end
   
   
   data_dag = test_supervisor()
   
   ```
   
   ### Operating System
   
   Cloud Composer
   
   ### Versions of Apache Airflow Providers
   
   _No response_
   
   ### Deployment
   
   Composer
   
   ### Deployment details
   
   _No response_
   
   ### Anything else
   
   _No response_
   
   ### Are you willing to submit PR?
   
   - [X] Yes I am willing to submit a PR!
   
   ### Code of Conduct
   
   - [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [airflow] boring-cyborg[bot] commented on issue #20681: "already registered for DAG" when dynamically generate tasks

[GitBox <gi...@apache.org>]

boring-cyborg[bot] commented on issue #20681:
URL: https://github.com/apache/airflow/issues/20681#issuecomment-1005684020


   Thanks for opening your first issue here! Be sure to follow the issue template!
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org