This is an automated email from the ASF dual-hosted git repository.
jason810496 pushed a commit to branch v3-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-4-test by this push:
new fea1a47ddff [v3-4-test] Java SDK: Reject IDs that collide with task
group join nodes (#74781) (#75199)
fea1a47ddff is described below
commit fea1a47ddff8c4396ec25cca2701502d077ba287
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Tue Oct 20 18:23:56 2026 +0800
[v3-4-test] Java SDK: Reject IDs that collide with task group join nodes
(#74781) (#75199)
(cherry picked from commit 1c4af0388744cd321ce618139aad0092112ab3db)
Signed-off-by: PoAn Yang <[email protected]>
Co-authored-by: PoAn Yang <[email protected]>
Co-authored-by: Rahul Vats <[email protected]>
---
.../org/apache/airflow/sdk/AirflowProcessor.kt | 15 ++-
.../org/apache/airflow/sdk/AirflowProcessorTest.kt | 123 +++++++++++++++++++++
.../main/kotlin/org/apache/airflow/sdk/DagDef.kt | 38 ++++++-
.../kotlin/org/apache/airflow/sdk/TaskGroupRef.kt | 13 ++-
.../kotlin/org/apache/airflow/sdk/internal/Ids.kt | 13 +++
.../kotlin/org/apache/airflow/sdk/TaskGroupTest.kt | 98 ++++++++++++++++
6 files changed, 294 insertions(+), 6 deletions(-)
diff --git
a/java-sdk/processor/src/main/kotlin/org/apache/airflow/sdk/AirflowProcessor.kt
b/java-sdk/processor/src/main/kotlin/org/apache/airflow/sdk/AirflowProcessor.kt
index 666ca2e23e2..36d2f1cfdd6 100644
---
a/java-sdk/processor/src/main/kotlin/org/apache/airflow/sdk/AirflowProcessor.kt
+++
b/java-sdk/processor/src/main/kotlin/org/apache/airflow/sdk/AirflowProcessor.kt
@@ -38,6 +38,7 @@ import org.apache.airflow.sdk.internal.Refs
import org.apache.airflow.sdk.internal.SchemaFields
import org.apache.airflow.sdk.internal.TaskArgs
import org.apache.airflow.sdk.internal.TypeRef
+import org.apache.airflow.sdk.internal.deriveJoinNodeIds
import org.apache.airflow.sdk.internal.foldArgName
import org.apache.airflow.sdk.internal.registrarName
import java.time.Duration
@@ -584,6 +585,8 @@ class AirflowProcessor : AbstractProcessor() {
/**
* Rejects two tasks of the Dag sharing an ID, a task and a task group
* sharing one, and two task methods whose generated classes would collide.
+ * Also rejects a task or task group that takes the ID of a group's join
+ * node.
*/
private fun checkIds(scope: Scope) {
val declarations = scope.allTasks()
@@ -591,10 +594,20 @@ class AirflowProcessor : AbstractProcessor() {
declarations.forEach { decl ->
require(taskIds.add(decl.id)) { "Tasks in Dag have duplicate ID:
${decl.id}" }
}
- scope.allGroups().forEach { group ->
+ val groups = scope.allGroups()
+ val groupIds = groups.mapTo(mutableSetOf()) { it.fullId }
+ groups.forEach { group ->
require(group.fullId !in taskIds) {
"Dag has both a task and a task group with ID '${group.fullId}';
rename one"
}
+ deriveJoinNodeIds(group.fullId).forEach { joinId ->
+ require(joinId !in taskIds) {
+ "Task '$joinId' has the ID of a join node of task group
'${group.fullId}'; rename the task"
+ }
+ require(joinId !in groupIds) {
+ "Task group '$joinId' has the ID of a join node of task group
'${group.fullId}'; rename the nested group"
+ }
+ }
}
val byClassName = mutableMapOf<String, TaskDeclaration>()
// A task that triggers a Dag run has no generated class to collide with.
diff --git
a/java-sdk/processor/src/test/kotlin/org/apache/airflow/sdk/AirflowProcessorTest.kt
b/java-sdk/processor/src/test/kotlin/org/apache/airflow/sdk/AirflowProcessorTest.kt
index c42d52413bd..11be0f957f1 100644
---
a/java-sdk/processor/src/test/kotlin/org/apache/airflow/sdk/AirflowProcessorTest.kt
+++
b/java-sdk/processor/src/test/kotlin/org/apache/airflow/sdk/AirflowProcessorTest.kt
@@ -1777,6 +1777,129 @@ class AirflowProcessorTest {
)
}
+ @Test
+ @DisplayName("reject a task in a task group that takes the id of one of the
group's join nodes")
+ fun rejectTaskTakingJoinNodeIdOfItsGroup() {
+ val compilation =
+ compile(
+ """
+ package org.apache.airflow.example;
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag
+ public class TestExample {
+ @Airflow.TaskGroup
+ static class Transform {
+ @Airflow.Task(id = "upstream_join_id") public void clean() {}
+ }
+
+ @Airflow.Deps
+ static class Wiring implements TestExampleDeps {
+ void depends() {}
+ }
+ }
+ """,
+ )
+ assertThat(compilation).failed()
+ assertThat(compilation).hadErrorContaining(
+ "Task 'Transform.upstream_join_id' has the ID of a join node of task
group 'Transform'; rename the task",
+ )
+ }
+
+ @Test
+ @DisplayName("reject a top-level task that takes the id of a task group's
join node")
+ fun rejectTopLevelTaskTakingJoinNodeId() {
+ val compilation =
+ compile(
+ """
+ package org.apache.airflow.example;
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag
+ public class TestExample {
+ @Airflow.Task(id = "Transform.downstream_join_id") public void
publish() {}
+
+ @Airflow.TaskGroup
+ static class Transform {
+ @Airflow.Task public void clean() {}
+ }
+
+ @Airflow.Deps
+ static class Wiring implements TestExampleDeps {
+ void depends() {}
+ }
+ }
+ """,
+ )
+ assertThat(compilation).failed()
+ assertThat(compilation).hadErrorContaining(
+ "Task 'Transform.downstream_join_id' has the ID of a join node of task
group 'Transform'; rename the task",
+ )
+ }
+
+ @Test
+ @DisplayName("reject a task in a nested task group that takes the id of one
of that group's join nodes")
+ fun rejectTaskTakingJoinNodeIdOfNestedGroup() {
+ val compilation =
+ compile(
+ """
+ package org.apache.airflow.example;
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag
+ public class TestExample {
+ @Airflow.TaskGroup
+ static class Transform {
+ @Airflow.TaskGroup
+ static class Checks {
+ @Airflow.Task(id = "downstream_join_id") public void nulls() {}
+ }
+ }
+
+ @Airflow.Deps
+ static class Wiring implements TestExampleDeps {
+ void depends() {}
+ }
+ }
+ """,
+ )
+ assertThat(compilation).failed()
+ assertThat(compilation).hadErrorContaining(
+ "Task 'Transform.Checks.downstream_join_id' has the ID of a join node of
task group 'Transform.Checks'; " +
+ "rename the task",
+ )
+ }
+
+ @Test
+ @DisplayName("reject a nested task group that takes the id of a join node of
the group holding it")
+ fun rejectNestedTaskGroupTakingJoinNodeId() {
+ listOf("upstream_join_id", "downstream_join_id").forEach { id ->
+ val compilation =
+ compile(
+ """
+ package org.apache.airflow.example;
+ import org.apache.airflow.sdk.Airflow;
+ @Airflow.Dag
+ public class TestExample {
+ @Airflow.TaskGroup
+ static class Transform {
+ @Airflow.TaskGroup(id = "$id")
+ static class Inner {
+ @Airflow.Task public void clean() {}
+ }
+ }
+
+ @Airflow.Deps
+ static class Wiring implements TestExampleDeps {
+ void depends() {}
+ }
+ }
+ """,
+ )
+ assertThat(compilation).failed()
+ assertThat(compilation).hadErrorContaining(
+ "Task group 'Transform.$id' has the ID of a join node of task group
'Transform'; rename the nested group",
+ )
+ }
+ }
+
@Test
@DisplayName("reject two task methods whose generated classes would collide")
fun rejectCollidingGeneratedTaskClasses() {
diff --git a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/DagDef.kt
b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/DagDef.kt
index a4150e8a9b6..913817d0909 100644
--- a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/DagDef.kt
+++ b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/DagDef.kt
@@ -23,6 +23,7 @@ import org.apache.airflow.sdk.internal.DagSource
import org.apache.airflow.sdk.internal.GROUP_ID
import org.apache.airflow.sdk.internal.SchemaFields
import org.apache.airflow.sdk.internal.checkConfigValue
+import org.apache.airflow.sdk.internal.deriveJoinNodeIds
import org.apache.airflow.sdk.internal.deriveTaskId
import org.apache.airflow.sdk.internal.validateTaskInput
import java.util.concurrent.atomic.AtomicLong
@@ -44,9 +45,10 @@ import kotlin.Throws
* ```
*
* Dag and task IDs are validated authoritatively by the Airflow server. This
- * class accepts any ID; IDs the server would reject are reported as
- * best-effort build-time warnings by the Gradle plugin's `checkAirflowBundle`
- * task, which runs [BundleInspector] against the assembled [Bundle].
+ * class does not check the characters or length of a Dag or task ID. IDs the
+ * server would reject are reported as best-effort build-time warnings by the
+ * Gradle plugin's `checkAirflowBundle` task, which runs [BundleInspector]
+ * against the assembled [Bundle].
*
* @param id Dag identifier; must be unique within a [Bundle].
*
@@ -161,6 +163,7 @@ class DagDef(
throw IllegalArgumentException("Task '${task.id}' already belongs to Dag
'${owner.id}'")
}
require(task.id !in groups) { "Dag '$id' already has a task group with ID:
${task.id}" }
+ requireNotJoinNode(task.id)
require(tasks.putIfAbsent(task.id, task) == null) {
"Tasks in Dag have duplicate ID: ${task.id}"
}
@@ -288,7 +291,9 @@ class DagDef(
* or dashes, and differ from every task and group ID in this Dag.
* @return The group, to declare tasks in and to wire edges with.
* @throws IllegalArgumentException if [id] is not a valid group ID, or the
- * Dag already has a task or task group with that ID.
+ * Dag already has a task or task group with that ID. Also thrown if a
+ * task already has the ID of one of the group's join nodes.
+ * [TaskGroupRef] explains join nodes.
*/
fun taskGroup(id: String): TaskGroupRef = addGroup(null, id)
@@ -303,12 +308,37 @@ class DagDef(
require(groupId !in tasks && groupId !in groups) {
"Dag '$id' already has a task or task group with ID: $groupId"
}
+ requireNotJoinNode(groupId)
+ // Python accepts a task that took a join node ID of the new group before
+ // the group existed. The Graph view then draws the edges into or out of
+ // the group through that task. This check rejects that order. The Go SDK
+ // rejects it too. Only a task can already have a join node ID of the new
+ // group. A group with that ID would have to be nested inside the new
+ // group, which does not exist yet.
+ deriveJoinNodeIds(groupId).firstOrNull { it in tasks }?.let { taken ->
+ throw IllegalArgumentException(
+ "Dag '$id' already has a task with ID: $taken, which is the ID of a
join node of task group '$groupId'",
+ )
+ }
return TaskGroupRef(this, groupId, parent).also {
groups[groupId] = it
parent?.children?.add(it)
}
}
+ /**
+ * Rejects [nodeId] if a task group of this Dag has a join node with the ID
+ * [nodeId]. A join node's ID is its group's ID, a dot, and a suffix with no
+ * dot in it. So at most one group can have a join node with the ID [nodeId].
+ * That group's ID is the part of [nodeId] before its last dot.
+ */
+ private fun requireNotJoinNode(nodeId: String) {
+ val groupId = nodeId.substringBeforeLast('.', "")
+ require(groupId !in groups || nodeId !in deriveJoinNodeIds(groupId)) {
+ "Dag '$id' already has a join node of task group '$groupId' with ID:
$nodeId"
+ }
+ }
+
/**
* What this Dag's task-group edges mean in terms of tasks.
*
diff --git
a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/TaskGroupRef.kt
b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/TaskGroupRef.kt
index 363f13072cd..49051aecf28 100644
--- a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/TaskGroupRef.kt
+++ b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/TaskGroupRef.kt
@@ -42,6 +42,15 @@ import org.apache.airflow.sdk.internal.deriveTaskId
* each edge as Python reads it, so a group's edges can be drawn before its
* tasks are declared. The edges still resolve in the order they were drawn.
*
+ * The Airflow UI draws the edges into a group through one join node and the
+ * edges out of it through a second join node. For `staging`, the join nodes
+ * have the IDs `staging.upstream_join_id` and `staging.downstream_join_id`,
+ * and no task or other group in the Dag can take either ID. For example,
+ * `staging.task("upstream_join_id", Stage.class)` fails because the new
+ * task's ID would be `staging.upstream_join_id`. `dag.taskGroup("staging")`
+ * also fails if the Dag already has a task with the ID
+ * `staging.upstream_join_id`.
+ *
* @property id Group ID, including any enclosing group's prefix.
*/
class TaskGroupRef internal constructor(
@@ -163,7 +172,9 @@ class TaskGroupRef internal constructor(
* underscores, or dashes.
* @return The nested group.
* @throws IllegalArgumentException if [id] is not a valid group ID, or the
- * Dag already has a task or task group with the resulting ID.
+ * Dag already has a task or task group with the resulting ID. Also
+ * thrown if the resulting ID is the ID of a join node of this group, or
+ * if a task already has the ID of a join node of the nested group.
*/
fun taskGroup(id: String): TaskGroupRef = dag.addGroup(this, id)
diff --git
a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/Ids.kt
b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/Ids.kt
index b2824be3c91..2f38b7c13dc 100644
--- a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/Ids.kt
+++ b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/Ids.kt
@@ -37,3 +37,16 @@ val GROUP_ID: Regex = Regex("[A-Za-z0-9_-]+")
* check an ID it derives the same way; not user-facing API.
*/
fun deriveTaskId(definition: Class<*>): String =
definition.simpleName.replaceFirstChar { it.lowercaseChar() }
+
+/**
+ * @suppress
+ *
+ * The IDs of the two join nodes that the Airflow UI draws for the task group
+ * with full ID [groupId]. Python's `TaskGroup.upstream_join_id` and
+ * `TaskGroup.downstream_join_id` return the same IDs. The UI draws the edges
+ * into the group through the first node and the edges out of it through the
+ * second. Python reserves both IDs once the group exists. The annotation
+ * processor checks these IDs too, so this function is public. It is not part
+ * of the user-facing API.
+ */
+fun deriveJoinNodeIds(groupId: String): List<String> =
listOf("$groupId.upstream_join_id", "$groupId.downstream_join_id")
diff --git
a/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/TaskGroupTest.kt
b/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/TaskGroupTest.kt
index 22c5972f32e..2a69f9155fe 100644
--- a/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/TaskGroupTest.kt
+++ b/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/TaskGroupTest.kt
@@ -79,6 +79,104 @@ internal class TaskGroupTest {
assertEquals("Dag 'd' already has a task group with ID: staging",
error.message)
}
+ @Test
+ @DisplayName("Should reject a task in a group that takes the ID of one of
the group's join nodes")
+ fun shouldRejectTaskIdTakenByJoinNode() {
+ listOf("upstream_join_id", "downstream_join_id").forEach { localId ->
+ val transform = DagDef("d").taskGroup("transform")
+ val checks = transform.taskGroup("checks")
+
+ listOf(transform, checks).forEach { group ->
+ val error =
+ assertThrows(IllegalArgumentException::class.java) {
group.task<Unit>(localId, NoopTask::class.java) }
+
+ assertEquals(
+ "Dag 'd' already has a join node of task group '${group.id}' with
ID: ${group.id}.$localId",
+ error.message,
+ )
+ }
+ }
+ }
+
+ @Test
+ @DisplayName("Should reject a top-level task that takes the ID of a join
node of an earlier group")
+ fun shouldRejectRootTaskIdTakenByJoinNode() {
+ listOf("upstream_join_id", "downstream_join_id").forEach { suffix ->
+ val dag = DagDef("d")
+ dag.taskGroup("transform")
+
+ val error =
+ assertThrows(IllegalArgumentException::class.java) {
+ dag.task<Unit>("transform.$suffix", NoopTask::class.java)
+ }
+
+ assertEquals(
+ "Dag 'd' already has a join node of task group 'transform' with ID:
transform.$suffix",
+ error.message,
+ )
+ }
+ }
+
+ @Test
+ @DisplayName("Should reject a group when a task already has the ID of one of
its join nodes")
+ fun shouldRejectGroupWhoseJoinNodeIdIsTaken() {
+ listOf("upstream_join_id", "downstream_join_id").forEach { suffix ->
+ val dag = DagDef("d")
+ dag.task<Unit>("transform.$suffix", NoopTask::class.java)
+ val staging = dag.taskGroup("staging")
+ staging.task<Unit>("checks.$suffix", NoopTask::class.java)
+
+ val error = assertThrows(IllegalArgumentException::class.java) {
dag.taskGroup("transform") }
+ val nestedError = assertThrows(IllegalArgumentException::class.java) {
staging.taskGroup("checks") }
+
+ assertEquals(
+ "Dag 'd' already has a task with ID: transform.$suffix, " +
+ "which is the ID of a join node of task group 'transform'",
+ error.message,
+ )
+ assertEquals(
+ "Dag 'd' already has a task with ID: staging.checks.$suffix, " +
+ "which is the ID of a join node of task group 'staging.checks'",
+ nestedError.message,
+ )
+ assertEquals(listOf("staging"), dag.groups.keys.toList())
+ }
+ }
+
+ @Test
+ @DisplayName("Should reject a nested group that takes the ID of a join node
of the group holding it")
+ fun shouldRejectNestedGroupIdTakenByJoinNode() {
+ listOf("upstream_join_id", "downstream_join_id").forEach { localId ->
+ val transform = DagDef("d").taskGroup("transform")
+
+ val error = assertThrows(IllegalArgumentException::class.java) {
transform.taskGroup(localId) }
+
+ assertEquals(
+ "Dag 'd' already has a join node of task group 'transform' with ID:
transform.$localId",
+ error.message,
+ )
+ }
+ }
+
+ @Test
+ @DisplayName("Should tell a group's join node IDs apart from top-level IDs
named upstream_join_id or downstream_join_id")
+ fun shouldTellJoinNodeIdsFromIdsNamedLikeThem() {
+ val dag = DagDef("d")
+ dag.task<Unit>("downstream_join_id", NoopTask::class.java)
+ val group = dag.taskGroup("upstream_join_id")
+
+ val error =
+ assertThrows(IllegalArgumentException::class.java) {
+ group.task<Unit>("downstream_join_id", NoopTask::class.java)
+ }
+
+ assertEquals(
+ "Dag 'd' already has a join node of task group 'upstream_join_id' with
ID: upstream_join_id.downstream_join_id",
+ error.message,
+ )
+ assertEquals(listOf("downstream_join_id"), dag.tasks.keys.toList())
+ }
+
@Test
@DisplayName("Should wire a group upstream from its leaves and downstream to
its roots on registration")
fun shouldExpandGroupEdgesOntoRootsAndLeaves() {
