Skip to content

Fix task config defaults for language SDKs #75267

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
methatp0 wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Fix task config defaults for language SDKs #75267

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 11 additions & 0 deletions airflow-core/adr/lang-sdk/0004-dag-parsing.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -231,6 +231,17 @@ The language runtime must produce a `DagFileParsingResult` that matches Python A
| `@continuous` | `airflow.timetables.simple.ContinuousTimetable` | `{}` |
| cron expr | `airflow.timetables.trigger.CronTriggerTimetable` | `{expression, timezone, interval, run_immediately}` |

**Task config defaults:** A runtime writes every task field the author explicitly set, even when
its value equals the schema default. A missing task field is filled from the optional
`client_defaults.tasks` table. Because language runtimes cannot read Airflow's configuration,
core supplies this table from its own config when the payload does not already include it. Only
config-backed task fields are supplied (`owner`, `queue`, `retries`, `retry_delay`,
`weight_rule`, `execution_timeout`, `email_on_failure`, and `email_on_retry`). A task field
present in the serialized task takes precedence, so an explicit `retries=0` is not replaced by
a non-zero `[core] default_task_retries`. The Go SDK also preserves explicit schema-default values
when its spec represents them as set; its non-pointer scalar zero values are treated as unset and
receive the configured default.
Comment on lines +234 to +243

@jason810496 jason810496 Oct 21, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Worth stating as one rule rather than a Go exception.

Suggested change
**Task config defaults:** A runtime writes every task field the author explicitly set, even when
its value equals the schema default. A missing task field is filled from the optional
`client_defaults.tasks` table. Because language runtimes cannot read Airflow's configuration,
core supplies this table from its own config when the payload does not already include it. Only
config-backed task fields are supplied (`owner`, `queue`, `retries`, `retry_delay`,
`weight_rule`, `execution_timeout`, `email_on_failure`, and `email_on_retry`). A task field
present in the serialized task takes precedence, so an explicit `retries=0` is not replaced by
a non-zero `[core] default_task_retries`. The Go SDK also preserves explicit schema-default values
when its spec represents them as set; its non-pointer scalar zero values are treated as unset and
receive the configured default.
**Task config defaults:** A runtime writes every task field the author set and leaves out the
ones they did not; a value equal to the schema default is still written. A missing task field is
filled from the optional `client_defaults.tasks` table, which core supplies from its own config
because a language runtime cannot read it. Only config-backed task fields are supplied (`owner`,
`queue`, `retries`, `retry_delay`, `weight_rule`, `execution_timeout`, `email_on_failure` and
`email_on_retry`). A field present in the serialized task wins over the table, so an explicit
`retries=0` is not replaced by a non-zero `[core] default_task_retries`. In Go a non-pointer
scalar at its zero value counts as unset and takes the configured default; a pointer field
expresses an explicit zero.


**Task encoding:**

