kaxil commented on code in PR #74438: URL: https://github.com/apache/airflow/pull/74438#discussion_r4218032839
########## java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Condition.kt: ########## @@ -0,0 +1,218 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.airflow.sdk + +import kotlin.Throws + +/** + * A task whose boolean decides which of two tasks runs; the other is skipped. + * + * Register one with [DagDef.If], then name the sides with [ConditionRef.then] + * and [ConditionRef.orElse]: + * + * ```java + * public class HasRows implements ConditionTask { + * @Override + * public boolean decide(Context context, Client client) { + * return ((Number) client.getXCom("extract")).longValue() > 0; + * } + * } + * ``` + * + * The SDK runs [decide] and pushes its result as this task's return value, so + * [execute] is never called. + * + * @see DagDef.If + */ +interface ConditionTask : Task { + /** + * Decides which side of the condition runs. + * + * Any exception thrown marks the task instance as failed, and nothing is + * then pushed or skipped. + * + * @param context Runtime context for the current execution workload. + * @param client Client for Airflow API calls scoped to this execution. + * @return True to run the [ConditionRef.then] side, false the + * [ConditionRef.orElse] side. + * @throws Exception on failure; the task instance is marked failed. + */ + @Throws(Exception::class) + fun decide( + context: Context, + client: Client, + ): Boolean + + /** Never called: the SDK runs a condition through [decide]. */ + override fun execute( + context: Context, + client: Client, + ): Unit = + throw IllegalStateException( + "Condition '${javaClass.name}' runs through decide(), so execute() is never called", + ) +} + +/** + * A condition registered with a Dag: name the task each outcome runs. + * + * ```java + * dag.If(HasRows.class).then(load).orElse(reportEmpty); + * ``` + * + * A named task runs after the condition, so naming it records that edge, as + * [Deps.Flow.before] would. The condition skips only the side not taken; what + * runs after both sides needs a trigger rule that tolerates one skipped + * upstream, such as `none_failed_min_one_success`. + * + * @see DagDef.If + */ +class ConditionRef private constructor( + private val ref: TaskRef<Boolean>, + private val sides: ConditionDef, +) : Deps.Flow { + companion object { + /** + * @suppress + * + * Marks a registered task as a condition. Public so a generated wiring + * view can call it; user code reaches a condition through [DagDef.If]. + */ + @JvmStatic + fun of(ref: TaskRef<Boolean>): ConditionRef { + val sides = ConditionDef() + require(ref.def.decider == null) { Review Comment: A no-arg `@Builder.If` method called twice in the wiring trips this. `Refs.node` memoizes by task id and hands back the same `TaskRef`, so the second generated `ConditionRef.of(...)` finds the decider already set and fails with "already decides what to skip; declare it once", e.g. `extract().before(hasRows()); hasRows().then(load());`. Every no-arg `@Builder.Task` view method allows that reuse. Could `of` return a `ConditionRef` over the existing `ConditionDef` when there is one? `check` still rejects naming a side twice, and `DagDef.If` / `TaskGroupRef.If` can't reach it because `task(id, ...)` already rejects a duplicate id. A BuilderTest calling a no-arg condition twice would cover it. ########## java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Condition.kt: ########## @@ -0,0 +1,218 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.airflow.sdk + +import kotlin.Throws + +/** + * A task whose boolean decides which of two tasks runs; the other is skipped. + * + * Register one with [DagDef.If], then name the sides with [ConditionRef.then] + * and [ConditionRef.orElse]: + * + * ```java + * public class HasRows implements ConditionTask { + * @Override + * public boolean decide(Context context, Client client) { + * return ((Number) client.getXCom("extract")).longValue() > 0; + * } + * } + * ``` + * + * The SDK runs [decide] and pushes its result as this task's return value, so + * [execute] is never called. + * + * @see DagDef.If + */ +interface ConditionTask : Task { + /** + * Decides which side of the condition runs. + * + * Any exception thrown marks the task instance as failed, and nothing is + * then pushed or skipped. + * + * @param context Runtime context for the current execution workload. + * @param client Client for Airflow API calls scoped to this execution. + * @return True to run the [ConditionRef.then] side, false the + * [ConditionRef.orElse] side. + * @throws Exception on failure; the task instance is marked failed. + */ + @Throws(Exception::class) + fun decide( + context: Context, + client: Client, + ): Boolean + + /** Never called: the SDK runs a condition through [decide]. */ + override fun execute( + context: Context, + client: Client, + ): Unit = + throw IllegalStateException( + "Condition '${javaClass.name}' runs through decide(), so execute() is never called", + ) +} + +/** + * A condition registered with a Dag: name the task each outcome runs. + * + * ```java + * dag.If(HasRows.class).then(load).orElse(reportEmpty); + * ``` + * + * A named task runs after the condition, so naming it records that edge, as + * [Deps.Flow.before] would. The condition skips only the side not taken; what + * runs after both sides needs a trigger rule that tolerates one skipped + * upstream, such as `none_failed_min_one_success`. + * + * @see DagDef.If + */ +class ConditionRef private constructor( + private val ref: TaskRef<Boolean>, + private val sides: ConditionDef, +) : Deps.Flow { + companion object { + /** + * @suppress + * + * Marks a registered task as a condition. Public so a generated wiring + * view can call it; user code reaches a condition through [DagDef.If]. + */ + @JvmStatic + fun of(ref: TaskRef<Boolean>): ConditionRef { + val sides = ConditionDef() + require(ref.def.decider == null) { + "Task '${ref.def.id}' already decides what to skip; declare it once" + } + ref.def.decider = sides + return ConditionRef(ref, sides) + } + } + + /** Airflow task ID of the deciding task. */ + val id: String get() = ref.def.id + + /** + * Names the task that runs when the condition holds. + * + * @param task Task of the same Dag, which the condition skips when it does + * not hold. + * @return This condition, so [orElse] can follow. + * @throws IllegalArgumentException if the task belongs to another Dag, or + * this side is already named. + */ + fun then(task: TaskRef<*>): ConditionRef = name("then", task) { sides.whenTrue = it } + + /** + * Names the task that runs when the condition does not hold. A condition + * needs no `orElse`; without one, nothing is skipped when it holds. + * + * @param task Task of the same Dag, which the condition skips when it holds. + * @return This condition, for chaining. + * @throws IllegalArgumentException if the task belongs to another Dag, is + * the `then` task, or this side is already named. + */ + fun orElse(task: TaskRef<*>): ConditionRef = name("orElse", task) { sides.whenFalse = it } Review Comment: `If` is capitalized because `if` is a Java keyword, but `else` is one too, and here it becomes `orElse`. ADR 0008 names the constructs after the host language's `if`/`else`, and Go ships `Then`/`Else`. Would `Else(...)` read more consistently with `If`? Renaming after a release needs a deprecation, so now is the cheap time to decide. ########## java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/ConditionTest.kt: ########## @@ -0,0 +1,179 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.airflow.sdk + +import org.apache.airflow.sdk.execution.serializeDag +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertFalse +import org.junit.jupiter.api.Assertions.assertThrows +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.DisplayName +import org.junit.jupiter.api.Test + +/** A condition the tests register; what it decides is driven from the runtime tests. */ +class HasRows : ConditionTask { + override fun decide( + context: Context, + client: Client, + ): Boolean = decision + + companion object { + var decision: Boolean = true Review Comment: nit: `decision` is never assigned, and the runtime tests in `TaskTest` use their own `TestCondition`, so the KDoc above ("driven from the runtime tests") doesn't hold. Returning `true` directly would do. Same for `FakeTransport.skipped` in `ClientTest`, which nothing asserts. ########## java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Condition.kt: ########## @@ -0,0 +1,218 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.airflow.sdk + +import kotlin.Throws + +/** + * A task whose boolean decides which of two tasks runs; the other is skipped. + * + * Register one with [DagDef.If], then name the sides with [ConditionRef.then] + * and [ConditionRef.orElse]: + * + * ```java + * public class HasRows implements ConditionTask { + * @Override + * public boolean decide(Context context, Client client) { + * return ((Number) client.getXCom("extract")).longValue() > 0; + * } + * } + * ``` + * + * The SDK runs [decide] and pushes its result as this task's return value, so + * [execute] is never called. + * + * @see DagDef.If + */ +interface ConditionTask : Task { + /** + * Decides which side of the condition runs. + * + * Any exception thrown marks the task instance as failed, and nothing is + * then pushed or skipped. + * + * @param context Runtime context for the current execution workload. + * @param client Client for Airflow API calls scoped to this execution. + * @return True to run the [ConditionRef.then] side, false the + * [ConditionRef.orElse] side. + * @throws Exception on failure; the task instance is marked failed. + */ + @Throws(Exception::class) + fun decide( + context: Context, + client: Client, + ): Boolean + + /** Never called: the SDK runs a condition through [decide]. */ + override fun execute( Review Comment: Since `ConditionTask` extends `Task`, `dag.task("has_rows", HasRows.class)` compiles and registers fine. At runtime there's no decider, so this throws on every try and the generic catch in `runTask` returns `failure(shouldRetry)`, burning the retries before it fails. Could `Bundle.register` require that a task's class implements `ConditionTask` exactly when its decider is a `ConditionDef`? That moves the mistake to import time, next to the other structural checks there. ########## java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/execution/TaskTest.kt: ########## @@ -271,9 +337,84 @@ class TaskTest { ): Unit = throw UnsupportedOperationException("not used in test") override fun clearTaskStateStore(tiId: UUID): Unit = throw UnsupportedOperationException("not used in test") + + override fun skipDownstreamTasks(taskIds: List<String>): Unit = throw UnsupportedOperationException("not used in test") }, ) + /** Decides what [decision] holds; a null one throws, standing for a condition body that fails. */ + class TestCondition : ConditionTask { + override fun decide( + context: Context, + client: Client, + ): Boolean = decision ?: throw IllegalStateException("boom") + + companion object { + var decision: Boolean? = true + } + } + + /** Records what a deciding task pushed and asked to skip, in order. */ Review Comment: The KDoc says "in order", but `xComs` and `skipped` are separate lists, so swapping the `skipmixin_key` push and the `skipDownstreamTasks` call in `decide()` keeps every test green. That order is what makes a cleared side get skipped again. Recording one event list and asserting `[xcom skipmixin_key, skip [...], xcom return_value]` would pin it. Nothing exercises `CoordinatorClient.skipDownstreamTasks` either, since every test goes through a fake; a `CommTest` round trip for the new frame would cover the encoding and the empty-list short-circuit. -- 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]
