eitan-shalev opened a new issue, #72588:
URL: https://github.com/apache/airflow/issues/72588
## Problem
A custom operator can override a `BaseOperator` constructor default for
direct task construction, but the overridden default is lost when the same
operator is dynamically mapped.
`BaseOperator.partial()` fills missing values in `partial_kwargs` from the
hard-coded `OPERATOR_DEFAULTS` mapping. `MappedOperator.unmap()` then passes
those values explicitly to the custom operator constructor. A base-class
default therefore overrides the custom operator's Python default.
This makes a task behave differently when it is mapped, even when no mapped
argument or DAG default touches the property.
## Reproducer
Tested with Apache Airflow 3.3.1.
```python
from airflow.sdk import BaseOperator, DAG, ExceptionRetryPolicy
CUSTOM_POLICY = ExceptionRetryPolicy(rules=[])
class CustomRetryOperator(BaseOperator):
def __init__(self, *, value: str, retry_policy=CUSTOM_POLICY, **kwargs):
super().__init__(retry_policy=retry_policy, **kwargs)
self.value = value
def execute(self, context):
pass
with DAG(dag_id="mapped_custom_default", schedule=None):
direct = CustomRetryOperator(task_id="direct", value="direct")
mapped =
CustomRetryOperator.partial(task_id="mapped").expand(value=["mapped"])
unmapped = mapped.unmap({"value": "mapped"})
assert direct.retry_policy is CUSTOM_POLICY
assert mapped.partial_kwargs["retry_policy"] is None
assert unmapped.retry_policy is CUSTOM_POLICY # Fails: actual value is None
```
The same pattern should affect any property whose subclass default differs
from the corresponding entry in `OPERATOR_DEFAULTS`, not only `retry_policy`.
## Expected behavior
If a property is not explicitly supplied to `.partial()`, `.expand()`, or
DAG/task `default_args`, an unmapped task instance should receive the same
custom-operator default as a directly constructed task.
Explicit partial arguments and DAG/task defaults should keep their current
precedence.
## Suggested direction
When constructing `partial_kwargs`, resolve absent defaults from the
effective `operator_class.__init__` signature for parameters declared by that
class, instead of unconditionally materializing the `BaseOperator` value from
`OPERATOR_DEFAULTS`.
An alternative with a smaller compatibility surface would be a documented
class hook for custom operators to provide mapping-time defaults. In either
approach, add a regression test for a custom operator that overrides a
BaseOperator field default and assert that direct and mapped/unmapped
construction produce the same value.
## Relevant implementation
- `task-sdk/src/airflow/sdk/bases/operator.py`: `partial()` fills
`OPERATOR_DEFAULTS` into `partial_kwargs`.
- `task-sdk/src/airflow/sdk/definitions/mappedoperator.py`: `unmap()`
re-instantiates the custom operator from those kwargs.
--
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]
