"already registered for DAG" when dynamically generate tasks

[henriqueribeiro]

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:

[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:

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?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[boring-cyborg[bot]]

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

[potiuk]

The error is correct - decorated callable is a single "task" and you try to addi to dag multiple times.

If you are adding it in a loop you are adding the same task (named by the function name) muiltiple times. You cannot make multiple tasks with single decorator this way. There is an AIP-42 in progess https://cwiki.apache.org/confluence/display/AIRFLOW/AIP-42+Dynamic+Task+Mapping tha add capability of even more flexibility - you will be able to fan-out <> fan-in variable number of similar tasks but currently I do not think there is an easy way to use task_flow for that. I think classic operators are better for that purpose for now - also if you have essentially "fixed" set of tasks..

[henriqueribeiro]

I understand your point, but giving the tasks a unique name, I still get the same error. Here is the new dag

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(num):
    results = []
    for i in NUMS:
        result = task(task_id=f"run_{num}_{i}")(print_id)(i)
        results.append(result)

    return results


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(i)
            aggregation = task(task_id=f"{i}_agg")(agg)(results)

            groups.append(tg)

    task_start >> groups >> task_end


data_dag = test_supervisor()

[potiuk]

You can't use decorator like that :).

But maybe (I am not 100%) that woudl work (when you decorate agg with @task

aggregation =agg(task_id='<unique_id>', results)

Never tried it but I think it should work actually (not sure if the decorator will strip the task_id but I think so).

[henriqueribeiro]

I'm pretty sure that it's possible to use the decorator as I did. anyway I can try your approach.

[henriqueribeiro]

I just confirmed that your approach doesn't work. The approach that I suggested is giving the same error:

Dependency <Task(_PythonDecoratedOperator): 1_num_group.1_agg>, 1_num_group.run_1_2 already registered for DAG: test_warn_msg
Dependency <Task(_PythonDecoratedOperator): 1_num_group.run_1_1>, 1_num_group.1_agg already registered for DAG: test_warn_msg
Dependency <Task(_PythonDecoratedOperator): 1_num_group.1_agg>, 1_num_group.run_1_1 already registered for DAG: test_warn_msg
Dependency <Task(_PythonDecoratedOperator): 1_num_group.run_1_2>, 1_num_group.1_agg already registered for DAG: test_warn_msg
Dependency <Task(_PythonDecoratedOperator): 1_num_group.1_agg>, 1_num_group.run_1_2 already registered for DAG: test_warn_msg
Dependency <Task(_PythonDecoratedOperator): 2_num_group.run_2_1>, 2_num_group.2_agg already registered for DAG: test_warn_msg
Dependency <Task(_PythonDecoratedOperator): 2_num_group.2_agg>, 2_num_group.run_2_1 already registered for DAG: test_warn_msg

[potiuk]

Then I have no idea.

[adh-wonolo]

Just wanted to chime in here that I was investigating something similar and was trying to cut things down significantly to further investigate on 2.2.3. Using an example right from the docs I am able to get these warnings to show up without even generating a task group/dynamically generating anything:

 import json
 from datetime import datetime

 from airflow.providers.http.operators.http import SimpleHttpOperator
 from airflow.decorators import dag, task


 @dag(schedule_interval=None, start_date=datetime(2021, 1, 1), catchup=False, tags=['example'])
 def my_dag():
     get_api_results_task = SimpleHttpOperator(
         task_id="get_api_results",
         endpoint="/api/query",
         do_xcom_push=True,
         http_conn_id="http",
     )


     @task
     def parse_results(api_results):
         return "test"


     parsed_results = parse_results(api_results=get_api_results_task.output)

 dag = my_dag()

Running a simple test: airflow tasks test my_dag parse_results 2022-02-01
Includes a bunch of these warnings:

[2022-02-15 17:02:05,945] {baseoperator.py:1393} WARNING - Dependency <Task(SimpleHttpOperator): get_api_results>, parse_results already registered for DAG: my_dag
[2022-02-15 17:02:05,945] {baseoperator.py:1393} WARNING - Dependency <Task(_PythonDecoratedOperator): parse_results>, get_api_results already registered for DAG: my_dag
[2022-02-15 17:02:05,945] {baseoperator.py:1393} WARNING - Dependency <Task(SimpleHttpOperator): get_api_results>, parse_results already registered for DAG: my_dag
[2022-02-15 17:02:05,945] {baseoperator.py:1393} WARNING - Dependency <Task(_PythonDecoratedOperator): parse_results>, get_api_results already registered for DAG: my_dag
[2022-02-15 17:02:05,946] {baseoperator.py:1393} WARNING - Dependency <Task(SimpleHttpOperator): get_api_results>, parse_results already registered for DAG: my_dag
[2022-02-15 17:02:05,946] {baseoperator.py:1393} WARNING - Dependency <Task(_PythonDecoratedOperator): parse_results>, get_api_results already registered for DAG: my_dag
[2022-02-15 17:02:05,946] {baseoperator.py:1393} WARNING - Dependency <Task(SimpleHttpOperator): get_api_results>, parse_results already registered for DAG: my_dag
[2022-02-15 17:02:05,946] {baseoperator.py:1393} WARNING - Dependency <Task(_PythonDecoratedOperator): parse_results>, get_api_results already registered for DAG: my_dag
[2022-02-15 17:02:05,946] {baseoperator.py:1393} WARNING - Dependency <Task(SimpleHttpOperator): get_api_results>, parse_results already registered for DAG: my_dag
[2022-02-15 17:02:05,946] {baseoperator.py:1393} WARNING - Dependency <Task(_PythonDecoratedOperator): parse_results>, get_api_results already registered for DAG: my_dag

If I don't reference something upstream (eg pass None to parse_results) the warning doesn't appear.

I'll note that this doesn't effect the dags from functioning.

[potiuk]

It's likely the example might be somewhat misleading @turbaszek @dstandish @josh-fell ?

The examples here
https://airflow.apache.org/docs/apache-airflow/2.2.3/tutorial_taskflow_api.html#consuming-xcoms-with-decorated-tasks-from-regular-tasks

Looks a bit out-of-context and I belive the way how they are added by @adh-wonolo is wrong - maybe worth to add a more complete /full dag examples to show how it could look like

[adh-wonolo]

Yep, I'm just following the example verbatim in https://airflow.apache.org/docs/apache-airflow/2.2.3/tutorial_taskflow_api.html#consuming-xcoms-between-decorated-and-traditional-tasks if that's not accurate I'd be happy to be pointed to the correct way to accomplish this so I can fix up all of the DAGs we've build that are impacted by this.

[adh-wonolo]

Just a fast follow up here, I ran the example https://github.com/apache/airflow/blob/main/airflow/example_dags/example_dag_decorator.py and the output is the same (IP replaced with 0.0.0.0):

AIRFLOW_CTX_DAG_OWNER=airflow
AIRFLOW_CTX_DAG_ID=example_dag_decorator
AIRFLOW_CTX_TASK_ID=prepare_email
AIRFLOW_CTX_EXECUTION_DATE=2022-02-19T23:04:28.485498+00:00
AIRFLOW_CTX_DAG_RUN_ID=manual__2022-02-19T23:04:28.485498+00:00
[2022-02-19, 18:04:36 UTC] {python.py:175} INFO - Done. Returned value was: {'subject': 'Server connected from 0.0.0.0', 'body': 'Seems like today your server executing Airflow is connected from IP 0.0.0.0<br>'}
[2022-02-19, 18:04:36 UTC] {taskinstance.py:1267} INFO - Marking task as SUCCESS. dag_id=example_dag_decorator, task_id=prepare_email, execution_date=20220219T230428, start_date=20220219T230435, end_date=20220219T230436
[2022-02-19, 18:04:36 UTC] {local_task_job.py:154} INFO - Task exited with return code 0
[2022-02-19, 18:04:36 UTC] {baseoperator.py:1393} WARNING - Dependency <Task(_PythonDecoratedOperator): prepare_email>, send_email already registered for DAG: example_dag_decorator
[2022-02-19, 18:04:36 UTC] {baseoperator.py:1393} WARNING - Dependency <Task(EmailOperator): send_email>, prepare_email already registered for DAG: example_dag_decorator
[2022-02-19, 18:04:36 UTC] {baseoperator.py:1393} WARNING - Dependency <Task(_PythonDecoratedOperator): prepare_email>, send_email already registered for DAG: example_dag_decorator
[2022-02-19, 18:04:36 UTC] {baseoperator.py:1393} WARNING - Dependency <Task(EmailOperator): send_email>, prepare_email already registered for DAG: example_dag_decorator
[2022-02-19, 18:04:36 UTC] {baseoperator.py:1393} WARNING - Dependency <Task(_PythonDecoratedOperator): prepare_email>, send_email already registered for DAG: example_dag_decorator
[2022-02-19, 18:04:36 UTC] {baseoperator.py:1393} WARNING - Dependency <Task(EmailOperator): send_email>, prepare_email already registered for DAG: example_dag_decorator
[2022-02-19, 18:04:36 UTC] {baseoperator.py:1393} WARNING - Dependency <Task(_PythonDecoratedOperator): prepare_email>, send_email already registered for DAG: example_dag_decorator
[2022-02-19, 18:04:36 UTC] {baseoperator.py:1393} WARNING - Dependency <Task(EmailOperator): send_email>, prepare_email already registered for DAG: example_dag_decorator
[2022-02-19, 18:04:36 UTC] {baseoperator.py:1393} WARNING - Dependency <Task(_PythonDecoratedOperator): prepare_email>, send_email already registered for DAG: example_dag_decorator
[2022-02-19, 18:04:36 UTC] {baseoperator.py:1393} WARNING - Dependency <Task(EmailOperator): send_email>, prepare_email already registered for DAG: example_dag_decorator
[2022-02-19, 18:04:36 UTC] {local_task_job.py:264} INFO - 1 downstream tasks scheduled from follow-on schedule check

[lyso]

I think I found the root cause of this WARNING message.

This WARNING message is only printed when XcomArg is consumed as an argument by a Traditional Operators (e.g. Python Operator) . For example:

@task
def get_a_number():
    return 123

PythonOperator(
        task_id="print_a_number",
        python_callable=lambda x: print(x),
        op_args=(get_a_number(),)
    )

The WARNING message is printed out when mini scheduler getting a partial deep copy of the dag. Tasks get deep copied one by one.

When deep copy a TaskInstance, the downstream relationship was re-built. Because the relationship existed, the WARNING message was printed out.

Why the TaskFlow API Task (a.k.a DecoratedOperator) will not print the WARNING?
When a TaskFlow API Task (TFAT) take the XcomArg as input, there will be no WARNING message. For example:

@task
def get_a_number():
    return 123

@task
def print_a_number(x):
    print(x)

print_a_number(get_a_number())

This is because TFAT will generate a new unique task_id if the duplicated task_id detected. This could avoid the WARNING, as there is not relationship to the new task_id.

How to fix this issue

This is just a WARNING message, we could ignore if it does not cause any trouble.

But it is reported the WARNING message could be too massive to be ignored in a relatively complicated DAG in AWS MWAA. A large number of this WARNING could bring significant workload on Workers and CloudWatch. Therefore, it is quite necessary to fix this issue.

I dont know what is the best way to fix it permanently in Airflow. Just list some of my ideas:

1. Simply remove the logging this WARNING, or reduce the level of log to DEBUG.
2. Change deep copy to shallow copy in partial_subset()

Ways to avoid in authoring DAG

1. Change all Operator to TaskFlow API or DecoratedOperator. (Is there ways to declare provider operator in the TaskFlow API? like AWS Glue Operators)
2. Get the Xcom instead of XcomArg in Jinjia template or inside task callback function

[enrique-ayala]

1. Change all Operator to TaskFlow API or DecoratedOperator. (Is there ways to declare provider operator in the TaskFlow API? like AWS Glue Operators) <-- This still prints a WARNING log on scheduler

[manugarri]

Has this issue been resolved? I am experiencing it when using taskgroups in Airflow 2.2.2

[ShinyoungAshleyChoi]

I have the same problem on airflow 2.5.0. the log was never showed on 2.3.2
the warnings are for both decorated and traditional operator.

in my case, the DAGs are generated dynamically, not using dynamic task mapping for tasks

FYI, PythonParallelOperator is custom operator by me which inherits PythonOperator.

[2022-12-14 15:31:46,005: WARNING/ForkPoolWorker-1] Dependency <Task(PythonOperator): extract.get_success_list>, extract.Branching_file_extension already registered for DAG: CN_PDF
[2022-12-14 15:31:46,005: WARNING/ForkPoolWorker-1] Dependency <Task(PythonParallelOperator): extract.Branching_file_extension>, extract.get_success_list already registered for DAG: CN_PDF
[2022-12-14 15:31:46,005: WARNING/ForkPoolWorker-1] Dependency <Task(PythonParallelOperator): extract.Branching_file_extension>, extract.Branching_need_2nd_expand already registered for DAG: CN_PDF
[2022-12-14 15:31:46,005: WARNING/ForkPoolWorker-1] Dependency <Task(_BranchPythonDecoratedOperator):

[potiuk]

Issue created here #28499 and marked for 2.5.1 - hopefully it will be fixed.

[kobethuwis]

Also experiencing this issue when dynamically generating GlueCrawlerOperators & other custom operators after bumping from version 2.3.0 to 2.4.1.