You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "neel-astro (via GitHub)" <gi...@apache.org> on 2026/08/31 11:57:31 UTC
[PR] Fix asset event partition_key breaking older Task SDK workers [airflow]
neel-astro opened a new pull request, #72327:
URL: https://github.com/apache/airflow/pull/72327
### Why
`AssetEventDagRunReference.partition_key` was added in 3.3.0 (#69115) but registered under the already-released `2026-04-06` Execution API version. Cadwyn applies a version's downgrade instructions only for clients *older* than that version, so a client requesting exactly `2026-04-06` still receives the field.
Task SDK 1.2.x (Airflow 3.2.x) speaks `2026-04-06`, and its `AssetEventDagRunReference` has no `partition_key`, with `extra="forbid"`. Every asset-triggered task run by a 3.2.x worker against a 3.3.x API server fails before it starts:
```
airflow/sdk/api/client.py:231 in start
return TIRunContext.model_validate_json(resp.read())
pydantic_core._pydantic_core.ValidationError: 1 validation error for TIRunContext
dag_run.consumed_asset_events.0.partition_key
Extra inputs are not permitted [type=extra_forbidden, input_value=None, input_type=NoneType]
```
The value is `null`, so this affects every asset-triggered run, not just partitioned assets. The task dies at supervision start.
Reading `partition_key` out of each released task-sdk wheel's `airflow/sdk/api/datamodels/_generated.py`:
| task-sdk | API_VERSION | `AssetEventDagRunReference` |
|---|---|---|
| 1.1.5 (3.1.x) | 2025-11-05 | absent |
| 1.2.0 (3.2.0) | 2026-04-06 | **absent** |
| 1.2.1 (3.2.1) | 2026-04-06 | **absent** |
| 1.2.2 (3.2.2) | 2026-04-06 | **absent** |
| 1.3.0 (3.3.0) | 2026-06-30 | present |
| 1.3.1 (3.3.1) | 2026-06-30 | present |
`AssetEventDagRunReference` is the only one of the five models in `AddPartitionKeyField` that did not gain the field at `2026-04-06`. `DagRun`, `AssetEventResponse`, `TriggerDAGRunPayload` and `DagRunAssetReference` all carry it in 1.2.0+, so their gating is correct and is left alone.
### What
Move the `AssetEventDagRunReference.partition_key` schema instruction and its `consumed_asset_events` response converter out of `v2026_04_06.py` into a new `AddConsumedAssetEventPartitionKeyField` in `v2026_06_30.py`, matching how the sibling `AddPartitionDateField` handles `DagRun.partition_date` (also new in 3.3.0). The HEAD datamodel and the generated Task SDK models are unchanged.
Both halves of the gate move together. The `schema(...).didnt_exist` instruction drives the per-version OpenAPI document and the converter mutates the wire body, so leaving either behind would ship a spec that contradicts the response.
Only `TIRunContext` gets a converter, where the two siblings in the same file register three each. `DagRun.safe_extract_from_orm` defaults `consumed_asset_events` to `[]` whenever the relationship is not already loaded, and `PATCH /task-instances/{id}/run` is the only route that loads it, so bare `DagRun` responses never carry one. Two tests in `versions/v2026_06_30/test_dag_runs.py` pin that invariant so it fails loudly if a future change adds eager loading there.
### Tests
`TestConsumedEventPartitionKeyBackwardCompat` moves to the `v2026_06_30` suite, where `old_ver_client` is pinned to exactly `2026-04-06` — the version Task SDK 1.2.x sends. In its previous home it ran at `2025-11-05`, older than the mis-placed gate, which is why this passed CI.
Three assertions cover the three ways this can go wrong:
- a `2026-04-06` client gets the event-level `partition_key` stripped while the Dag-run-level `partition_key` survives, in one response (under-moving the gate)
- a client at exactly `2026-06-30` and a HEAD client both still receive it (over-moving the gate would silently break 1.3.x, which sends `2026-06-30` and carries the field)
- the served OpenAPI for `AssetEventDagRunReference` matches the wire body at both versions (moving only one half of the gate)
Beyond the suite, the released 1.2.2 `TIRunContext` — the model that raised the traceback above — was loaded from its wheel and used to validate a real captured `2026-04-06` response body: it accepts the server's current output and still rejects it once `partition_key` is re-injected. That checks response bytes against the real released client model; it is not a live 1.2.x worker running end to end.
One pre-existing test fixture changed: `old_ver_client` in `versions/v2026_06_30/test_task_instances.py` moved from `2026-06-16` to `2026-04-06`. This is behaviour-preserving — `2026-06-16` was never a registered version and resolved to the same bundle — and it aligns the module with its siblings in that directory, which already pin `2026-04-06`.
### Backport
3.3.0 and 3.3.1 are both released with this bug, so a `main`-only fix does not help anyone currently affected. Requesting the **3.3.2** milestone.
related: #69115
related: #71861 — that PR fixes a different class of gap in the same two files (missing registrations on the compat previous-run route and nested `created_dagruns`). The two are semantically orthogonal, but both touch `remove_partition_key_from_dag_run`, so whichever merges second will conflict there. The resolution is to keep the `consumed_asset_events` loop deleted and keep that PR's added converters; re-adding the loop alongside the new `VersionChange` is a silent no-op that the test suite cannot detect.
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes — Claude Code (Opus 5)
Generated-by: Claude Code (Opus 5) following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
--
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-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix asset event partition_key breaking older Task SDK workers [airflow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #72327:
URL: https://github.com/apache/airflow/pull/72327#issuecomment-5606242615
### Backport successfully created: v3-3-test
Note: As of [Merging PRs targeted for Airflow 3.X](https://github.com/apache/airflow/blob/main/dev/README_AIRFLOW3_DEV.md#merging-prs-targeted-for-airflow-3x)
the committer who merges the PR is responsible for backporting the PRs that are bug fixes (generally speaking) to the maintenance branches.
In matter of doubt please ask in [#release-management](https://apache-airflow.slack.com/archives/C03G9H97MM2) Slack channel.
<table>
<tr>
<th>Status</th>
<th>Branch</th>
<th>Result</th>
</tr>
<tr>
<td>✅</td>
<td>v3-3-test</td>
<td><a href="https://github.com/apache/airflow/pull/72827"><img src="https://img.shields.io/badge/PR-72827-blue" alt="PR Link"></a></td>
</tr>
</table>
--
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-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix asset event partition_key breaking older Task SDK workers [airflow]
Posted by "vatsrahul1001 (via GitHub)" <gi...@apache.org>.
vatsrahul1001 merged PR #72327:
URL: https://github.com/apache/airflow/pull/72327
--
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-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix asset event partition_key breaking older Task SDK workers [airflow]
Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on PR #72327:
URL: https://github.com/apache/airflow/pull/72327#issuecomment-5606216653
Awesome work, congrats on your first merged pull request! You are invited to check our [Issue Tracker](https://github.com/apache/airflow/issues) for additional contributions.
--
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-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix asset event partition_key breaking older Task SDK workers [airflow]
Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on PR #72327:
URL: https://github.com/apache/airflow/pull/72327#issuecomment-5477978783
Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our [Contributors' Guide](https://github.com/apache/airflow/blob/main/contributing-docs/README.rst)
Here are some useful points:
- Pay attention to the quality of your code (ruff, mypy and type annotations). Our [prek-hooks]( https://github.com/apache/airflow/blob/main/contributing-docs/08_static_code_checks.rst#prerequisites-for-prek-hooks) will help you with that.
- In case of a new feature add useful documentation (in docstrings or in `docs/` directory). Adding a new operator? Check this short [guide](https://github.com/apache/airflow/blob/main/airflow-core/docs/howto/custom-operator.rst) Consider adding an example Dag that shows how users should use it.
- Consider using [Breeze environment](https://github.com/apache/airflow/blob/main/dev/breeze/doc/README.rst) for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
- Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
- Please follow [ASF Code of Conduct](https://www.apache.org/foundation/policies/conduct) for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
- Be sure to read the [Airflow Coding style]( https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#coding-style-and-best-practices).
- Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
Apache Airflow is a community-driven project and together we are making it better 🚀.
In case of doubts contact the developers at:
Mailing List: dev@airflow.apache.org
Slack: https://s.apache.org/airflow-slack
--
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-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix asset event partition_key breaking older Task SDK workers [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #72327:
URL: https://github.com/apache/airflow/pull/72327#discussion_r3963892349
##########
airflow-core/tests/unit/api_fastapi/execution_api/versions/v2026_06_30/test_dag_runs.py:
##########
@@ -88,3 +91,53 @@ def test_get_previous_dag_run_without_a_match(old_ver_client):
assert response.status_code == 200
assert response.json() is None
+
+
+@pytest.fixture
+def dag_run_with_consumed_event(session, dag_maker):
+ """A Dag run that consumed an asset event carrying a ``partition_key``."""
+ # dag_maker resets Dag/DagRun tables between tests but not asset tables, and this fixture
+ # commits (the API request needs to see the row), so a leftover asset from a prior test
+ # using this fixture would collide on the name/uri unique constraint.
+ clear_db_assets()
+ with dag_maker(dag_id="test_dag_run_consumed_event", session=session, serialized=True):
+ pass
+ run = dag_maker.create_dagrun(
+ state=DagRunState.SUCCESS,
+ logical_date=timezone.datetime(2025, 1, 1),
+ run_id="run1",
+ )
+ asset = AssetModel(name="upstream", uri="s3://bucket/upstream", group="asset", extra={})
+ session.add_all([asset, AssetActive.for_asset(asset)])
+ session.flush()
+ run.consumed_asset_events.append(
+ AssetEvent(asset_id=asset.id, source_dag_id="src", source_run_id="r1", partition_key="2024-01-15")
+ )
+ session.commit()
+ yield
+ clear_db_assets()
+
+
+@pytest.mark.usefixtures("dag_run_with_consumed_event")
+def test_get_dag_run_strips_consumed_event_partition_key(old_ver_client):
+ """A bare-DagRun route must not leak an event-level partition_key to a pre-2026-06-30 client."""
+ response = old_ver_client.get("/execution/dag-runs/test_dag_run_consumed_event/run1")
+
+ assert response.status_code == 200
+ assert all("partition_key" not in event for event in response.json()["consumed_asset_events"])
Review Comment:
`consumed_asset_events` is always `[]` on this route: the handler runs a plain `select(DagRunModel)` with no `joinedload`, `SessionDep` is a fresh `create_session(scoped=False)` so the fixture's in-session append is not visible, and `safe_extract_from_orm` defaults the unloaded relationship. So this `all(...)` iterates nothing and would pass with the version gate deleted entirely. Same for the `/previous` test below. Asserting `response.json()["consumed_asset_events"] == []` would pin the invariant that actually holds today, and would still fail loudly if a later change adds eager loading here.
--
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-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix asset event partition_key breaking older Task SDK workers [airflow]
Posted by "neel-astro (via GitHub)" <gi...@apache.org>.
neel-astro commented on code in PR #72327:
URL: https://github.com/apache/airflow/pull/72327#discussion_r3969227328
##########
airflow-core/tests/unit/api_fastapi/execution_api/versions/v2026_06_30/test_dag_runs.py:
##########
@@ -88,3 +91,53 @@ def test_get_previous_dag_run_without_a_match(old_ver_client):
assert response.status_code == 200
assert response.json() is None
+
+
+@pytest.fixture
+def dag_run_with_consumed_event(session, dag_maker):
+ """A Dag run that consumed an asset event carrying a ``partition_key``."""
+ # dag_maker resets Dag/DagRun tables between tests but not asset tables, and this fixture
+ # commits (the API request needs to see the row), so a leftover asset from a prior test
+ # using this fixture would collide on the name/uri unique constraint.
+ clear_db_assets()
+ with dag_maker(dag_id="test_dag_run_consumed_event", session=session, serialized=True):
+ pass
+ run = dag_maker.create_dagrun(
+ state=DagRunState.SUCCESS,
+ logical_date=timezone.datetime(2025, 1, 1),
+ run_id="run1",
+ )
+ asset = AssetModel(name="upstream", uri="s3://bucket/upstream", group="asset", extra={})
+ session.add_all([asset, AssetActive.for_asset(asset)])
+ session.flush()
+ run.consumed_asset_events.append(
+ AssetEvent(asset_id=asset.id, source_dag_id="src", source_run_id="r1", partition_key="2024-01-15")
+ )
+ session.commit()
+ yield
+ clear_db_assets()
+
+
+@pytest.mark.usefixtures("dag_run_with_consumed_event")
+def test_get_dag_run_strips_consumed_event_partition_key(old_ver_client):
+ """A bare-DagRun route must not leak an event-level partition_key to a pre-2026-06-30 client."""
+ response = old_ver_client.get("/execution/dag-runs/test_dag_run_consumed_event/run1")
+
+ assert response.status_code == 200
+ assert all("partition_key" not in event for event in response.json()["consumed_asset_events"])
Review Comment:
Good catch 🫡 , fixing it
--
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-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org