Skip to content

Enable Comms in the TriggerRunner to make Connections, Variables and XComs accessible to Triggers#48239

Merged
pierrejeambrun merged 16 commits into
apache:mainfrom
astronomer:48183-init-trigger-comms
Mar 31, 2025
Merged

Enable Comms in the TriggerRunner to make Connections, Variables and XComs accessible to Triggers#48239
pierrejeambrun merged 16 commits into
apache:mainfrom
astronomer:48183-init-trigger-comms

Conversation

@pierrejeambrun

@pierrejeambrun pierrejeambrun commented Mar 25, 2025

Copy link
Copy Markdown
Member

Closes: #48183

Init the comms to be able to retrieve connections, variables and xcoms from the Trigger.

Comment thread airflow-core/src/airflow/hooks/base.py Fixed
Comment thread airflow-core/src/airflow/hooks/base.py Fixed
@pierrejeambrun

Copy link
Copy Markdown
Member Author

cc: @dstandish

Comment thread airflow-core/src/airflow/jobs/triggerer_job_runner.py Outdated
Comment thread airflow-core/src/airflow/jobs/triggerer_job_runner.py
Comment thread airflow-core/src/airflow/jobs/triggerer_job_runner.py
Comment thread airflow-core/src/airflow/jobs/triggerer_job_runner.py
Comment thread task-sdk/src/airflow/sdk/execution_time/task_runner.py
@pierrejeambrun pierrejeambrun force-pushed the 48183-init-trigger-comms branch from 0c0521a to e3a3ea0 Compare March 25, 2025 14:08
Comment thread airflow-core/src/airflow/hooks/base.py

@ashb ashb left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes the approach looks about right.

I'm a little concerned about the interleaving/mixign of cancel and conn response messages in the async side read. Let me re-read that code

Comment thread airflow-core/src/airflow/jobs/triggerer_job_runner.py Outdated
Comment thread airflow-core/src/airflow/jobs/triggerer_job_runner.py Outdated
Comment thread task-sdk/src/airflow/sdk/execution_time/context.py
Comment thread task-sdk/src/airflow/sdk/execution_time/task_runner.py
@pierrejeambrun

Copy link
Copy Markdown
Member Author

Can also retrieve Variables.

Comment thread airflow-core/src/airflow/dag_processing/processor.py
Comment thread airflow-core/src/airflow/jobs/triggerer_job_runner.py Outdated
@gopidesupavan gopidesupavan reopened this Mar 28, 2025
@gopidesupavan

Copy link
Copy Markdown
Member

Triggering tests..

@pierrejeambrun pierrejeambrun force-pushed the 48183-init-trigger-comms branch from d134e29 to d2d9553 Compare March 29, 2025 02:51
@pierrejeambrun

pierrejeambrun commented Mar 29, 2025

Copy link
Copy Markdown
Member Author

Well that's progress. test_trigger_can_access_variables_connections_and_xcoms is now working in all tests beside one timeout.

It seems that there's still one side effect to fix on test_calls_to_verify_integrity_with_mapped_task_zero_length_at_runtime

@ashb ashb force-pushed the 48183-init-trigger-comms branch from d2d9553 to d11954f Compare March 29, 2025 11:30
@pierrejeambrun pierrejeambrun force-pushed the 48183-init-trigger-comms branch from d11954f to 65f90f9 Compare March 31, 2025 11:37
Comment thread airflow-core/tests/unit/models/test_dagrun.py
@pierrejeambrun pierrejeambrun merged commit 683cae4 into apache:main Mar 31, 2025
@pierrejeambrun pierrejeambrun deleted the 48183-init-trigger-comms branch March 31, 2025 13:52
@pankajkoti

pankajkoti commented Apr 1, 2025

Copy link
Copy Markdown
Member

Thank you @pierrejeambrun for working on this. I am now able to use connections in triggerer and the DAG is running fine with deferral mode 👏🏽

shubham-pyc pushed a commit to shubham-pyc/airflow that referenced this pull request Apr 2, 2025
…XComs accessible to Triggers (apache#48239)

* Init triggerer comms

* Remove cache from in_process_api_server

* Add lock for trigger threads to wait for their messages

* Move comms init code to TriggerRunner class

* Add TODO for empty line investigation

* Update kind to type for common discriminator

* Add test, add xcom handling

* Address PR comments

* Rename test supervisor class

* Add XCom test and code

* Fix following code review

* Fix CI and add comment

* Update task-sdk/src/airflow/sdk/api/client.py

* Small adjustments

* Remove unused fixture

* Fix CI
nailo2c pushed a commit to nailo2c/airflow that referenced this pull request Apr 4, 2025
…XComs accessible to Triggers (apache#48239)

* Init triggerer comms

* Remove cache from in_process_api_server

* Add lock for trigger threads to wait for their messages

* Move comms init code to TriggerRunner class

* Add TODO for empty line investigation

* Update kind to type for common discriminator

* Add test, add xcom handling

* Address PR comments

* Rename test supervisor class

* Add XCom test and code

* Fix following code review

* Fix CI and add comment

* Update task-sdk/src/airflow/sdk/api/client.py

* Small adjustments

* Remove unused fixture

* Fix CI
simonprydden pushed a commit to simonprydden/airflow that referenced this pull request Apr 8, 2025
…XComs accessible to Triggers (apache#48239)

* Init triggerer comms

* Remove cache from in_process_api_server

* Add lock for trigger threads to wait for their messages

* Move comms init code to TriggerRunner class

* Add TODO for empty line investigation

* Update kind to type for common discriminator

* Add test, add xcom handling

* Address PR comments

* Rename test supervisor class

* Add XCom test and code

* Fix following code review

* Fix CI and add comment

* Update task-sdk/src/airflow/sdk/api/client.py

* Small adjustments

* Remove unused fixture

* Fix CI
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

area:task-execution-interface-aip72 AIP-72: Task Execution Interface (TEI) aka Task SDK

Projects

No open projects

Development

Successfully merging this pull request may close these issues.

Triggerer code does not seem to be able to use an Airflow connection & variable

8 participants