macksclark opened a new issue, #49778:
URL: https://github.com/apache/airflow/issues/49778
### Apache Airflow version
2.10.5
### If "Other Airflow 2 version" selected, which one?
_No response_
### What happened?
I have 2 tasks, one after the other (in a task group) where the 2nd one
depends on the Xcom variables from the first task. But the second task doesn't
wait for the xcom_push() call in the first task before running xcom_pull
seemingly.
The full code is this:
```
import pendulum
from airflow.decorators import dag, task, task_group
from airflow import models
from airflow.utils.dates import days_ago
PROJECT_ID = models.Variable.get("project_id")
REGION = models.Variable.get("region")
default_args = {
# Tell airflow to start one day ago, so that it runs as soon as you
upload it
"start_date": days_ago(1),
"project_id": PROJECT_ID,
"region": REGION,
}
@dag(
dag_id="testing-xcoms",
# Tell airflow to start one day ago, so that it runs once it's uploaded
start_date=pendulum.yesterday(),
catchup=False,
default_args=default_args,
# Runs daily by default.
schedule_interval=None,
)
def go():
@task_group(group_id='setup')
def setup(**context) -> None:
"""
Contains the setup & variable assignment tasks.
"""
@task()
def extract_logical_date(**context):
start_time = context["logical_date"]
context['task_instance'].xcom_push(
key='logical_date_s', value=start_time.int_timestamp)
@task()
def configure_table_names(**context):
logical_date_s = context['task_instance'].xcom_pull(
task_ids='extract_logical_date', key='logical_date_s')
context['task_instance'].xcom_push(key='table_1',
value=f'one_{logical_date_s}')
context['task_instance'].xcom_push(key='table_2',
value=f'two_{logical_date_s}')
context['task_instance'].xcom_push(key='table_3',
value=f'three_{logical_date_s}')
# Any other setup required should be added here.
extract_logical_date() >> configure_table_names()
setup()
go()
```
Then when I look at the XComs page on Airflow, you can see that the
`table_1/2/3` XCom variables do not use the `logical_date_s` XCom variable, and
that they get set to `one/two/three_None`. I need these new variables to use
the `logical_date_s` otherwise they will not be correct for my downstream steps
after this setup.
However, when you look at the DAG run, you can see that the
`extract_logical_date` task runs before `configure_table_names`:
### What you think should happen instead?
The variables in configure_table_names should get set to `one_<value from
logical_date_s>`, for example. I would expect there to be a dependency between
the step that outputs the Xcom, and the one that pulls it in, since I'm
referring to the XCom using `task_ids='extract_logical_date'`, I would assume
that it can understand to wait for that task to complete.
### How to reproduce
Copy attached code into a file and run the DAG.
### Operating System
Linux
### Versions of Apache Airflow Providers
_No response_
### Deployment
Google Cloud 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
- [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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
