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 95a48f50c17 [v3-3-test] Fix asset event ingestion crash for Dags using
FixedKeyMapper (#69311) (#69326)
95a48f50c17 is described below
commit 95a48f50c170b1340881cc945f579ef21cd755c1
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Fri Jul 3 17:04:03 2026 +0530
[v3-3-test] Fix asset event ingestion crash for Dags using FixedKeyMapper
(#69311) (#69326)
* Fix asset event ingestion crash for Dags using FixedKeyMapper
Core FixedKeyMapper's generated __init__ never called
PartitionMapper.__init__, so instances had no max_downstream_keys attribute.
Since the Execution API reads mapper.max_downstream_keys with no surrounding
error handling when queuing partitioned Dag runs, every producer task emitting
an asset event consumed by a Dag with a bare FixedKeyMapper failed to record
SUCCESS and became a zombie.
* fixup! Fix asset event ingestion crash for Dags using FixedKeyMapper
(cherry picked from commit 0643e33e3f287bbf7eadc38f8cf091edc6d99e3c)
Co-authored-by: Wei Lee <[email protected]>
---
airflow-core/src/airflow/partition_mappers/base.py | 4 ++++
.../src/airflow/partition_mappers/fixed_key.py | 23 ++++++++++++----------
airflow-core/src/airflow/serialization/encoders.py | 5 ++++-
.../tests/unit/partition_mappers/test_base.py | 14 +++++++++++++
.../tests/unit/partition_mappers/test_fixed_key.py | 20 +++++++++++++++++++
.../sdk/definitions/partition_mappers/fixed_key.py | 2 +-
6 files changed, 56 insertions(+), 12 deletions(-)
diff --git a/airflow-core/src/airflow/partition_mappers/base.py
b/airflow-core/src/airflow/partition_mappers/base.py
index f9d05b6d8d3..b0edd02f645 100644
--- a/airflow-core/src/airflow/partition_mappers/base.py
+++ b/airflow-core/src/airflow/partition_mappers/base.py
@@ -38,6 +38,10 @@ class PartitionMapper(ABC):
is_rollup: ClassVar[bool] = False
+ # Class-level default so the attribute resolves even on subclasses whose
+ # __init__ (e.g. attrs-generated in plugins) never calls this base
__init__.
+ max_downstream_keys: int | None = None
+
def __init__(self, *, max_downstream_keys: int | None = None) -> None:
if max_downstream_keys is not None and (
not isinstance(max_downstream_keys, int) or max_downstream_keys < 1
diff --git a/airflow-core/src/airflow/partition_mappers/fixed_key.py
b/airflow-core/src/airflow/partition_mappers/fixed_key.py
index 89388bfed12..c77d9d695ca 100644
--- a/airflow-core/src/airflow/partition_mappers/fixed_key.py
+++ b/airflow-core/src/airflow/partition_mappers/fixed_key.py
@@ -18,12 +18,9 @@ from __future__ import annotations
from typing import Any
-import attrs
-
from airflow.partition_mappers.base import PartitionMapper
[email protected]
class FixedKeyMapper(PartitionMapper):
"""
Collapse every upstream partition key onto one fixed downstream key.
@@ -46,20 +43,26 @@ class FixedKeyMapper(PartitionMapper):
:raises ValueError: if *downstream_key* is not a non-empty ``str``.
"""
- downstream_key: str = attrs.field()
+ downstream_key: str
- @downstream_key.validator
- def _validate_downstream_key(self, attribute: attrs.Attribute, value: str)
-> None:
- if not isinstance(value, str) or value == "":
- raise ValueError(f"FixedKeyMapper downstream_key must be a
non-empty str; got {value!r}.")
+ def __init__(self, downstream_key: str, *, max_downstream_keys: int | None
= None) -> None:
+ if not downstream_key or not isinstance(downstream_key, str):
+ raise ValueError(
+ f"FixedKeyMapper downstream_key must be a non-empty str; got
{downstream_key!r}."
+ )
+ super().__init__(max_downstream_keys=max_downstream_keys)
+ self.downstream_key = downstream_key
def to_downstream(self, key: str) -> str:
"""Return the fixed downstream key regardless of *key*."""
return self.downstream_key
def serialize(self) -> dict[str, Any]:
- return {"downstream_key": self.downstream_key}
+ data: dict[str, Any] = {"downstream_key": self.downstream_key}
+ if self.max_downstream_keys is not None:
+ data["max_downstream_keys"] = self.max_downstream_keys
+ return data
@classmethod
def deserialize(cls, data: dict[str, Any]) -> FixedKeyMapper:
- return cls(data["downstream_key"])
+ return cls(data["downstream_key"],
max_downstream_keys=data.get("max_downstream_keys"))
diff --git a/airflow-core/src/airflow/serialization/encoders.py
b/airflow-core/src/airflow/serialization/encoders.py
index 8ab533b888a..ab37b78550d 100644
--- a/airflow-core/src/airflow/serialization/encoders.py
+++ b/airflow-core/src/airflow/serialization/encoders.py
@@ -481,7 +481,10 @@ class _Serializer:
@serialize_partition_mapper.register
def _(self, partition_mapper: FixedKeyMapper) -> dict[str, Any]:
- return {"downstream_key": partition_mapper.downstream_key}
+ data: dict[str, Any] = {"downstream_key":
partition_mapper.downstream_key}
+ if partition_mapper.max_downstream_keys is not None:
+ data["max_downstream_keys"] = partition_mapper.max_downstream_keys
+ return data
@serialize_partition_mapper.register(StartOfHourMapper)
@serialize_partition_mapper.register(StartOfDayMapper)
diff --git a/airflow-core/tests/unit/partition_mappers/test_base.py
b/airflow-core/tests/unit/partition_mappers/test_base.py
index ba48057cc12..5e730e49dd8 100644
--- a/airflow-core/tests/unit/partition_mappers/test_base.py
+++ b/airflow-core/tests/unit/partition_mappers/test_base.py
@@ -181,6 +181,20 @@ class TestPartitionMapperMaxDownstreamKeysValidator:
):
IdentityMapper(max_downstream_keys=bad_value)
+ def test_subclass_skipping_base_init_still_resolves_to_none(self):
+ """A subclass whose __init__ never calls the base __init__ (e.g. an
+ attrs-generated one in a plugin) must still resolve max_downstream_keys
+ via the class-level default instead of raising AttributeError."""
+
+ class NoSuperMapper(PartitionMapper):
+ def __init__(self):
+ pass
+
+ def to_downstream(self, key: str) -> str:
+ return key
+
+ assert NoSuperMapper().max_downstream_keys is None
+
class TestRollupMapperMaxDownstreamKeys:
def test_max_downstream_keys_encode_decode_roundtrip(self):
diff --git a/airflow-core/tests/unit/partition_mappers/test_fixed_key.py
b/airflow-core/tests/unit/partition_mappers/test_fixed_key.py
index efdd5be3313..2b80fdf0987 100644
--- a/airflow-core/tests/unit/partition_mappers/test_fixed_key.py
+++ b/airflow-core/tests/unit/partition_mappers/test_fixed_key.py
@@ -66,6 +66,26 @@ class TestFixedKeyMapper:
assert isinstance(restored, FixedKeyMapper)
assert restored.downstream_key == "bucket"
+ def test_max_downstream_keys_defaults_to_none(self):
+ assert FixedKeyMapper("all").max_downstream_keys is None
+
+ def test_constructor_accepts_max_downstream_keys(self):
+ m = FixedKeyMapper("all", max_downstream_keys=5)
+ assert m.max_downstream_keys == 5
+
+ def
test_serialize_deserialize_round_trip_carries_max_downstream_keys(self):
+ m = FixedKeyMapper("bucket", max_downstream_keys=3)
+ restored = FixedKeyMapper.deserialize(m.serialize())
+ assert restored.max_downstream_keys == 3
+
+ def
test_sdk_encode_core_decode_round_trip_carries_max_downstream_keys(self):
+ # Dag-author code constructs the SDK class; the scheduler encodes it
via
+ # encoders.py and decodes it into the core class above.
+ sdk_mapper = SdkFixedKeyMapper("all_regions", max_downstream_keys=7)
+ restored = decode_partition_mapper(encode_partition_mapper(sdk_mapper))
+ assert isinstance(restored, FixedKeyMapper)
+ assert restored.max_downstream_keys == 7
+
class TestCategoricalRollupEquivalence:
"""RollupMapper(FixedKeyMapper, SegmentWindow) behaves like old
SegmentMapper."""
diff --git
a/task-sdk/src/airflow/sdk/definitions/partition_mappers/fixed_key.py
b/task-sdk/src/airflow/sdk/definitions/partition_mappers/fixed_key.py
index bf2b07905a8..cb4264f4000 100644
--- a/task-sdk/src/airflow/sdk/definitions/partition_mappers/fixed_key.py
+++ b/task-sdk/src/airflow/sdk/definitions/partition_mappers/fixed_key.py
@@ -44,7 +44,7 @@ class FixedKeyMapper(PartitionMapper):
@downstream_key.validator
def _validate_downstream_key(self, attribute: attrs.Attribute, value: str)
-> None:
- if not isinstance(value, str) or value == "":
+ if not value or not isinstance(value, str):
raise ValueError(f"FixedKeyMapper downstream_key must be a
non-empty str; got {value!r}.")
def to_downstream(self, key: str) -> str:
