You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "jason810496 (via GitHub)" <gi...@apache.org> on 2026/09/23 07:21:10 UTC
[PR] Java SDK: Resolve a task's arguments from the Dag's own wiring [airflow]
jason810496 opened a new pull request, #73596:
URL: https://github.com/apache/airflow/pull/73596
- depends on: #CONFIG
- Diff for early review: https://github.com/jason810496/airflow/compare/feature/java-sdk-dag-schema-config...feature/java-sdk-wired-inputs
> **Merge order:**
> 1. #71188 — Honor TaskFlow arg bindings sent by the supervisor
> 2. #EDGES — Draw a Java-authored Dag's edges with then and dependsOn
> 3. #CONFIG — Generate Dag and task configuration from the serialization schema
> 4. **#WIRED — Resolve a task's arguments from the Dag's own wiring** **(current one)**
> 5. #71189 — Declare a Dag's task graph with @Builder.Deps
> 6. #71190 — Serialize native Dags to DagSerialization v3
>
> Every PR targets `main` because GitHub cannot base a pull request on a branch that exists only on a fork, so these diffs are cumulative — the compare link above shows only this layer. #69757 and #71057, which the stack was originally built on, are merged.
## Why
A Dag authored in Java has no `@task.stub` call site, so the supervisor sends no argument bindings for it and a task's data parameters had nothing to resolve against.
## How
- Where the supervisor sent no bindings, a task's flat parameters and its `TaskInput` resolve against the inputs the Dag recorded for that task: an upstream handle reads that task's XCom, a literal is passed straight through.
- Runtime bindings win wherever both exist, so a stub-backed task behaves exactly as before.
- `TaskArgs.of` checks the declared parameter count against the supplied one, and names which side supplied what when they disagree.
- A value that resolves to nothing still fails the same way: `null` for a boxed or reference parameter, `MissingXComException` naming the task for a primitive.
This is the runtime half on its own. Nothing declares wired inputs until the next PR adds `@Builder.Deps`, so there is no user-visible change yet and the tests wire inputs directly.
---
##### Was generative AI tooling used to co-author this PR?
- [x] Yes, with help of Claude Code Opus 5 following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Java SDK: Resolve a task's arguments from the Dag's own wiring [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #73596:
URL: https://github.com/apache/airflow/pull/73596#discussion_r4164663749
##########
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/ArgValues.kt:
##########
@@ -147,6 +173,104 @@ object ArgValues {
type: Type,
): Any? = decode(client.resolveBinding(binding), type)
+ /**
+ * Reports a Dag that wired more or fewer inputs than a [TaskInput] can take.
+ * A [TaskInput] is a task's only data parameter, so exactly one input feeds
+ * it; a mismatch warns and the first input is used, leaving the task to run
+ * on what it can bind rather than failing the run outright.
+ */
+ private fun warnWiredArity(
+ client: Client,
+ type: Class<*>,
+ wired: Int,
+ ) {
+ if (wired == 1) return
+ logger.warning(
+ "Dag wired a different number of argument(s) than the task handler declares",
Review Comment:
Thanks, addressed in https://github.com/apache/airflow/pull/73596/commits/b1a2e15ebdb4f88c1efb9a18056ddfdfb16347dc.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Java SDK: Resolve a task's arguments from the Dag's own wiring [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #73596:
URL: https://github.com/apache/airflow/pull/73596#discussion_r4153177058
##########
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Context.kt:
##########
@@ -123,9 +123,14 @@ data class TaskInstance(
data class Context(
@JvmField val dagRun: DagRun,
@JvmField val ti: TaskInstance,
+ /** Registration of the executing task; resolves wired data inputs. */
+ internal val taskDef: TaskDef? = null,
Review Comment:
Thanks, moved `taskDef` out of the constructor as suggested, so the public constructor, `equals` and `toString` stay as before. Fixed in 24bffde50d.
##########
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/ArgValues.kt:
##########
@@ -147,6 +173,113 @@ object ArgValues {
type: Type,
): Any? = decode(client.resolveBinding(binding), type)
+ /**
+ * Reports a Dag that wired more or fewer inputs than a [TaskInput] can take.
+ * A [TaskInput] is a task's only data parameter, so exactly one input feeds
+ * it; a mismatch warns and the first input is used, leaving the task to run
+ * on what it can bind rather than failing the run outright.
+ */
+ private fun warnWiredArity(
+ client: Client,
+ type: Class<*>,
+ wired: Int,
+ ) {
+ if (wired == 1) return
+ logger.warning(
+ "Dag wired a different number of argument(s) than the task handler declares",
+ mapOf(
+ "task_id" to client.details.ti.taskId,
+ "input" to type.simpleName,
+ "declared" to 1,
+ "wired" to wired,
+ ),
+ )
+ }
+
+ /**
+ * Resolves every wired input at once: each upstream is read once however
+ * many parameters it feeds, and several upstreams are read concurrently.
+ * The supervisor protocol matches responses to requests by id, so a task
+ * wired to several upstreams waits roughly one round trip rather than one
+ * per parameter.
+ */
+ internal fun resolveWiredAll(
+ inputs: List<Arg<*>>,
+ client: Client,
+ ): List<Any?> {
+ val upstreams = inputs.filterIsInstance<TaskRef<*>>().map { it.def.id }.distinct()
+ val fetched =
+ when (upstreams.size) {
+ 0 -> emptyMap()
+ 1 -> mapOf(upstreams[0] to client.getXCom(taskId = upstreams[0]))
+ else ->
+ runBlocking {
+ upstreams.map { taskId -> async(Dispatchers.IO) { taskId to client.getXCom(taskId = taskId) } }.awaitAll()
+ }.toMap()
+ }
+ return inputs.map { input ->
+ when (input) {
+ is TaskRef<*> -> fetched[input.def.id]
+ is LiteralArg<*> -> input.value
+ }
+ }
+ }
+
+ /** Decodes an already-resolved wired value into [type], passing null through. */
+ internal fun decodeWired(
+ value: Any?,
+ type: Type,
+ ): Any? = decode(value, type)
+
+ /**
+ * The inputs the Dag wired for this task, or null when the run's arguments
+ * come from the stub call site. A task with no wired inputs reads the
+ * bindings, so a stub call that bound nothing keeps its own diagnostics.
+ */
+ internal fun wiredInputs(
+ context: Context,
+ client: Client,
+ ): List<Arg<*>>? = if (client.argBindings.isEmpty()) context.taskDef?.inputs?.takeIf { it.isNotEmpty() } else null
+
+ /**
+ * The failure for a wired argument that resolved to nothing where a value is
+ * required, naming [target] — the position of the parameter it feeds.
+ */
+ internal fun missingWired(
+ input: Arg<*>,
+ target: String,
+ ): MissingXComException =
+ when (input) {
+ is TaskRef<*> -> MissingXComException(input.def.id, target)
+ is LiteralArg<*> ->
+ MissingXComException(
+ "Task parameter '$target' is wired to a null literal, but has a primitive type that cannot " +
+ "be null; declare a boxed type (e.g. Integer instead of int) to receive null.",
+ )
+ }
+
+ /** The failure for a wired input that resolved to nothing for a [TaskInput]. */
+ private fun missingInput(
+ input: Arg<*>,
+ target: String,
+ ): MissingXComException =
+ when (input) {
+ is TaskRef<*> ->
+ MissingXComException(
+ "Input '$target' requires an XCom from task '${input.def.id}', but none was pushed.",
+ )
+ is LiteralArg<*> -> MissingXComException("Input '$target' is wired to a null literal, so there is nothing to bind.")
+ }
+
+ private fun resolveWired(
Review Comment:
Good catch, removed `resolveWired` and reused `resolveWiredAll`. Fixed in 24bffde50d.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Java SDK: Resolve a task's arguments from the Dag's own wiring [airflow]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #73596:
URL: https://github.com/apache/airflow/pull/73596#discussion_r4151225262
##########
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Context.kt:
##########
@@ -123,9 +123,14 @@ data class TaskInstance(
data class Context(
@JvmField val dagRun: DagRun,
@JvmField val ti: TaskInstance,
+ /** Registration of the executing task; resolves wired data inputs. */
+ internal val taskDef: TaskDef? = null,
Review Comment:
Agree with this comment. Adding taskDef to the constructor also breaks Java callers. We can change it like following:
```diff
--- a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Context.kt
+++ b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Context.kt
data class Context(
@JvmField val dagRun: DagRun,
@JvmField val ti: TaskInstance,
- /** Registration of the executing task; resolves wired data inputs. */
- internal val taskDef: TaskDef? = null,
) {
+ /** Registration of the executing task; resolves wired data inputs. */
+ internal var taskDef: TaskDef? = null
+
internal companion object {
fun from(
request: StartupDetails,
taskDef: TaskDef? = null,
): Context =
Context(
dagRun = ...,
ti = with(request.ti) { TaskInstance(dagId, runId, taskId, mapIndex, tryNumber) },
- taskDef = taskDef,
- )
+ ).also { it.taskDef = taskDef }
}
}
```
```diff
--- a/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/ArgTestSupport.kt
- return taskContext().copy(taskDef = def)
+ return taskContext().also { it.taskDef = def }
--- a/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/internal/ArgValuesTest.kt
- return contextWithoutTaskDef().copy(taskDef = def)
+ return contextWithoutTaskDef().also { it.taskDef = def }
```
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Java SDK: Resolve a task's arguments from the Dag's own wiring [airflow]
Posted by "henry3260 (via GitHub)" <gi...@apache.org>.
henry3260 commented on code in PR #73596:
URL: https://github.com/apache/airflow/pull/73596#discussion_r4163857477
##########
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/ArgValues.kt:
##########
@@ -147,6 +173,104 @@ object ArgValues {
type: Type,
): Any? = decode(client.resolveBinding(binding), type)
+ /**
+ * Reports a Dag that wired more or fewer inputs than a [TaskInput] can take.
+ * A [TaskInput] is a task's only data parameter, so exactly one input feeds
+ * it; a mismatch warns and the first input is used, leaving the task to run
+ * on what it can bind rather than failing the run outright.
+ */
+ private fun warnWiredArity(
+ client: Client,
+ type: Class<*>,
+ wired: Int,
+ ) {
+ if (wired == 1) return
+ logger.warning(
+ "Dag wired a different number of argument(s) than the task handler declares",
Review Comment:
nit: Lines 130/153 already use the two wordings ADR-0007 asks every SDK to share;
this one invents a third, so a cross-language grep misses the native-Dag case.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Java SDK: Resolve a task's arguments from the Dag's own wiring [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 merged PR #73596:
URL: https://github.com/apache/airflow/pull/73596
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Java SDK: Resolve a task's arguments from the Dag's own wiring [airflow]
Posted by "guan404ming (via GitHub)" <gi...@apache.org>.
guan404ming commented on code in PR #73596:
URL: https://github.com/apache/airflow/pull/73596#discussion_r4144907628
##########
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/ArgValues.kt:
##########
@@ -147,6 +173,113 @@ object ArgValues {
type: Type,
): Any? = decode(client.resolveBinding(binding), type)
+ /**
+ * Reports a Dag that wired more or fewer inputs than a [TaskInput] can take.
+ * A [TaskInput] is a task's only data parameter, so exactly one input feeds
+ * it; a mismatch warns and the first input is used, leaving the task to run
+ * on what it can bind rather than failing the run outright.
+ */
+ private fun warnWiredArity(
+ client: Client,
+ type: Class<*>,
+ wired: Int,
+ ) {
+ if (wired == 1) return
+ logger.warning(
+ "Dag wired a different number of argument(s) than the task handler declares",
+ mapOf(
+ "task_id" to client.details.ti.taskId,
+ "input" to type.simpleName,
+ "declared" to 1,
+ "wired" to wired,
+ ),
+ )
+ }
+
+ /**
+ * Resolves every wired input at once: each upstream is read once however
+ * many parameters it feeds, and several upstreams are read concurrently.
+ * The supervisor protocol matches responses to requests by id, so a task
+ * wired to several upstreams waits roughly one round trip rather than one
+ * per parameter.
+ */
+ internal fun resolveWiredAll(
+ inputs: List<Arg<*>>,
+ client: Client,
+ ): List<Any?> {
+ val upstreams = inputs.filterIsInstance<TaskRef<*>>().map { it.def.id }.distinct()
+ val fetched =
+ when (upstreams.size) {
+ 0 -> emptyMap()
+ 1 -> mapOf(upstreams[0] to client.getXCom(taskId = upstreams[0]))
+ else ->
+ runBlocking {
+ upstreams.map { taskId -> async(Dispatchers.IO) { taskId to client.getXCom(taskId = taskId) } }.awaitAll()
+ }.toMap()
+ }
+ return inputs.map { input ->
+ when (input) {
+ is TaskRef<*> -> fetched[input.def.id]
+ is LiteralArg<*> -> input.value
+ }
+ }
+ }
+
+ /** Decodes an already-resolved wired value into [type], passing null through. */
+ internal fun decodeWired(
+ value: Any?,
+ type: Type,
+ ): Any? = decode(value, type)
+
+ /**
+ * The inputs the Dag wired for this task, or null when the run's arguments
+ * come from the stub call site. A task with no wired inputs reads the
+ * bindings, so a stub call that bound nothing keeps its own diagnostics.
+ */
+ internal fun wiredInputs(
+ context: Context,
+ client: Client,
+ ): List<Arg<*>>? = if (client.argBindings.isEmpty()) context.taskDef?.inputs?.takeIf { it.isNotEmpty() } else null
+
+ /**
+ * The failure for a wired argument that resolved to nothing where a value is
+ * required, naming [target] — the position of the parameter it feeds.
+ */
+ internal fun missingWired(
+ input: Arg<*>,
+ target: String,
+ ): MissingXComException =
+ when (input) {
+ is TaskRef<*> -> MissingXComException(input.def.id, target)
+ is LiteralArg<*> ->
+ MissingXComException(
+ "Task parameter '$target' is wired to a null literal, but has a primitive type that cannot " +
+ "be null; declare a boxed type (e.g. Integer instead of int) to receive null.",
+ )
+ }
+
+ /** The failure for a wired input that resolved to nothing for a [TaskInput]. */
+ private fun missingInput(
+ input: Arg<*>,
+ target: String,
+ ): MissingXComException =
+ when (input) {
+ is TaskRef<*> ->
+ MissingXComException(
+ "Input '$target' requires an XCom from task '${input.def.id}', but none was pushed.",
+ )
+ is LiteralArg<*> -> MissingXComException("Input '$target' is wired to a null literal, so there is nothing to bind.")
+ }
+
+ private fun resolveWired(
Review Comment:
`resolveWired` matches `resolveWiredAll` for a single input, so we could reuse that one instead.
##########
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Context.kt:
##########
@@ -123,9 +123,14 @@ data class TaskInstance(
data class Context(
@JvmField val dagRun: DagRun,
@JvmField val ti: TaskInstance,
+ /** Registration of the executing task; resolves wired data inputs. */
+ internal val taskDef: TaskDef? = null,
Review Comment:
Adding `taskDef` here changes the public data class constructor, `equals` and `toString`. Keeping it outside would avoid that.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org