```
Expand Down
34 changes: 30 additions & 4 deletions airflow-core/src/airflow/serialization/serialized_objects.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2153,11 +2153,11 @@ def from_dict(cls, serialized_obj: dict) -> SerializedDAG:
@classmethod
def fill_config_defaults(cls, serialized_obj: dict[str, Any]) -> None:
"""
Fill in the Dag settings a serialized Dag leaves unset from the Airflow config, as a Python Dag does.
Fill config-backed defaults for a serialized Lang-SDK Dag, as a Python Dag does.

A Lang-SDK runtime cannot read the Airflow config, so it leaves ``max_active_tasks``,
``max_active_runs``, ``max_consecutive_failed_dag_runs``, ``catchup`` and
``disable_bundle_versioning`` out unless the Dag sets them. A value the Dag sets is kept.
A Lang-SDK runtime cannot read the Airflow config, so it leaves config-backed Dag settings out
unless the Dag sets them. Airflow fills those settings and supplies config-backed task defaults
through ``client_defaults.tasks``, without replacing task fields the SDK explicitly serialized.
*serialized_obj* is changed in place.
"""
dag = serialized_obj.get("dag")
Expand All @@ -2179,6 +2179,32 @@ def fill_config_defaults(cls, serialized_obj: dict[str, Any]) -> None:
if key not in dag:
dag[key] = get(section, option)

# Native runtimes cannot read Airflow's task defaults. Supply only the config-backed
# fields the Python serializer exposes through client_defaults, so explicit values in
# the task itself (including values equal to schema defaults) retain precedence.
task_default_keys = {
"owner",
"queue",
"retries",
"retry_delay",
"weight_rule",
"execution_timeout",
"email_on_failure",
"email_on_retry",
}
task_defaults = {
key: value
for key, value in OperatorSerialization.generate_client_defaults().items()
if key in task_default_keys
}
Comment on lines +2182 to +2199

@jason810496 jason810496 Oct 21, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The whitelist is a second copy of knowledge generate_client_defaults() already has: it drops every OPERATOR_DEFAULTS value equal to its schema default, which today leaves exactly these eight keys (pool_slots and priority_weight are hardcoded 1, allow_nested_operators is True on both sides, the rest are None or empty). Keeping the list means a native Dag silently diverges from a Python Dag the day a new config-backed task default lands in OPERATOR_DEFAULTS.

Suggested change
# Native runtimes cannot read Airflow's task defaults. Supply only the config-backed
# fields the Python serializer exposes through client_defaults, so explicit values in
# the task itself (including values equal to schema defaults) retain precedence.
task_default_keys = {
"owner",
"queue",
"retries",
"retry_delay",
"weight_rule",
"execution_timeout",
"email_on_failure",
"email_on_retry",
}
task_defaults = {
key: value
for key, value in OperatorSerialization.generate_client_defaults().items()
if key in task_default_keys
}
# Native runtimes cannot read Airflow's task defaults, so supply the table a Python Dag
# carries. generate_client_defaults() already drops every value equal to its schema
# default, which leaves exactly the config-backed fields.
task_defaults = OperatorSerialization.generate_client_defaults()

if task_defaults:
client_defaults = serialized_obj.setdefault("client_defaults", {})
if isinstance(client_defaults, dict):
existing_task_defaults = client_defaults.setdefault("tasks", {})
if isinstance(existing_task_defaults, dict):
Comment on lines +2200 to +2204

@jason810496 jason810496 Oct 21, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No need the if task_defaults: and if isinstance here, we can assert the client_defaults is correct as they're Airflow core module.

for key, value in task_defaults.items():
existing_task_defaults.setdefault(key, value)

@classmethod
def validate_serialized_dag(cls, serialized_obj: dict[str, Any]) -> SerializedDAG:
"""
Expand Down
71 changes: 71 additions & 0 deletions airflow-core/tests/unit/serialization/test_dag_serialization.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5466,6 +5466,77 @@ def test_keeps_a_setting_the_dag_sets(self):

assert data == before

def test_fills_unset_task_defaults_from_config(self, monkeypatch):
task_defaults = {
"owner": "configured-owner",
"queue": "celery_q",
"retries": 3,
"retry_delay": 240.0,
"weight_rule": "upstream",
"execution_timeout": 600.0,
"email_on_failure": False,
"email_on_retry": True,
"pool": "default_pool",
"priority_weight": 10,
}
monkeypatch.setattr(
OperatorSerialization,
"generate_client_defaults",
classmethod(lambda _: task_defaults),
)
task = {"__type": "operator", "__var": {"task_id": "explicit", "language": "typescript", "retries": 0}}
data = {"__version": 3, "dag": {"dag_id": "native_dag", "tasks": [task]}}

DagSerialization.fill_config_defaults(data)

assert data["client_defaults"] == {
"tasks": {
"owner": "configured-owner",
"queue": "celery_q",
"retries": 3,
"retry_delay": 240.0,
"weight_rule": "upstream",
"execution_timeout": 600.0,
"email_on_failure": False,
"email_on_retry": True,
}
}
assert data["dag"]["tasks"][0]["__var"]["retries"] == 0

def test_merges_task_defaults_without_replacing_explicit_defaults(self, monkeypatch):
monkeypatch.setattr(
OperatorSerialization,
"generate_client_defaults",
classmethod(lambda _: {"retries": 3, "queue": "celery_q"}),
)
data = {
"__version": 3,
"dag": {"dag_id": "native_dag", "tasks": [{"__type": "operator", "__var": {"task_id": "explicit", "language": "java"}}]},
"client_defaults": {"tasks": {"retries": 1}},
}

DagSerialization.fill_config_defaults(data)

assert data["client_defaults"] == {"tasks": {"retries": 1, "queue": "celery_q"}}

def test_fills_task_defaults_for_go_sdk_payloads_too(self, monkeypatch):
monkeypatch.setattr(
OperatorSerialization,
"generate_client_defaults",
classmethod(lambda _: {"retries": 3, "queue": "celery_q"}),
)
data = {
"__version": 3,
"dag": {
"dag_id": "native_dag",
"tasks": [{"__type": "operator", "__var": {"task_id": "task", "language": "go"}}],
},
}

DagSerialization.fill_config_defaults(data)

assert data["client_defaults"] == {"tasks": {"retries": 3, "queue": "celery_q"}}

@jason810496 jason810496 Oct 21, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We should also add a new test that define a real Dag with some operators like in #74869 to exercise the set default behavior.

def test_leaves_a_payload_without_a_dag_to_validation(self):
data = {"__version": 3}

Expand Down
16 changes: 10 additions & 6 deletions go-sdk/airflow/serialize.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -202,6 +202,8 @@ type specRules struct {
// set names the lists that Python keeps in a set, so that Python writes the list sorted and
// without repeats. writeSpecFields writes it the same way.
set map[string]bool
// keepDefaults preserves an explicitly set task field when it equals the schema default.
keepDefaults bool
Comment on lines +205 to +206

@jason810496 jason810496 Oct 21, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

A per-SDK flag for a rule that should hold everywhere. Dropping it here, in taskSpecRules and in the write condition leaves schemaDefault and isSameJSON with no reader.

Suggested change
// keepDefaults preserves an explicitly set task field when it equals the schema default.
keepDefaults bool

// skip names the fields that writeSpecFields never writes. The comment on each says why.
skip map[string]bool
}
Expand All @@ -218,7 +220,8 @@ var dagSpecRules = specRules{
}

var taskSpecRules = specRules{
fields: taskSpecFields,
fields: taskSpecFields,
keepDefaults: true,
Comment on lines +223 to +224

@jason810496 jason810496 Oct 21, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
fields: taskSpecFields,
keepDefaults: true,
fields: taskSpecFields,

skip: map[string]bool{
// Python writes email_on_failure and email_on_retry only for an operator that has an email
// recipient, and a TaskSpec has no field for a recipient.
Expand All @@ -234,10 +237,11 @@ var taskSpecRules = specRules{
// from the defaults of Python's TaskGroup instead.
var taskGroupSpecRules = specRules{fields: taskGroupSpecFields}

// writeSpecFields writes into data each field of spec that rules does not skip. It leaves out a
// field that is unset or that holds its schema default. A field is unset when it holds its zero
// value. A pointer field is unset only when it is nil, so a pointer field can set the zero value of
// the type it points to, such as false or 0.
// writeSpecFields writes into data each field of spec that rules does not skip. It leaves out
// fields that are unset, and normally those that hold their schema default. Task rules preserve
// explicitly set schema-default values so Airflow config defaults cannot replace them. A field
// is unset when it holds its zero value. A pointer field is unset only when nil, so it can explicitly
// set the zero value of its type, such as false or 0.
Comment on lines +240 to +244

@jason810496 jason810496 Oct 21, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
// writeSpecFields writes into data each field of spec that rules does not skip. It leaves out
// fields that are unset, and normally those that hold their schema default. Task rules preserve
// explicitly set schema-default values so Airflow config defaults cannot replace them. A field
// is unset when it holds its zero value. A pointer field is unset only when nil, so it can explicitly
// set the zero value of its type, such as false or 0.
// writeSpecFields writes into data each field of spec that rules does not skip. It leaves out only
// the fields that are unset: Airflow fills an unset config-backed field from its own config, so a
// value the author set has to be written even when it equals the schema default. A field is unset
// when it holds its zero value. A pointer field is unset only when nil, so it can explicitly set
// the zero value of its type, such as false or 0.

func writeSpecFields(data map[string]any, spec any, rules specRules) {
value := reflect.ValueOf(spec)
for i := range value.NumField() {
Expand All @@ -252,7 +256,7 @@ func writeSpecFields(data map[string]any, spec any, rules specRules) {
continue
}
encoded, set := encodeSpecValue(value.Field(i), rules.set[name])
if !set || (field.schemaDefault != nil && isSameJSON(encoded, field.schemaDefault)) {
if !set || (!rules.keepDefaults && field.schemaDefault != nil && isSameJSON(encoded, field.schemaDefault)) {

@jason810496 jason810496 Oct 21, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
if !set || (!rules.keepDefaults && field.schemaDefault != nil && isSameJSON(encoded, field.schemaDefault)) {
if !set {

continue
}
data[field.key] = encoded
Expand Down
14 changes: 12 additions & 2 deletions go-sdk/airflow/serialize_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -283,7 +283,7 @@ func TestSerializeWritesTheFieldsOfTaskSpec(t *testing.T) {
}`, withoutKeys(got, goTaskFields...))
}

