uranusjr commented on code in PR #71057:
URL: https://github.com/apache/airflow/pull/71057#discussion_r3773767261
##########
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/DagDef.kt:
##########
@@ -24,46 +24,72 @@ import kotlin.Throws
/**
* A collection of tasks with directional dependencies.
*
- * Create a [Dag] directly and register tasks with [addTask].
+ * Create a [DagDef] directly and register [TaskDef]s with [addTask].
*
* The [Builder.Dag] annotation should generally be preferred in user code,
* where the annotation processor generates the wiring for you. Only use this
- * class directly if you need to do low-level plumbing.
+ * class directly if you need to do low-level plumbing:
+ *
+ * ```java
+ * var dag = new DagDef("java_etl")
+ * .addTask(new TaskDef("extract", Extract.class))
+ * .addTask(new TaskDef("load", Load.class));
+ * ```
*
* @param id Dag identifier. Must contain only ASCII alphanumeric characters,
* dashes, dots, or underscores; must be unique within a [Bundle].
*
* @see Builder.Dag
*/
-class Dag(
+class DagDef(
val id: String, // TODO: charset check?
) {
- internal var tasks = mutableMapOf<String, Class<out Task>>()
+ internal val tasks = linkedMapOf<String, TaskDef>()
/**
* Registers a task with this Dag.
*
- * The class must have a public no-argument constructor and implement [Task].
- * Task IDs must be unique within a Dag.
+ * A [TaskDef] belongs to at most one [DagDef]; registering the same instance
+ * with a second Dag, or twice with the same one, fails. Task IDs must be
+ * unique within a Dag.
*
- * @param id Task identifier, unique within this Dag.
- * @param definition Class that implements [Task]. Must have a public no-arg
- * constructor.
+ * @param task Task definition to register.
* @return This Dag, for chaining.
- * @throws IllegalArgumentException if a task already exists in the Dag with
- * the same ID.
+ * @throws IllegalArgumentException if the task already belongs to a Dag or a
+ * task with the same ID is already registered.
*/
- fun addTask(
- id: String,
- definition: Class<out Task>,
- ): Dag {
- require(tasks.putIfAbsent(id, definition) == null) {
- "Tasks in Dag have duplicate ID: $id"
+ fun addTask(task: TaskDef): DagDef {
+ require(task.owner == null) {
+ "Task '${task.id}' already belongs to Dag '${task.owner?.id}'"
Review Comment:
```suggestion
"Task '${task.id}' already belongs to Dag '${task.owner!!.id}'"
```
Since the `require` call already established the value is not null.
But personally I would do this instead
```kotlin
task.owner?.let { owner ->
throw IllegalArgumentException("Task '${task.id}' already belongs to Dag
'${owner.id}'")
}
```
to avoid the second unwrap.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
