hussein-awala opened a new pull request, #76212: URL: https://github.com/apache/airflow/pull/76212
A trigger can end its task directly with a `TaskSuccessEvent`, `TaskFailedEvent` or `TaskSkippedEvent`. The `BaseTaskEndEvent` handler of `handle_event_submit` pushes that event's XComs [without the session it was given](https://github.com/apache/airflow/blob/b95498cd56/airflow-core/src/airflow/models/trigger.py#L746-L750). `TaskInstance.xcom_push` is wrapped in `provide_session`, so it enters [`create_session()`](https://github.com/apache/airflow/blob/b95498cd56/airflow-core/src/airflow/utils/session.py#L31-L48). When the caller's session also comes from the scoped `settings.Session`, as it does in the triggerer, that returns the caller's own session, which `create_session()` commits and closes after each key. Pass the session through, as every other write in the handler already does. related: #76209 On `main`, `Trigger.submit_event` is therefore not one transaction. The task instance's terminal state, its callback request and the first XCom are committed with the first key, each further key commits again, and the rest of the event runs in a new transaction. Closing the session also detaches every object the caller had loaded. Measured on SQLite, through the caller's session and without a session as the triggerer [calls it](https://github.com/apache/airflow/blob/b95498cd56/airflow-core/src/airflow/jobs/triggerer_job_runner.py#L824-L826): | Scenario | Before | After | |---|---|---| | Two XComs, the caller rolls back afterwards | 2 commits inside `submit_event`; the task stays `success` with both XComs and its callback request, and an object the caller added earlier is committed too | no commit; the task is still `deferred`, with no XCom and no callback request | | The second XCom write raises | `success`, the callback request and the first XCom are committed, the second XCom is missing; submitting the event again changes nothing | nothing is committed; submitting the event again ends the task with both XComs | | Triggerer call, every write succeeds | 3 commits, same end state | 1 commit, same end state | In the triggerer the common case ends as before, because the `provide_session` around `submit_event` commits whatever remains. The difference shows when a write fails after the first XCom: on `main` the task is already terminal with its callback queued and only some of its XComs, so downstream tasks pull nothing for the missing keys, and the trigger is cleaned up. With the change the task stays deferred with its `trigger_id`, so the trigger fires again once a triggerer picks it up, as a failing first key already does on `main`. Callers that pass their own session, such as the event batching planned in #76209, would otherwise lose control of their transaction. `xcom_push` is the only session-less `provide_session` call on this path: with the change, a recorder on `create_session` saw no call during `submit_event` for success, skipped, failed with and without a retry, a plain `TriggerEvent`, or a trigger with an asset watcher. The new test submits a `TaskSuccessEvent` with two XComs through the caller's session, counts commits with an `after_commit` listener, rolls back, and checks that the task is still deferred with no XComs and no new callback request. Without the change it fails with `assert 2 == 0`, one commit per key; the state and callback-request assertions fail on their own too. `v3-4-test` has the same code, and the commit applies to it without conflicts, so the fix can be backported. The batching PR planned in #76209 had carried this fix and will drop it. Validation: - `test_trigger.py` and `test_triggerer_job.py`: 206 passed, 1 skipped on SQLite. - Same tests through Breeze with the CI image: 207 passed on PostgreSQL 14; 206 passed, 1 skipped on MySQL 8.0 (the skip is a PostgreSQL-only test). - Regular and manual static checks passed on the full diff. - Selected core suites and the documentation build were not run locally; CI validation is pending. --- ##### Was generative AI tooling used to co-author this PR? - [X] Yes — Claude Code (Opus 5.5) Generated-by: Claude Code (Opus 5.5) following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions) 🤖 Generated with [Claude Code](https://claude.com/claude-code) -- 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]
