can not use output of task decorator as input for external_task_ids of ExternalTaskSensor

[MaiHoangViet1809]

Apache Airflow version

2.4.3, 2.5.0

What happened

when use output from task decorator as parameter (external_task_ids) in ExternalTaskSensor, it show up this log:

Broken DAG: [+++++/airflow/dags/TEST_NEW_PIPELINE.py] Traceback (most recent call last):
  File "+++++/env3.10.5/lib/python3.10/site-packages/airflow/models/baseoperator.py", line 408, in apply_defaults
    result = func(self, **kwargs, default_args=default_args)
  File "+++++/env3.10.5/lib/python3.10/site-packages/airflow/sensors/external_task.py", line 164, in __init__
    if external_task_ids and len(external_task_ids) > len(set(external_task_ids)):
TypeError: object of type 'PlainXComArg' has no len()

note: +++++ is just a mask for irrelevant information.

What you think should happen instead

this document https://airflow.apache.org/docs/apache-airflow/stable/tutorial/taskflow.html
show that we can use it without any warning, note about it.

found relative problem in https://github.com/apache/airflow/issues/27328
should move all the check in __init__ into poke method.

How to reproduce

from airflow.decorators import dag, task
from airflow.operators.python import get_current_context

from airflow.sensors.external_task import ExternalTaskSensor
from datetime import datetime

configure = {"dag_id": "test_new_skeleton",
                    "schedule": None,
                    "start_date": datetime(2022,1,1),
}

@task
def preprocess_dependency() -> list:
    return ["random-task-name"]

@dag(**configure)
def pipeline(): 
    t_preprocess = preprocess_dependency()
    task_dependency = ExternalTaskSensor(task_id=f"Check_Dependency",
                                         external_dag_id='random-dag-name-that-exist',
                                         external_task_ids=t_preprocess ,
                                         poke_interval=60,
                                         mode="reschedule",
                                         timeout=172800,
                                         allowed_states=['success'],
                                         failed_states=['failed', 'skipped'],
                                         check_existence=True,)


dag = pipeline()

Operating System

REHL 7

Versions of Apache Airflow Providers

No response

Deployment

Virtualenv installation

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

[potiuk]

This has been fixed in 2.5.0 #27251

[MaiHoangViet1809]

it was not fixed yet @potiuk, I believe the code still unchanged

[potiuk]

it was not fixed yet @potiuk, I believe the code still unchanged

Are you able to reproduce in 2.5.0 @MaiHoangViet1809 ?

I believe it does not require code change in the place where it raised the warning - I think the root cause was elsewhere.

Can you confirm please it has not been fixed in 2.5.0 ? Logs?

[MaiHoangViet1809]

@potiuk yes, I've checked in source code of v2.5.0:

airflow/airflow/sensors/external_task.py

Lines 165 to 166 in 5fad8a4

	if external_task_ids and len(external_task_ids) > len(set(external_task_ids)):
	raise ValueError("Duplicate task_ids passed in external_task_ids parameter")

[potiuk]

But did you actually reproduce it ? @MaiHoangViet1809 ?

I belive the problem has been solved in the way that len(external_task_ids) will not raise the exception in the first place (this was the fix in #27251). Please double-check if you can reproduce it there and post logs in case it is not.

[MaiHoangViet1809]

But did you actually reproduce it ? @MaiHoangViet1809 ?

I belive the problem has been solved in the way that len(external_task_ids) will not raise the exception in the first place (this was the fix in #27251). Please double-check if you can reproduce it there and post logs in case it is not.

oh, #27251 resolved another issues I reported before this, #27209.

this one is still error in v2.5.0 after I upgraded airflow from 2.4.3 to 2.5.0
error log is the same with v2.4.3.

[potiuk]

Can you post the logs from v2.5.0 showing the error (I am asking this for the third time, so please don't make me ask it again - just reproduce and post it).

[MaiHoangViet1809]

the log from v2.5.0 is below:

Broken DAG: [/********/airflow/dags/reproduce_error_for_27952.py] Traceback (most recent call last):
  File "/********/env3.10.5/lib/python3.10/site-packages/airflow/models/baseoperator.py", line 411, in apply_defaults
    result = func(self, **kwargs, default_args=default_args)
  File "/********/env3.10.5/lib/python3.10/site-packages/airflow/sensors/external_task.py", line 165, in __init__
    if external_task_ids and len(external_task_ids) > len(set(external_task_ids)):
TypeError: object of type 'PlainXComArg' has no len()

note: masked path with ********

[potiuk]

Ah ok. Thanks. yep. Not fixed it seems.