This is an automated email from the ASF dual-hosted git repository.
jason810496 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new bc28bb0804c Go SDK: Fix the mixed language task handler binding
semantic (#73648)
bc28bb0804c is described below
commit bc28bb0804c4c9d3fec2dc0c636e739f823a213c
Author: Jason(Zhe-You) Liu <[email protected]>
AuthorDate: Tue Sep 29 16:40:35 2026 +0800
Go SDK: Fix the mixed language task handler binding semantic (#73648)
* Go SDK: require every struct field to bind, warn on extra arguments
* Go SDK: add arg:"-" opt-out, correct the README binding rules
* Go SDK: log struct name mismatches both ways instead of failing
* Go SDK: name each binding mismatch direction, exercise both e2e
* Go SDK: warn on an empty spec too, assert both warnings e2e
* Go SDK e2e: assert binding warnings by record field
---
.../0007-taskflow-across-language-boundary.md | 39 ++++++++
.../go_sdk_tests/test_go_sdk_taskflow_binding.py | 72 ++++++++++++--
go-sdk/README.md | 14 ++-
go-sdk/adr/0006-cross-language-argument-binding.md | 15 ++-
.../cmd/airflow-go-pack/pack_integration_test.go | 4 +-
go-sdk/dags/go_examples.py | 28 ++++--
go-sdk/example/bundle/main.go | 14 ++-
.../bundle/taskflowbinding/taskflowbinding.go | 61 +++++++++---
.../bundle/taskflowbinding/taskflowbinding_test.go | 44 ++++++---
go-sdk/pkg/binding/binding.go | 56 ++++++++---
go-sdk/pkg/binding/binding_test.go | 103 +++++++++++++++------
11 files changed, 360 insertions(+), 90 deletions(-)
diff --git
a/airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md
b/airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md
index b3dcbba997c..c1f9d06fcf6 100644
--- a/airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md
+++ b/airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md
@@ -105,6 +105,45 @@ Materializing is what makes the contract work at all: the
spec is the *only* thi
receives, so it must be complete and self-describing rather than a reference
the runtime is
expected to resolve on its own.
+#### What a runtime does when the two signatures disagree
+
+The spec describes the stub signature. Nothing checks it against the foreign
handler's own
+signature, in either direction, so every SDK has to answer the same two
questions the same way.
+"More" means the call passed arguments the handler does not take; "fewer"
means the handler
+declared something no argument supplies.
+
+| the handler binds | more | fewer |
+| --- | --- | --- |
+| by position | fail | fail |
+| by name | log, do not fail | log, do not fail |
+
+Positional binding fails either way because positions carry the whole meaning
of the binding: an
+argument dropped or added shifts every later one, so the handler would read
values it has mistaken
+for others. The one exception is a trailing `from_default` entry, which the
runtime may drop to
+match the handler's arity, since a captured default carries no intent from the
call site.
+
+Name binding cannot shift, so neither direction is worth failing a task over:
a field nothing fills
+takes the language's absent value, and an argument no field claims changes
nothing the handler
+reads. Both are logged instead, before the task runs, so a mistyped name
surfaces in the task log
+ahead of anything the task itself prints. `from_default` entries are not
reported, since a handler
+ignoring a captured default is the normal case.
+
+A single call can be wrong in both directions at once, so each gets its own
message rather than one
+line a reader has to untangle. SDKs implementing this use the same two
wordings, so one query finds
+the mismatch across languages:
+
+- `Dag's call passed argument(s) the task handler does not declare`
+- `Task handler declares argument(s) the Dag's call did not pass`
+
+An SDK may log more besides. The TypeScript SDK also warns when a handler
reads a name nothing
+bound, which is a run-time event rather than a signature mismatch and carries
its own wording.
+
+Where the declaration comes from differs by language and is not a difference
in the rule. Go's sole
+struct and Java's `TaskInput` are field lists the runtime can read directly.
The TypeScript SDK has
+no such list, because parameter types are erased, so it reads the handler's
destructuring pattern
+instead; a handler that takes the whole argument object has declared nothing
to compare a call
+against, and nothing is reported for it.
+
### B. Materialization belongs in core Dag serialization, behind a generic
`is_stub` flag
The spec is built in Airflow core, from
`OperatorSerialization._serialize_node`, for a non-mapped
diff --git
a/airflow-e2e-tests/tests/airflow_e2e_tests/go_sdk_tests/test_go_sdk_taskflow_binding.py
b/airflow-e2e-tests/tests/airflow_e2e_tests/go_sdk_tests/test_go_sdk_taskflow_binding.py
index d078261f8b1..d01924994c3 100644
---
a/airflow-e2e-tests/tests/airflow_e2e_tests/go_sdk_tests/test_go_sdk_taskflow_binding.py
+++
b/airflow-e2e-tests/tests/airflow_e2e_tests/go_sdk_tests/test_go_sdk_taskflow_binding.py
@@ -26,6 +26,7 @@ the tests here check the run outcome and the summary XCom it
pushes.
from __future__ import annotations
+import time
from dataclasses import dataclass
from datetime import datetime, timezone
@@ -36,6 +37,9 @@ from airflow_e2e_tests.e2e_test_utils.clients import
AirflowClient
# Allow time for coordinator startup.
_GO_TASK_TIMEOUT = 300
+# Task logs land shortly after the run finishes.
+_LOG_FETCH_TIMEOUT = 60
+
_DAG_ID = "taskflow_binding_dag"
@@ -53,6 +57,30 @@ class _CompletedRun:
"value"
)
+ def logs(self, task_id: str, try_number: int = 1) -> str:
+ """Return the concatenated task-log records for *task_id*, retrying
until present."""
+ deadline = time.monotonic() + _LOG_FETCH_TIMEOUT
+ while True:
+ text = "\n".join(str(entry) for entry in self.log_records(task_id,
try_number))
+ if text.strip() or time.monotonic() > deadline:
+ return text
+ time.sleep(3)
+
+ def log_records(self, task_id: str, try_number: int = 1) -> list[dict]:
+ """Return the structured task-log records (parsed JSON dicts) for
*task_id*."""
+ resp = self.client.get_task_logs(
+ dag_id=_DAG_ID, run_id=self.run_id, task_id=task_id,
try_number=try_number
+ )
+ return [entry for entry in resp.get("content", []) if
isinstance(entry, dict)]
+
+ def warning(self, task_id: str, event: str) -> dict:
+ """Return the first log record of *task_id* whose event is *event*,
failing if none."""
+ self.logs(task_id)
+ records = self.log_records(task_id)
+ found = next((r for r in records if r.get("event") == event), None)
+ assert found is not None, [r.get("event") for r in records]
+ return found
+
@pytest.fixture(scope="module")
def completed_run() -> _CompletedRun:
@@ -79,7 +107,9 @@ def test_all_tasks_succeeded(completed_run: _CompletedRun):
"via_flat_args",
"via_struct_no_tags",
"via_struct_arg_tag",
- "via_struct_unmatched_arg",
+ "via_struct_default_arg",
+ "via_struct_more_args",
+ "via_struct_fewer_args",
"via_flat_map",
"via_struct_map",
"via_plain_map",
@@ -137,17 +167,39 @@ def
test_via_struct_arg_tag_reflects_bound_arguments(completed_run: _CompletedRu
}
-def test_via_struct_unmatched_arg_reflects_zero_valued_field(completed_run:
_CompletedRun):
- """``via_struct_unmatched_arg`` demonstrates mismatch tolerance in both
directions:
- a struct field whose name has no corresponding TaskFlow call argument
stays at its
- Go zero value instead of failing the task (kwarg-style, an unpassed name
simply
- isn't bound), and the stub's defaulted ``sample_rate`` -- captured into
the spec as
- ``from_default`` -- needs no matching struct field. The task succeeding at
all
- proves the second half."""
- assert completed_run.xcom("via_struct_unmatched_arg") == {
+def test_via_struct_default_arg_tolerates_unclaimed_default(completed_run:
_CompletedRun):
+ """``via_struct_default_arg`` proves a captured stub default needs no
struct field:
+ ``sample_rate`` is unpassed, so the spec carries it as ``from_default``
and no Go
+ field claims it. The task succeeding at all is the assertion."""
+ assert completed_run.xcom("via_struct_default_arg") == {"region":
"eu-west-1"}
+
+
+def test_via_struct_more_args_warns_and_runs(completed_run: _CompletedRun):
+ """The call passes ``unused_label``, which the Go struct does not declare.
Name
+ binding cannot shift, so the extra argument is warned about rather than
failing
+ the task, and everything the struct does declare still binds."""
+ assert completed_run.xcom("via_struct_more_args") == {"region":
"eu-west-1"}
+ warning = completed_run.warning(
+ "via_struct_more_args", "Dag's call passed argument(s) the task
handler does not declare"
+ )
+ assert warning.get("passed_not_declared") == ["unused_label"], warning
+
+
+def test_via_struct_fewer_args_warns_and_runs(completed_run: _CompletedRun):
+ """The Go struct declares ``not_in_dag``, which the stub has no parameter
for. The
+ field keeps its Go zero value and the mismatch is warned about rather than
failing
+ the task, so the two sides can drift without breaking the Dag.
+
+ The warning is the assertion that matters: the zero-valued field alone
would look
+ the same as the older behaviour that filled it silently."""
+ assert completed_run.xcom("via_struct_fewer_args") == {
"region": "eu-west-1",
- "missing_was_empty": True,
+ "not_in_dag_was_empty": True,
}
+ warning = completed_run.warning(
+ "via_struct_fewer_args", "Task handler declares argument(s) the Dag's
call did not pass"
+ )
+ assert warning.get("declared_not_passed") == ['NotInDag (argument
"not_in_dag")'], warning
def test_via_flat_map_decodes_single_dict_whole(completed_run: _CompletedRun):
diff --git a/go-sdk/README.md b/go-sdk/README.md
index 14bed678abc..e8fa38cf3e1 100644
--- a/go-sdk/README.md
+++ b/go-sdk/README.md
@@ -161,9 +161,17 @@ An exported field binds the argument matching its own Go
name, folding case and
tag when the names genuinely differ, as `Region` does above. Declaration order
is irrelevant on both
sides, and embedded structs contribute their fields just as they do to
`encoding/json`.
-A field no argument matches is left at its Go zero value, like an unpassed
keyword argument. The
-reverse is an error: every argument the Dag author explicitly passed must land
in some field, so a
-typo'd tag fails the task instead of silently dropping the value.
+Taking **more or fewer arguments** than the Python side passes does not fail
the task. The runtime
+logs a warning before the task runs and carries on, one message per direction,
so a call that does
+both at once says so twice:
+
+- `Dag's call passed argument(s) the task handler does not declare`
+- `Task handler declares argument(s) the Dag's call did not pass`
+
+Name-based struct binding is what keeps a mixed-language task working while
the two sides drift:
+adding a parameter to the stub, or dropping a field from the struct, is a
warning rather than a
+broken Dag. A field nothing matches keeps its Go zero value. Arguments the
call left at their stub
+default are not reported.
A struct that is **not** the sole data parameter is decoded whole from its one
positional argument
instead, so `arg:` tags only apply to the sole-parameter form; pairing a
tagged struct with other
diff --git a/go-sdk/adr/0006-cross-language-argument-binding.md
b/go-sdk/adr/0006-cross-language-argument-binding.md
index cdf2f5c0060..392fbfa062c 100644
--- a/go-sdk/adr/0006-cross-language-argument-binding.md
+++ b/go-sdk/adr/0006-cross-language-argument-binding.md
@@ -108,8 +108,8 @@ ordered bindings / raw values | named bindings
/ raw values
arity must match data slots | `arg:` exact name or untagged
folded name
captured defaults may be dropped |
| claimed value -> field
- | unmatched field -> zero value
- | unclaimed explicit arg -> error
+ | unmatched field -> zero value +
warning
+ | unclaimed explicit arg -> warning
| untagged + one explicit arg + no
field claim
| -> decode whole struct
@@ -135,8 +135,15 @@ captured defaults may be dropped |
current Dag run; independent XCom pulls run concurrently and cancel together
on failure.
- Captured Python defaults may remain unclaimed. Flat bindings must match the
data-parameter
count. A sole struct claims arguments by field name, except that one
explicit argument may
- decode as the whole value when the struct is untagged and no field matches.
Remaining explicit
- arguments and incompatible schemas or Go types fail before the task body
runs.
+ decode as the whole value when the struct is untagged and no field matches.
+- Neither direction of a sole-struct name mismatch fails the task, following
the cross-language
+ rule in [lang-SDK ADR
0007](../../airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md):
+ an unfilled field keeps its zero value and an unclaimed explicit argument is
ignored, and both
+ are logged. Flat bindings still fail either way, because a dropped or added
positional argument
+ shifts every later one. Incompatible schemas or Go types still fail before
the task body runs.
+- A spec that arrives empty is the same mismatch with every field unfilled,
and warns like any
+ other. `build_arg_bindings` sends nothing at all for a stub called with no
arguments, so an
+ empty spec is an ordinary argless call rather than a signal that anything is
wrong.
- A known JSON Schema shape is checked against the Go target type before
strict JSON decoding.
Missing or unknown schema forms remain unconstrained and rely on the decoder.
- Cross-language TaskFlow binding uses the coordinator path described in
diff --git a/go-sdk/cmd/airflow-go-pack/pack_integration_test.go
b/go-sdk/cmd/airflow-go-pack/pack_integration_test.go
index f0b735ebf2c..e1cc29fe68e 100644
--- a/go-sdk/cmd/airflow-go-pack/pack_integration_test.go
+++ b/go-sdk/cmd/airflow-go-pack/pack_integration_test.go
@@ -162,7 +162,9 @@ dags:
- "via_flat_args"
- "via_struct_no_tags"
- "via_struct_arg_tag"
- - "via_struct_unmatched_arg"
+ - "via_struct_default_arg"
+ - "via_struct_more_args"
+ - "via_struct_fewer_args"
- "via_flat_map"
- "via_struct_map"
- "via_plain_map"
diff --git a/go-sdk/dags/go_examples.py b/go-sdk/dags/go_examples.py
index 46367234e4b..c87fb3a682f 100644
--- a/go-sdk/dags/go_examples.py
+++ b/go-sdk/dags/go_examples.py
@@ -151,7 +151,19 @@ def via_struct_arg_tag(region_code: str, threshold:
float): ...
@task.stub(queue="golang")
-def via_struct_unmatched_arg(region_code: str, sample_rate: float = 0.1): ...
+def via_struct_default_arg(region_code: str, sample_rate: float = 0.1): ...
+
+
+# The Go struct declares only region_code, so `unused_label` is passed but not
+# declared: a warning, not a failure.
[email protected](queue="golang")
+def via_struct_more_args(region_code: str, unused_label: str): ...
+
+
+# The Go struct also declares `not_in_dag`, which this signature has no
+# parameter for: declared but not passed, and it keeps its Go zero value.
[email protected](queue="golang")
+def via_struct_fewer_args(region_code: str): ...
@task.stub(queue="golang")
@@ -173,16 +185,18 @@ def taskflow_binding_dag():
A Go task declares either flat data parameters, which bind *positionally*
(order matters, every one must be filled), or a single struct, whose fields
- bind by *name* like keyword arguments -- an unmatched field stays at its Go
- zero value instead of failing the task.
+ bind by *name* like keyword arguments. Name binding tolerates a signature
+ mismatch in either direction: taking more or fewer arguments than the call
+ passes is warned about rather than failing the task.
* ``via_flat_args``: every scalar literal, an array literal, keyword args,
an unpassed ``None`` default, and XComs fanned in from two upstream
tasks.
* ``via_struct_no_tags``: fields fall back to their own Go names, matched
case- and underscore-insensitively.
* ``via_struct_arg_tag``: fields bind via explicit ``arg:`` tags.
- * ``via_struct_unmatched_arg``: a Go field no argument names, and a stub
- default no Go field claims.
+ * ``via_struct_default_arg``: a stub default no Go field claims.
+ * ``via_struct_more_args`` / ``via_struct_fewer_args``: the two ways a
name-bound
+ struct and the call can disagree. Each warns and runs rather than
failing.
* ``via_flat_map`` / ``via_struct_map`` / ``via_plain_map``: one dict bound
whole into a struct, onto a struct's map field, and into a plain Go map.
@@ -202,7 +216,9 @@ def taskflow_binding_dag():
region = make_region()
via_struct_no_tags(region_code=region, threshold=0.75)
via_struct_arg_tag(region_code=region, threshold=0.75)
- via_struct_unmatched_arg(region_code=region)
+ via_struct_default_arg(region_code=region)
+ via_struct_more_args(region_code=region, unused_label="ignored")
+ via_struct_fewer_args(region_code=region)
via_flat_map(config={"region": "eu-west-1", "count": 3})
via_struct_map(payload={"region": "eu-west-1", "count": 3})
via_plain_map(labels={"team": "data", "tier": "gold"})
diff --git a/go-sdk/example/bundle/main.go b/go-sdk/example/bundle/main.go
index d9bcfcfab07..00c346d420d 100644
--- a/go-sdk/example/bundle/main.go
+++ b/go-sdk/example/bundle/main.go
@@ -61,8 +61,18 @@ func main() {
),
airflow.TaskHandler(
"taskflow_binding_dag",
- "via_struct_unmatched_arg",
- taskflowbinding.ViaStructUnmatchedArg,
+ "via_struct_default_arg",
+ taskflowbinding.ViaStructDefaultArg,
+ ),
+ airflow.TaskHandler(
+ "taskflow_binding_dag",
+ "via_struct_more_args",
+ taskflowbinding.ViaStructMoreArgs,
+ ),
+ airflow.TaskHandler(
+ "taskflow_binding_dag",
+ "via_struct_fewer_args",
+ taskflowbinding.ViaStructFewerArgs,
),
airflow.TaskHandler("taskflow_binding_dag", "via_flat_map",
taskflowbinding.ViaFlatMap),
airflow.TaskHandler("taskflow_binding_dag", "via_struct_map",
taskflowbinding.ViaStructMap),
diff --git a/go-sdk/example/bundle/taskflowbinding/taskflowbinding.go
b/go-sdk/example/bundle/taskflowbinding/taskflowbinding.go
index 0b1556fca14..3d9ed2bfa4c 100644
--- a/go-sdk/example/bundle/taskflowbinding/taskflowbinding.go
+++ b/go-sdk/example/bundle/taskflowbinding/taskflowbinding.go
@@ -171,34 +171,67 @@ func ViaStructArgTag(actx airflow.Context, input
ViaStructArgTagInput) (any, err
}, nil
}
-// ViaStructUnmatchedArgInput includes a field no argument supplies.
-type ViaStructUnmatchedArgInput struct {
- Region string `arg:"region_code"`
- Missing string `arg:"does_not_exist"`
+// ViaStructDefaultArgInput claims only the explicitly passed argument.
+type ViaStructDefaultArgInput struct {
+ Region string `arg:"region_code"`
}
-// ViaStructUnmatchedArg exercises unmatched fields and captured defaults.
-func ViaStructUnmatchedArg(
+// ViaStructDefaultArg exercises a captured stub default no struct field
claims.
+func ViaStructDefaultArg(
actx airflow.Context,
- input ViaStructUnmatchedArgInput,
+ input ViaStructDefaultArgInput,
) (any, error) {
if input.Region != "eu-west-1" {
return nil, fmt.Errorf("struct field bound incorrectly:
region=%q", input.Region)
}
- if input.Missing != "" {
+
+ actx.Logger().
+ InfoContext(actx, "Bound struct (defaulted arg unclaimed)",
"region", input.Region)
+ return map[string]any{"region": input.Region}, nil
+}
+
+// ViaStructMoreArgsInput declares fewer fields than its call passes arguments.
+type ViaStructMoreArgsInput struct {
+ Region string `arg:"region_code"`
+}
+
+// ViaStructMoreArgs exercises the Dag passing an argument the struct does not
+// declare: warned about, and the task still runs.
+func ViaStructMoreArgs(actx airflow.Context, input ViaStructMoreArgsInput)
(any, error) {
+ if input.Region != "eu-west-1" {
+ return nil, fmt.Errorf("struct field bound incorrectly:
region=%q", input.Region)
+ }
+
+ actx.Logger().InfoContext(actx, "Bound struct (call passed more)",
"region", input.Region)
+ return map[string]any{"region": input.Region}, nil
+}
+
+// ViaStructFewerArgsInput declares more fields than its call passes arguments.
+type ViaStructFewerArgsInput struct {
+ Region string `arg:"region_code"`
+ NotInDag string `arg:"not_in_dag"`
+}
+
+// ViaStructFewerArgs exercises the struct declaring an argument the Dag's call
+// does not pass: warned about, and the field keeps its Go zero value.
+func ViaStructFewerArgs(actx airflow.Context, input ViaStructFewerArgsInput)
(any, error) {
+ if input.Region != "eu-west-1" {
+ return nil, fmt.Errorf("struct field bound incorrectly:
region=%q", input.Region)
+ }
+ if input.NotInDag != "" {
return nil, fmt.Errorf(
- "expected the unmatched field to stay at its Go zero
value, got missing=%q",
- input.Missing,
+ "expected the undeclared field to keep its zero value,
got not_in_dag=%q",
+ input.NotInDag,
)
}
- actx.Logger().InfoContext(actx, "Bound struct (unmatched arg)",
+ actx.Logger().InfoContext(actx, "Bound struct (call passed fewer)",
"region", input.Region,
- "missing_was_empty", input.Missing == "",
+ "not_in_dag_was_empty", input.NotInDag == "",
)
return map[string]any{
- "region": input.Region,
- "missing_was_empty": input.Missing == "",
+ "region": input.Region,
+ "not_in_dag_was_empty": input.NotInDag == "",
}, nil
}
diff --git a/go-sdk/example/bundle/taskflowbinding/taskflowbinding_test.go
b/go-sdk/example/bundle/taskflowbinding/taskflowbinding_test.go
index beb046aa29a..4bf3a4bab3d 100644
--- a/go-sdk/example/bundle/taskflowbinding/taskflowbinding_test.go
+++ b/go-sdk/example/bundle/taskflowbinding/taskflowbinding_test.go
@@ -107,24 +107,44 @@ func TestViaStructArgTagRejectsWrongBinding(t *testing.T)
{
assert.ErrorContains(t, err, "struct fields bound incorrectly")
}
-func TestViaStructUnmatchedArg(t *testing.T) {
- got, err := ViaStructUnmatchedArg(testContext(t),
ViaStructUnmatchedArgInput{
- Region: "eu-west-1",
- Missing: "",
- })
+func TestViaStructDefaultArg(t *testing.T) {
+ got, err := ViaStructDefaultArg(testContext(t),
ViaStructDefaultArgInput{Region: "eu-west-1"})
+ require.NoError(t, err)
+
+ summary, ok := got.(map[string]any)
+ require.True(t, ok, "ViaStructDefaultArg should return a map summary,
got %T", got)
+ assert.Equal(t, "eu-west-1", summary["region"])
+}
+
+func TestViaStructDefaultArgRejectsWrongBinding(t *testing.T) {
+ _, err := ViaStructDefaultArg(testContext(t),
ViaStructDefaultArgInput{Region: "wrong-region"})
+ assert.ErrorContains(t, err, "struct field bound incorrectly")
+}
+
+func TestViaStructMoreArgs(t *testing.T) {
+ got, err := ViaStructMoreArgs(testContext(t),
ViaStructMoreArgsInput{Region: "eu-west-1"})
+ require.NoError(t, err)
+
+ summary, ok := got.(map[string]any)
+ require.True(t, ok, "ViaStructMoreArgs should return a map summary, got
%T", got)
+ assert.Equal(t, "eu-west-1", summary["region"])
+}
+
+func TestViaStructFewerArgs(t *testing.T) {
+ got, err := ViaStructFewerArgs(testContext(t),
ViaStructFewerArgsInput{Region: "eu-west-1"})
require.NoError(t, err)
summary, ok := got.(map[string]any)
- require.True(t, ok, "ViaStructUnmatchedArg should return a map summary,
got %T", got)
- assert.Equal(t, true, summary["missing_was_empty"])
+ require.True(t, ok, "ViaStructFewerArgs should return a map summary,
got %T", got)
+ assert.Equal(t, true, summary["not_in_dag_was_empty"])
}
-func TestViaStructUnmatchedArgRejectsNonZeroMissingField(t *testing.T) {
- _, err := ViaStructUnmatchedArg(testContext(t),
ViaStructUnmatchedArgInput{
- Region: "eu-west-1",
- Missing: "unexpected",
+func TestViaStructFewerArgsRejectsFilledField(t *testing.T) {
+ _, err := ViaStructFewerArgs(testContext(t), ViaStructFewerArgsInput{
+ Region: "eu-west-1",
+ NotInDag: "unexpected",
})
- assert.ErrorContains(t, err, "expected the unmatched field to stay at
its Go zero value")
+ assert.ErrorContains(t, err, "expected the undeclared field to keep its
zero value")
}
func TestViaFlatMap(t *testing.T) {
diff --git a/go-sdk/pkg/binding/binding.go b/go-sdk/pkg/binding/binding.go
index cb35a35f594..bcc3752ec9a 100644
--- a/go-sdk/pkg/binding/binding.go
+++ b/go-sdk/pkg/binding/binding.go
@@ -164,7 +164,7 @@ func (p *Plan) Resolve(
ti, dagRun := storedRunMetadata(ctx)
out[0] = newAirflowContext(ctx, logger, client, ti, dagRun)
if p.loneStruct {
- return p.resolveLoneStructParam(ctx, client, args, out)
+ return p.resolveLoneStructParam(ctx, logger, client, args, out)
}
return p.resolveFlatParams(ctx, client, args, out)
}
@@ -219,6 +219,7 @@ func (p *Plan) resolveFlatParams(
func (p *Plan) resolveLoneStructParam(
ctx context.Context,
+ logger *slog.Logger,
c sdk.XComClient,
args []Arg,
out []reflect.Value,
@@ -255,6 +256,7 @@ func (p *Plan) resolveLoneStructParam(
argIdx int
}
binds := make([]fieldBind, 0, len(plan.fields))
+ var unfilled []string
for _, sf := range plan.fields {
idx, ok := byName[sf.argName]
if !ok && !sf.tagged {
@@ -264,6 +266,7 @@ func (p *Plan) resolveLoneStructParam(
}
}
if !ok {
+ unfilled = append(unfilled, fmt.Sprintf("%s (argument
%q)", sf.goName, sf.argName))
continue
}
claimed[idx] = true
@@ -277,14 +280,25 @@ func (p *Plan) resolveLoneStructParam(
}
}
- if len(args) == 0 && len(plan.fields) > 0 {
- return nil, fmt.Errorf(
- "task function %s: no TaskFlow arg bindings arrived but
the struct declares "+
- "%d bindable field(s); nothing can fill them on
this execution path",
- p.fnName, len(plan.fields),
+ // Neither direction of a name mismatch is fatal, because a struct
binds by
+ // name: an unfilled field keeps its Go zero value and an unclaimed
argument
+ // changes nothing the handler reads. Both are logged so the mismatch
is still
+ // visible, since the spec carries one entry per stub parameter and
either side
+ // of it means the Go signature and the stub signature disagree.
+ //
+ // A spec that arrived empty is the same thing with every field
unfilled, and
+ // is what an argless call looks like: build_arg_bindings sends nothing
at all
+ // when a stub is called with no arguments.
+ if len(unfilled) > 0 {
+ logger.Warn(
+ "Task handler declares argument(s) the Dag's call did
not pass",
+ "function", p.fnName,
+ "declared_not_passed", unfilled,
+ "passed", passedArgNames(args),
)
}
+ // Captured defaults are the normal case of an unclaimed argument and
stay silent.
var unclaimed []string
for i, c := range claimed {
if c {
@@ -295,15 +309,16 @@ func (p *Plan) resolveLoneStructParam(
}
name := "<nil>"
if args[i] != nil {
- name = fmt.Sprintf("%q", args[i].ArgName())
+ name = args[i].ArgName()
}
unclaimed = append(unclaimed, name)
}
if len(unclaimed) > 0 {
- return nil, fmt.Errorf(
- "task function %s: %d TaskFlow call argument(s) not
claimed by any struct "+
- "field: %s",
- p.fnName, len(unclaimed), strings.Join(unclaimed, ", "),
+ logger.Warn(
+ "Dag's call passed argument(s) the task handler does
not declare",
+ "function", p.fnName,
+ "passed_not_declared", unclaimed,
+ "declared", declaredFieldNames(plan.fields),
)
}
@@ -337,6 +352,25 @@ func (p *Plan) resolveLoneStructParam(
return out, nil
}
+func declaredFieldNames(fields []structField) []string {
+ names := make([]string, 0, len(fields))
+ for _, sf := range fields {
+ names = append(names, sf.argName)
+ }
+ return names
+}
+
+func passedArgNames(args []Arg) []string {
+ names := make([]string, 0, len(args))
+ for _, a := range args {
+ if a == nil {
+ continue
+ }
+ names = append(names, a.ArgName())
+ }
+ return names
+}
+
func dropDefaultedArgs(args []Arg) []Arg {
kept := make([]Arg, 0, len(args))
for _, a := range args {
diff --git a/go-sdk/pkg/binding/binding_test.go
b/go-sdk/pkg/binding/binding_test.go
index 7f803b19d6b..704ad9a90d9 100644
--- a/go-sdk/pkg/binding/binding_test.go
+++ b/go-sdk/pkg/binding/binding_test.go
@@ -18,6 +18,7 @@
package binding
import (
+ "bytes"
"context"
"io"
"log/slog"
@@ -108,14 +109,31 @@ func analyze(s *BindingSuite, fn any) *Plan {
}
func (s *BindingSuite) resolve(fn any, args []Arg, client sdk.Client)
([]reflect.Value, error) {
+ values, _, err := s.resolveWithLogs(fn, args, client)
+ return values, err
+}
+
+// resolveWithLogs is resolve with the logger's output captured, for the checks
+// that warn rather than fail.
+func (s *BindingSuite) resolveWithLogs(
+ fn any,
+ args []Arg,
+ client sdk.Client,
+) ([]reflect.Value, string, error) {
+ var logs bytes.Buffer
plan := analyze(s, fn)
- values, err := plan.Resolve(runtimeCtx(), slog.Default(), client, args)
+ values, err := plan.Resolve(
+ runtimeCtx(),
+ slog.New(slog.NewTextHandler(&logs, nil)),
+ client,
+ args,
+ )
if err != nil {
- return nil, err
+ return nil, logs.String(), err
}
s.Require().True(values[0].IsValid(), "parameter 0 must be bound")
s.Require().IsType(contexttest.Context{}, values[0].Interface())
- return values[1:], nil
+ return values[1:], logs.String(), nil
}
func (s *BindingSuite) TestAnalyzeClassification() {
@@ -865,28 +883,35 @@ type taggedRegionInput struct {
func (s *BindingSuite) TestResolveTaggedStructNeverFallsBackToWholeValue() {
fn := func(actx contexttest.Context, input taggedRegionInput) error {
return nil }
- _, err := s.resolve(fn, []Arg{
+ got, logs, err := s.resolveWithLogs(fn, []Arg{
LiteralArg{Name: "region_code", Value: "eu-west-1",
ValueSchema: argSchema("string")},
}, &fakeXComClient{})
- if s.Assert().Error(err) {
- s.Contains(err.Error(), `not claimed by any struct field:
"region_code"`)
- }
+ s.Require().NoError(err)
+ s.Equal(
+ "",
+ got[0].Interface().(taggedRegionInput).Region,
+ "no whole-value fallback for a tagged struct",
+ )
+ // A typo'd tag shows up from both sides at once, which is what names
it.
+ s.Contains(logs, `Region (argument \"regon_code\")`)
+ s.Contains(logs, "the task handler does not declare")
}
-func (s *BindingSuite) TestResolveStructUnclaimedArgFailsLoudly() {
- fn := func(actx contexttest.Context, input combineInput) error { return
nil }
- _, err := s.resolve(fn, []Arg{
+func (s *BindingSuite) TestResolveStructUnclaimedArgWarns() {
+ fn := func(actx contexttest.Context, input simpleInput) error { return
nil }
+ got, logs, err := s.resolveWithLogs(fn, []Arg{
LiteralArg{Name: "Name", Value: "widget", ValueSchema:
argSchema("string")},
LiteralArg{Name: "typo", Value: "x", ValueSchema:
argSchema("string")},
}, &fakeXComClient{})
- if s.Assert().Error(err) {
- s.Contains(err.Error(), `not claimed by any struct field:
"typo"`)
- }
+ s.Require().NoError(err, "an argument no field claims changes nothing
the handler reads")
+ s.Equal("widget", got[0].Interface().(simpleInput).Name)
+ s.Contains(logs, "the task handler does not declare")
+ s.Contains(logs, "typo")
}
-func (s *BindingSuite) TestResolveStructUnclaimedFromDefaultAllowed() {
- fn := func(actx contexttest.Context, input combineInput) error { return
nil }
- got, err := s.resolve(fn, []Arg{
+func (s *BindingSuite) TestResolveStructUnclaimedFromDefaultStaysSilent() {
+ fn := func(actx contexttest.Context, input simpleInput) error { return
nil }
+ got, logs, err := s.resolveWithLogs(fn, []Arg{
LiteralArg{Name: "Name", Value: "widget", ValueSchema:
argSchema("string")},
LiteralArg{
Name: "threshold",
@@ -896,24 +921,28 @@ func (s *BindingSuite)
TestResolveStructUnclaimedFromDefaultAllowed() {
},
}, &fakeXComClient{})
s.Require().NoError(err)
- s.Equal("widget", got[0].Interface().(combineInput).Name)
+ s.Equal("widget", got[0].Interface().(simpleInput).Name)
+ s.NotContains(logs, "the task handler does not declare")
}
-func (s *BindingSuite) TestResolveStructEmptySpecFailsLoudly() {
+func (s *BindingSuite) TestResolveStructEmptySpecWarns() {
+ // An argless call sends no spec at all, so this is the ordinary shape
of a
+ // stub called as `my_task()`, not a sign of an Airflow that cannot
send one.
fn := func(actx contexttest.Context, input simpleInput) error { return
nil }
for name, args := range map[string][]Arg{"nil-spec": nil, "empty-spec":
{}} {
s.Run(name, func() {
- _, err := s.resolve(fn, args, &fakeXComClient{})
- if s.Assert().Error(err) {
- s.Contains(err.Error(), "no TaskFlow arg
bindings arrived")
- }
+ got, logs, err := s.resolveWithLogs(fn, args,
&fakeXComClient{})
+ s.Require().NoError(err)
+ s.Equal("", got[0].Interface().(simpleInput).Name,
"every field keeps its zero value")
+ s.Contains(logs, "the Dag's call did not pass")
+ s.Contains(logs, "passed=[]")
})
}
}
func (s *BindingSuite) TestResolveStructOnlyDefaultsZeroValues() {
fn := func(actx contexttest.Context, input twoFieldInput) error {
return nil }
- got, err := s.resolve(fn, []Arg{
+ got, logs, err := s.resolveWithLogs(fn, []Arg{
LiteralArg{
Name: "threshold",
Value: 0.75,
@@ -925,17 +954,37 @@ func (s *BindingSuite)
TestResolveStructOnlyDefaultsZeroValues() {
input := got[0].Interface().(twoFieldInput)
s.Equal("", input.Name, "no explicit entry arrived; fields keep
kwarg-style zero values")
s.Equal("", input.Missing)
+ s.Contains(logs, "the Dag's call did not pass")
}
-func (s *BindingSuite) TestResolveStructUnmatchedFieldZeroValued() {
+func (s *BindingSuite) TestResolveStructUnmatchedFieldWarns() {
fn := func(actx contexttest.Context, input twoFieldInput) error {
return nil }
- got, err := s.resolve(fn, []Arg{
+ got, logs, err := s.resolveWithLogs(fn, []Arg{
LiteralArg{Name: "Name", Value: "widget", ValueSchema:
argSchema("string")},
}, &fakeXComClient{})
- s.Require().NoError(err)
+ s.Require().NoError(err, "a struct binds by name, so an unfilled field
is not fatal")
input := got[0].Interface().(twoFieldInput)
s.Equal("widget", input.Name, "the matched field binds normally")
- s.Equal("", input.Missing, "the unmatched field is left at its Go zero
value, not an error")
+ s.Equal("", input.Missing, "the unmatched field keeps its Go zero
value")
+ s.Contains(logs, "the Dag's call did not pass")
+ s.Contains(logs, `Missing (argument \"missing\")`)
+ s.NotContains(logs, "the task handler does not declare", "every
argument was claimed")
+}
+
+func (s *BindingSuite) TestResolveStructWarnsInEachDirectionAtOnce() {
+ // A call can be wrong both ways at the same time, so each direction is
its
+ // own message rather than one line a reader has to untangle.
+ fn := func(actx contexttest.Context, input twoFieldInput) error {
return nil }
+ got, logs, err := s.resolveWithLogs(fn, []Arg{
+ LiteralArg{Name: "Name", Value: "widget", ValueSchema:
argSchema("string")},
+ LiteralArg{Name: "typo", Value: "x", ValueSchema:
argSchema("string")},
+ }, &fakeXComClient{})
+ s.Require().NoError(err)
+ s.Equal("widget", got[0].Interface().(twoFieldInput).Name)
+ s.Contains(logs, "the Dag's call did not pass")
+ s.Contains(logs, `declared_not_passed="[Missing (argument
\"missing\")]"`)
+ s.Contains(logs, "the task handler does not declare")
+ s.Contains(logs, "passed_not_declared=[typo]")
}
func (s *BindingSuite) TestResolveFlatParamsToleratesCapturedDefaults() {