func TestSerializeLeavesOutAFieldAtItsSchemaDefault(t *testing.T) {
func TestSerializePreservesExplicitTaskFieldsAtTheirSchemaDefaults(t *testing.T) {
dag := Dag("etl", DagSpec{Queue: "default"})
dag.Task(ping, TaskSpec{
DoXComPush: ptr(true),
Expand All @@ -300,7 +300,17 @@ func TestSerializeLeavesOutAFieldAtItsSchemaDefault(t *testing.T) {

got := serializedDag(t, dag)

assert.Empty(t, withoutKeys(serializedTask(t, got, "ping"), goTaskFields...))
assertJSON(t, `{
"do_xcom_push": true,
"owner": "airflow",
"pool": "default_pool",
"pool_slots": 1,
"priority_weight": 1,
"queue": "default",
"retry_delay": 300,
"trigger_rule": "all_success",
"weight_rule": "downstream"
}`, withoutKeys(serializedTask(t, got, "ping"), goTaskFields...))
assert.Empty(t, withoutKeys(serializedTask(t, got, "extract"), goTaskFields...))
}

Expand Down
33 changes: 5 additions & 28 deletions java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Serde.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,6 @@

package org.apache.airflow.sdk.execution

import com.fasterxml.jackson.databind.ObjectMapper
import org.apache.airflow.sdk.Bundle
import org.apache.airflow.sdk.DagDef
import org.apache.airflow.sdk.EdgeLabel
Expand All @@ -32,8 +31,6 @@ import org.apache.airflow.sdk.TaskGroupRef
import org.apache.airflow.sdk.TaskRef
import org.apache.airflow.sdk.TriggerDagRun
import org.apache.airflow.sdk.execution.comm.DagFileParseRequest
import org.apache.airflow.sdk.internal.Field
import org.apache.airflow.sdk.internal.SchemaFields
import java.nio.file.InvalidPathException
import java.nio.file.Paths
import java.time.Duration
Expand All @@ -44,8 +41,6 @@ import java.time.OffsetDateTime
// SDK's serde (ts-sdk/src/coordinator/serde.ts), which in turn matches Python's
// DagSerialization output.

private val defaultsMapper = ObjectMapper()

// Python drops both unless the operator names an email recipient. A Java task
// has no email field to name one, so they are never written.
private val OMITTED_TASK_KEYS = setOf("email_on_failure", "email_on_retry")
Expand Down Expand Up @@ -150,14 +145,12 @@ private fun serializeTask(
}
// Lets NotPreviouslySkippedDep re-skip a cleared downstream, as Python's SkipMixin does.
if (def.decider != null) data["_can_skip_downstream"] = true
// Emit only config entries that differ from their schema default, mirroring
// Python BaseSerialization's "omit hard-coded default" behavior, which the Go
// and TypeScript SDKs mirror too. Operator fields are stored unwrapped, so the
// __type encoding is stripped. If core grows a task-level fill_config_defaults,
// every SDK has to keep explicitly set values instead, or an explicit retries=0
// reads as unset and picks up the configured default.
// Keep every config value the author set, even at a schema default. Airflow fills
// unset task fields from client_defaults.tasks, so omitting an explicit retries=0
// here could incorrectly replace it with the configured default. Operator fields
// are stored unwrapped, so the __type encoding is stripped.
def.configValues.forEach { (key, value) ->
if (key !in OMITTED_TASK_KEYS && !matchesSchemaDefault(SchemaFields.TASK[key], value)) {
if (key !in OMITTED_TASK_KEYS) {
data[key] = unwrapTypeEncoding(serializeValue(value))
}
}
Expand Down Expand Up @@ -599,22 +592,6 @@ internal fun unwrapTypeEncoding(value: Any?): Any? {
return if ("__var" in map) map["__var"] else value
}

/** Whether a config value equals the schema default and can be omitted. */
private fun matchesSchemaDefault(
field: Field?,
value: Any,
): Boolean {
val defaultJson = field?.defaultJson ?: return false
val node = defaultsMapper.readTree(defaultJson)
return when (value) {
is String -> node.isTextual && node.asText() == value
is Boolean -> node.isBoolean && node.asBoolean() == value
is Number -> node.isNumber && node.asDouble() == value.toDouble()
is Duration -> node.isNumber && node.asDouble() == value.toNanos() / 1e9
else -> false
}
}

private fun computeRelativeFileloc(
fileloc: String,
bundlePath: String?,
Expand Down
20 changes: 16 additions & 4 deletions java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/execution/SerdeTest.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -139,6 +139,18 @@ internal class SerdeTest {
assertEquals(1.7672256E9, serialized["start_date"])
}

@Test
@DisplayName("Should leave unset task config fields out for Airflow to fill")
fun shouldLeaveUnsetTaskConfigFieldsOut() {
val task = TaskDef("task", SerdeNoopTask::class.java)
val serialized = serializeDag(DagDef("d").addTask(task), "", ".")
val data = taskData(serialized, 0)

for (key in listOf("retries", "queue", "retry_delay", "weight_rule", "execution_timeout")) {
assertFalse(key in data, "$key should be omitted when unset")
}
}

@Test
@DisplayName("Should serialize tasks with identity fields, config, and sorted downstream ids")
fun shouldSerializeTasks() {
Expand All @@ -152,7 +164,7 @@ internal class SerdeTest {
val transformDef =
TaskDef("transform", SerdeNoopTask::class.java)
.dependsOn(extractDef)
// Explicitly at schema defaults: omitted from the serialized form.
// Explicitly set at schema defaults: these must override Airflow's config defaults.
.config("retries", 0)
.config("queue", "default")
.config("retry_delay", Duration.ofMinutes(5))
Expand All @@ -175,9 +187,9 @@ internal class SerdeTest {

val transform = taskData(serialized, 1)
assertEquals("transform", transform["task_id"])
assertFalse("retries" in transform)
assertFalse("queue" in transform)
assertFalse("retry_delay" in transform)
assertEquals(0, transform["retries"])
assertEquals("default", transform["queue"])
assertEquals(300.0, transform["retry_delay"])
assertFalse("downstream_task_ids" in transform)

assertEquals(
Expand Down
9 changes: 5 additions & 4 deletions scripts/ci/lang_sdk_serialization/compare.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -206,16 +206,17 @@ def compare_fields(
"""
Compare two serialized objects key by key.

A key the SDK leaves out is fine when Python wrote its schema default. Python keeps such a value
when its ``client_defaults`` table disagrees with the schema, a table a language SDK is never sent,
and Airflow reads a missing field as its default anyway.
A key either side leaves out is fine when the value the other side writes equals the schema
default. Python may omit a value matching its ``client_defaults`` table, and native SDKs now preserve
explicitly set task values even when they equal the schema default.
Comment on lines +209 to +211

@jason810496 jason810496 Oct 21, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

compare_dag still passes {} as the Dag-level defaults, so once the Dag path writes explicit defaults too, a fail_fast: false the author set is reported as a key Python does not write. It needs a Dag defaults table built the way get_task_defaults() builds the operator one.

Suggested change
A key either side leaves out is fine when the value the other side writes equals the schema
default. Python may omit a value matching its ``client_defaults`` table, and native SDKs now preserve
explicitly set task values even when they equal the schema default.
A key either side leaves out is fine when the value the other side writes equals the schema
default: Python omits a value matching its ``client_defaults`` table, and an SDK writes every
field the author set, including one that happens to equal the default.

"""
problems = []
for key in sorted((python.keys() | sdk.keys()) - not_compared):
if key in python and key in sdk:
problems.extend(find_differences(key, python[key], sdk[key]))
elif key in sdk:
problems.append(f"{key} is {sdk[key]!r}, which Python does not write")
if key not in defaults or not is_same_json(sdk[key], defaults[key]):
problems.append(f"{key} is {sdk[key]!r}, which Python does not write")
elif key not in defaults or not is_same_json(python[key], defaults[key]):
problems.append(f"{key} is missing, Python writes {python[key]!r}")
return problems
Expand Down
7 changes: 7 additions & 0 deletions scripts/tests/ci/lang_sdk_serialization/test_compare.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -89,6 +89,13 @@ def test_accepts_the_differences_an_sdk_is_allowed():
assert compare(PYTHON, SDK, DEFAULTS) == []


def test_accepts_an_explicit_task_field_at_its_schema_default():
sdk = copy.deepcopy(SDK)
get_task(sdk, index=1)["owner"] = get_task_defaults()["owner"]

assert compare(PYTHON, sdk, get_task_defaults()) == []


@pytest.mark.parametrize(
("change", "expected"),
[
Expand Down
Loading