This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch v3-3-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-3-test by this push:
new 334ca49afc7 [v3-3-test] Gate asset event partition_key behind
execution API 2026-06-30 (#72327) (#72827)
334ca49afc7 is described below
commit 334ca49afc73e443c7c9738b055efba61d0a83bf
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Thu Sep 10 16:26:40 2026 +0530
[v3-3-test] Gate asset event partition_key behind execution API 2026-06-30
(#72327) (#72827)
(cherry picked from commit 27e882439f66aa113d301eae37d45bba633c9acd)
Co-authored-by: Neel Dalsania <[email protected]>
Co-authored-by: Rahul Vats <[email protected]>
---
airflow-core/newsfragments/72327.bugfix.rst | 1 +
.../api_fastapi/execution_api/versions/__init__.py | 2 +
.../execution_api/versions/v2026_04_06.py | 7 +-
.../execution_api/versions/v2026_06_30.py | 23 ++++++
.../versions/v2026_04_06/test_task_instances.py | 54 +-------------
.../versions/v2026_06_30/test_dag_runs.py | 57 +++++++++++++++
.../versions/v2026_06_30/test_task_instances.py | 82 +++++++++++++++++++++-
7 files changed, 164 insertions(+), 62 deletions(-)
diff --git a/airflow-core/newsfragments/72327.bugfix.rst
b/airflow-core/newsfragments/72327.bugfix.rst
new file mode 100644
index 00000000000..05a35cdce40
--- /dev/null
+++ b/airflow-core/newsfragments/72327.bugfix.rst
@@ -0,0 +1 @@
+Fix asset-triggered tasks failing before they start for workers running Task
SDK 1.2.x (Airflow 3.2.x), which request execution API version ``2026-04-06``.
Those workers received a ``partition_key`` field on the Dag run's consumed
asset events that their models reject, raising an ``extra_forbidden``
validation error. The field is now gated behind version ``2026-06-30``, where
it was introduced; workers on ``2026-06-30`` or newer are unaffected.
diff --git
a/airflow-core/src/airflow/api_fastapi/execution_api/versions/__init__.py
b/airflow-core/src/airflow/api_fastapi/execution_api/versions/__init__.py
index dc7035d31e3..97a525436c8 100644
--- a/airflow-core/src/airflow/api_fastapi/execution_api/versions/__init__.py
+++ b/airflow-core/src/airflow/api_fastapi/execution_api/versions/__init__.py
@@ -44,6 +44,7 @@ from airflow.api_fastapi.execution_api.versions.v2026_06_30
import (
AddAssetsByAliasEndpoint,
AddAwaitingInputStatePayload,
AddConnectionTestEndpoint,
+ AddConsumedAssetEventPartitionKeyField,
AddPartitionDateField,
AddRetryPolicyFields,
AddTaskAndAssetStateStoreEndpoints,
@@ -65,6 +66,7 @@ bundle = VersionBundle(
AddTaskAndAssetStateStoreEndpoints,
AddAssetsByAliasEndpoint,
AddPartitionDateField,
+ AddConsumedAssetEventPartitionKeyField,
),
Version(
"2026-04-06",
diff --git
a/airflow-core/src/airflow/api_fastapi/execution_api/versions/v2026_04_06.py
b/airflow-core/src/airflow/api_fastapi/execution_api/versions/v2026_04_06.py
index e3b995011f4..4710de4f665 100644
--- a/airflow-core/src/airflow/api_fastapi/execution_api/versions/v2026_04_06.py
+++ b/airflow-core/src/airflow/api_fastapi/execution_api/versions/v2026_04_06.py
@@ -29,7 +29,6 @@ from airflow.api_fastapi.execution_api.datamodels.asset_event
import (
)
from airflow.api_fastapi.execution_api.datamodels.dagrun import
TriggerDAGRunPayload
from airflow.api_fastapi.execution_api.datamodels.taskinstance import (
- AssetEventDagRunReference,
DagRun,
TIDeferredStatePayload,
TIRunContext,
@@ -37,14 +36,13 @@ from
airflow.api_fastapi.execution_api.datamodels.taskinstance import (
class AddPartitionKeyField(VersionChange):
- """Add the `partition_key` field to DagRun model."""
+ """Add ``partition_key`` to the Dag run, asset event, asset reference and
trigger payload models."""
description = __doc__
instructions_to_migrate_to_previous_version = (
schema(DagRun).field("partition_key").didnt_exist,
schema(AssetEventResponse).field("partition_key").didnt_exist,
- schema(AssetEventDagRunReference).field("partition_key").didnt_exist,
schema(TriggerDAGRunPayload).field("partition_key").didnt_exist,
schema(DagRunAssetReference).field("partition_key").didnt_exist,
)
@@ -55,9 +53,6 @@ class AddPartitionKeyField(VersionChange):
dag_run = response.body.get("dag_run")
if isinstance(dag_run, dict):
dag_run.pop("partition_key", None)
- for event in dag_run.get("consumed_asset_events") or ():
- if isinstance(event, dict):
- event.pop("partition_key", None)
@convert_response_to_previous_version_for(AssetEventsResponse) # type:
ignore[arg-type]
def remove_partition_key_from_asset_events(response: ResponseInfo) ->
None: # type: ignore[misc]
diff --git
a/airflow-core/src/airflow/api_fastapi/execution_api/versions/v2026_06_30.py
b/airflow-core/src/airflow/api_fastapi/execution_api/versions/v2026_06_30.py
index 0a316810691..85d46420040 100644
--- a/airflow-core/src/airflow/api_fastapi/execution_api/versions/v2026_06_30.py
+++ b/airflow-core/src/airflow/api_fastapi/execution_api/versions/v2026_06_30.py
@@ -26,6 +26,7 @@ from cadwyn import (
)
from airflow.api_fastapi.execution_api.datamodels.taskinstance import (
+ AssetEventDagRunReference,
DagRun,
TaskInstance,
TIAwaitingInputStatePayload,
@@ -168,3 +169,25 @@ class AddPartitionDateField(VersionChange):
"""Strip ``partition_date`` from the previous-run response."""
if isinstance(response.body, dict):
response.body.pop("partition_date", None)
+
+
+class AddConsumedAssetEventPartitionKeyField(VersionChange):
+ """Expose the upstream partition key on the asset events that triggered a
consumer Dag run."""
+
+ description = __doc__
+
+ instructions_to_migrate_to_previous_version = (
+ schema(AssetEventDagRunReference).field("partition_key").didnt_exist,
+ )
+
+ # Only ``TIRunContext`` can carry these events:
``DagRun.safe_extract_from_orm`` defaults
+ # ``consumed_asset_events`` to ``[]`` whenever the relationship is not
already loaded, and
+ # ``/run`` is the only route that loads it -- so bare ``DagRun`` responses
never carry one.
+ @convert_response_to_previous_version_for(TIRunContext) # type:
ignore[arg-type]
+ def remove_partition_key_from_consumed_asset_events(response:
ResponseInfo) -> None: # type: ignore[misc]
+ """Strip ``partition_key`` from each consumed asset event for older
clients."""
+ dag_run = response.body.get("dag_run")
+ if isinstance(dag_run, dict):
+ for event in dag_run.get("consumed_asset_events") or ():
+ if isinstance(event, dict):
+ event.pop("partition_key", None)
diff --git
a/airflow-core/tests/unit/api_fastapi/execution_api/versions/v2026_04_06/test_task_instances.py
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/v2026_04_06/test_task_instances.py
index 4a8b59c362e..d8552917277 100644
---
a/airflow-core/tests/unit/api_fastapi/execution_api/versions/v2026_04_06/test_task_instances.py
+++
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/v2026_04_06/test_task_instances.py
@@ -20,11 +20,10 @@ from __future__ import annotations
import pytest
from airflow._shared.timezones import timezone
-from airflow.models.asset import AssetActive, AssetEvent, AssetModel
from airflow.serialization.serialized_objects import BaseSerialization
from airflow.utils.state import DagRunState, State
-from tests_common.test_utils.db import clear_db_assets, clear_db_runs
+from tests_common.test_utils.db import clear_db_runs
from tests_common.test_utils.format_datetime import
from_datetime_to_zulu_without_ms
pytestmark = pytest.mark.db_test
@@ -254,54 +253,3 @@ class TestNextKwargsBackwardCompat:
assert response.status_code == 200
# Head version gets the plain dict directly -- no BaseSerialization
wrapping
assert response.json()["next_kwargs"] == {"cheesecake": True, "event":
"payload"}
-
-
-class TestConsumedEventPartitionKeyBackwardCompat:
- """The partition_key on consumed asset events is stripped for
pre-2026-04-06 clients."""
-
- @pytest.fixture(autouse=True)
- def _freeze_time(self, time_machine):
- time_machine.move_to(TIMESTAMP_STR, tick=False)
-
- def setup_method(self):
- clear_db_runs()
- clear_db_assets()
-
- def teardown_method(self):
- clear_db_runs()
- clear_db_assets()
-
- def _create_ti_with_consumed_event(self, session, create_task_instance):
- ti = create_task_instance(
- task_id="test_consumed_event_partition_key_compat",
- state=State.QUEUED,
- session=session,
- start_date=TIMESTAMP,
- )
- asset = AssetModel(name="upstream", uri="s3://bucket/upstream",
group="asset", extra={})
- session.add_all([asset, AssetActive.for_asset(asset)])
- session.flush()
- ti.dag_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()
- return ti
-
- def test_old_version_strips_partition_key(self, old_ver_client, session,
create_task_instance):
- ti = self._create_ti_with_consumed_event(session, create_task_instance)
-
- response =
old_ver_client.patch(f"/execution/task-instances/{ti.id}/run",
json=RUN_PATCH_BODY)
-
- assert response.status_code == 200
- events = response.json()["dag_run"]["consumed_asset_events"]
- assert events
- assert all("partition_key" not in event for event in events)
-
- def test_head_version_keeps_partition_key(self, client, session,
create_task_instance):
- ti = self._create_ti_with_consumed_event(session, create_task_instance)
-
- response = client.patch(f"/execution/task-instances/{ti.id}/run",
json=RUN_PATCH_BODY)
-
- assert response.status_code == 200
- events = response.json()["dag_run"]["consumed_asset_events"]
- assert [event["partition_key"] for event in events] == ["2024-01-15"]
diff --git
a/airflow-core/tests/unit/api_fastapi/execution_api/versions/v2026_06_30/test_dag_runs.py
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/v2026_06_30/test_dag_runs.py
index d89c585d364..fe16f4ee8c8 100644
---
a/airflow-core/tests/unit/api_fastapi/execution_api/versions/v2026_06_30/test_dag_runs.py
+++
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/v2026_06_30/test_dag_runs.py
@@ -20,8 +20,11 @@ from __future__ import annotations
import pytest
from airflow._shared.timezones import timezone
+from airflow.models.asset import AssetActive, AssetEvent, AssetModel
from airflow.utils.state import DagRunState
+from tests_common.test_utils.db import clear_db_assets
+
pytestmark = pytest.mark.db_test
ADDED_IN_2026_06_30 = frozenset({"team_name", "partition_date"})
@@ -88,3 +91,57 @@ def
test_get_previous_dag_run_without_a_match(old_ver_client):
assert response.status_code == 200
assert response.json() is None
+
+
[email protected]
+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()
+
+
[email protected]("dag_run_with_consumed_event")
+def test_get_dag_run_returns_no_consumed_asset_events(old_ver_client):
+ """Bare-DagRun routes serialize no consumed asset events, so they need no
partition_key converter."""
+ response =
old_ver_client.get("/execution/dag-runs/test_dag_run_consumed_event/run1")
+
+ assert response.status_code == 200
+ # The Dag run really has a consumed event carrying a partition_key, but
this route does not
+ # eager-load the relationship and ``safe_extract_from_orm`` defaults it
away. Assert the empty
+ # list rather than the absence of the key: an ``all(...)`` over these
events would hold
+ # vacuously and pass even with the version gate deleted.
+ assert response.json()["consumed_asset_events"] == []
+
+
[email protected]("dag_run_with_consumed_event")
+def test_get_previous_dag_run_returns_no_consumed_asset_events(old_ver_client):
+ """Same invariant via /previous, which also returns a bare DagRun."""
+ response = old_ver_client.get(
+ "/execution/dag-runs/previous",
+ params={
+ "dag_id": "test_dag_run_consumed_event",
+ "logical_date": timezone.datetime(2025, 1, 2).isoformat(),
+ },
+ )
+
+ assert response.status_code == 200
+ assert response.json()["run_id"] == "run1"
+ assert response.json()["consumed_asset_events"] == []
diff --git
a/airflow-core/tests/unit/api_fastapi/execution_api/versions/v2026_06_30/test_task_instances.py
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/v2026_06_30/test_task_instances.py
index b2cc016be84..b2536aa4c07 100644
---
a/airflow-core/tests/unit/api_fastapi/execution_api/versions/v2026_06_30/test_task_instances.py
+++
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/v2026_06_30/test_task_instances.py
@@ -20,9 +20,10 @@ from __future__ import annotations
import pytest
from airflow._shared.timezones import timezone
+from airflow.models.asset import AssetActive, AssetEvent, AssetModel
from airflow.utils.state import DagRunState, State
-from tests_common.test_utils.db import clear_db_runs
+from tests_common.test_utils.db import clear_db_assets, clear_db_runs
pytestmark = pytest.mark.db_test
@@ -41,8 +42,8 @@ RUN_PATCH_BODY = {
@pytest.fixture
def old_ver_client(client):
- """Execution API version immediately before ``partition_date`` was
added."""
- client.headers["Airflow-API-Version"] = "2026-06-16"
+ """Last released execution API before this bundle -- the version Task SDK
1.2.x sends."""
+ client.headers["Airflow-API-Version"] = "2026-04-06"
return client
@@ -94,3 +95,78 @@ class TestPartitionDateFieldBackwardCompat:
dag_run = response.json()["dag_run"]
assert dag_run["partition_key"] == "2026-05-20"
assert dag_run["partition_date"] ==
PARTITION_DATE.isoformat().replace("+00:00", "Z")
+
+
+class TestConsumedEventPartitionKeyBackwardCompat:
+ """``partition_key`` on consumed asset events is stripped for clients
older than 2026-06-30."""
+
+ @pytest.fixture(autouse=True)
+ def _freeze_time(self, time_machine):
+ time_machine.move_to(TIMESTAMP_STR, tick=False)
+
+ def setup_method(self):
+ clear_db_runs()
+ clear_db_assets()
+
+ def teardown_method(self):
+ clear_db_runs()
+ clear_db_assets()
+
+ def _create_ti_with_consumed_event(self, session, create_task_instance):
+ ti = create_task_instance(
+ task_id="test_consumed_event_partition_key_compat",
+ state=State.QUEUED,
+ session=session,
+ start_date=TIMESTAMP,
+ )
+ asset = AssetModel(name="upstream", uri="s3://bucket/upstream",
group="asset", extra={})
+ session.add_all([asset, AssetActive.for_asset(asset)])
+ session.flush()
+ ti.dag_run.partition_key = "2026-05-20"
+ ti.dag_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()
+ return ti
+
+ def test_old_version_strips_event_key_but_keeps_dag_run_key(
+ self, old_ver_client, session, create_task_instance
+ ):
+ ti = self._create_ti_with_consumed_event(session, create_task_instance)
+
+ response =
old_ver_client.patch(f"/execution/task-instances/{ti.id}/run",
json=RUN_PATCH_BODY)
+
+ assert response.status_code == 200
+ dag_run = response.json()["dag_run"]
+ events = dag_run["consumed_asset_events"]
+ assert events
+ assert all("partition_key" not in event for event in events)
+ # The DagRun-level field really was released at 2026-04-06 and must
survive.
+ assert dag_run["partition_key"] == "2026-05-20"
+
+ @pytest.mark.parametrize("api_version", [None, "2026-06-30"])
+ def test_partition_key_kept_at_2026_06_30_and_newer(
+ self, client, session, create_task_instance, api_version
+ ):
+ """The gate must not over-move: 1.3.x clients send 2026-06-30 and do
carry this field."""
+ if api_version is not None:
+ client.headers["Airflow-API-Version"] = api_version
+ ti = self._create_ti_with_consumed_event(session, create_task_instance)
+
+ response = client.patch(f"/execution/task-instances/{ti.id}/run",
json=RUN_PATCH_BODY)
+
+ assert response.status_code == 200
+ dag_run = response.json()["dag_run"]
+ assert [event["partition_key"] for event in
dag_run["consumed_asset_events"]] == ["2024-01-15"]
+ assert dag_run["partition_key"] == "2026-05-20"
+
+ def test_schema_gates_event_partition_key_at_2026_06_30(self, client):
+ """
+ The schema half of the gate must move with the converter half, or the
served
+ OpenAPI contradicts the wire format.
+ """
+ old_schema =
client.get("/execution/openapi.json?version=2026-04-06").json()["components"]["schemas"]
+ new_schema =
client.get("/execution/openapi.json?version=2026-06-30").json()["components"]["schemas"]
+
+ assert "partition_key" not in
old_schema["AssetEventDagRunReference"]["properties"]
+ assert "partition_key" in
new_schema["AssetEventDagRunReference"]["properties"]
