[GitHub] [airflow] matthieucx opened a new issue, #32603: Issue when passing the output of a traditional operator to a Task Group

[via GitHub]

matthieucx opened a new issue, #32603:
URL: https://github.com/apache/airflow/issues/32603

   ### Apache Airflow version
   
   2.6.3
   
   ### What happened
   
   Hello!
   
   When passing the output of a traditional operator to a task group, the 
operator is added to the task group.
   
   
![operator_tg_issue](https://github.com/apache/airflow/assets/11246353/ca99d153-e506-44f7-adf5-6eff8a6cc463)
   
   Furthermore, when expanding the task group using this output, the task group 
is set as upstream of the operator, which causes a cyclic dependency and fails 
the DAG.
   
   
![operator_tg_expansion_issue](https://github.com/apache/airflow/assets/11246353/ede4bc69-b81f-4644-903b-d7877736ddf2)
   
   As a quick fix, adding a task that simply returns its input between the task 
group and the operator solves the issue.
   
   
![operator_tg_expansion_fix](https://github.com/apache/airflow/assets/11246353/a7870d76-f6c1-441f-8206-4423eb1f9406)
   
   
   
   ### What you think should happen instead
   
   The operator should not be added to the task group. It should behave like a 
`@task()` decorated function would.
   
   ### How to reproduce
   
   The following minimal DAG is enough to reproduce the issue:
   
   ```python
   from datetime import datetime
   
   from airflow.decorators import dag, task, task_group
   from airflow.operators.dummy_operator import DummyOperator
   
   dummy_task = DummyOperator(task_id="dummy_task")
   
   
   @task()
   def print_task(data):
       print(data)
   
   
   @task()
   def passtrough(data):
       return data
   
   
   @task_group()
   def tg(data):
       print_task(data)
   
   
   @dag(
       dag_id="expansion_classic_operator",
       schedule=None,
       catchup=False,
       start_date=datetime(2021, 1, 1),
       max_active_runs=1,
   )
   def ExpansionIssueDAG():
       data_list = passtrough(dummy_task.output)
       tg.expand(data=data_list)
   
   
   dag = ExpansionIssueDAG()
   ```
   
   ### Operating System
   
   Ubuntu 22.04.2 LTS
   
   ### Versions of Apache Airflow Providers
   
   _No response_
   
   ### Deployment
   
   Docker-Compose
   
   ### Deployment details
   
   _No response_
   
   ### Anything else
   
   I would be willing to contribute, but I haven't investigated where the issue 
could originate from and don't have a lot of time on my hand at the moment.
   
   ### 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-unsubscr...@airflow.apache.org.apache.org

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