kaxil commented on code in PR #74440:
URL: https://github.com/apache/airflow/pull/74440#discussion_r4216959274
##########
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/DagDef.kt:
##########
@@ -404,6 +429,20 @@ class TaskDef(
val id: String,
val definition: Class<out Task>,
) : Endpoint {
+ /**
+ * Declares a task that starts a run of another Dag instead of running Java
+ * code.
+ *
+ * [DagDef.task] is the spelling user code uses; this is what a generated
+ * wiring view calls.
+ *
+ * @param id Task identifier, unique within a [DagDef].
+ * @param trigger What to trigger, and how.
+ */
+ constructor(id: String, trigger: TriggerDagRun) : this(id,
TriggerDagRunPlaceholder::class.java) {
Review Comment:
Every trigger task gets `TriggerDagRunPlaceholder` as its class, and
`BranchDef.add` rejects two cases sharing a class when the decider is a
`BranchTask`. So `dag.Branch(Pick.class).option(dag.task("a", new
TriggerDagRun("x"))).option(dag.task("b", new TriggerDagRun("y")))` fails with
"both run 'org.apache.airflow.sdk.TriggerDagRunPlaceholder'", a class the
author never wrote. With a single trigger case, `byClass` can only pick it if
`choose()` returns that internal class. Picking which downstream Dag to trigger
seems like a natural use of a branch. Could `BranchDef.add` reject a trigger
case under a `BranchTask` with a message pointing at `TaskIdBranchTask` (which
works, since it matches by ID), or could trigger tasks stop sharing one class?
A test with a branch over trigger tasks would pin whichever you pick.
##########
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Task.kt:
##########
@@ -102,6 +102,19 @@ internal object TaskRunner {
val taskDef =
bundle.taskDef(request.ti.dagId, request.ti.taskId)
?: return TaskResult.of(TaskState.State.REMOVED)
+ taskDef.trigger?.let { trigger ->
Review Comment:
I couldn't find a test that runs a trigger task through `runTask`.
`TriggerRunnerTest` goes straight to `TriggerRunner.run`, and `TaskTest` only
adds throwing stubs. So if this dispatch got deleted, every trigger task would
instantiate the placeholder and throw, and the suite would still pass. Same for
the catch below. Could you add a `runTask` test that checks the
`trigger_run_id` XCom is pushed? Another where the client throws and the task
ends up FAILED would cover the catch.
##########
java-sdk/example/src/java/org/apache/airflow/example/nativedag/InterfaceExample.java:
##########
@@ -114,12 +114,18 @@ public static DagDef build() {
var reportLong = dag.task("report_long", ReportLong.class);
var reportShort = dag.task("report_short", ReportShort.class);
+ // A task that starts a run of another Dag; it runs no Java code.
+ var trigger =
+ dag.task(
+ "trigger_downstream",
+ new
TriggerDagRun("java_native_annotation_example").config("wait_for_completion",
false));
Review Comment:
This Dag triggers `java_native_annotation_example`, and
`AnnotationExample.triggerDownstream` triggers this one back. Both are
`@daily`, `load` always wins the condition with the fixed extract value, and
neither sets `trigger_run_id`, so every trigger creates a fresh `manual__<now>`
run and never hits the already-exists path. Once both are unpaused, one
scheduled run starts a chain of runs that never ends. Could one of them trigger
a Dag that doesn't trigger back, the way Python's
`example_trigger_controller_dag` only points one way?
##########
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/TriggerDagRun.kt:
##########
@@ -0,0 +1,202 @@
+/*
+ * 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 java.time.Duration
+import java.time.Instant
+import java.time.OffsetDateTime
+
+/** A Dag run state, as `allowed_states` and `failed_states` name one. */
+internal val DAG_RUN_STATES = listOf("queued", "running", "success", "failed")
+
+/** Shape of one `TriggerDagRun` setting, for [TriggerDagRun.config] to check
a value against. */
+private enum class Setting {
+ STRING,
+ BOOLEAN,
+ DURATION,
+ DATETIME,
+ STATES,
+ CONF,
+}
+
+/**
+ * Settings named as `TriggerDagRunOperator` names its parameters. The Dag ID
+ * is the constructor argument, so `trigger_dag_id` is not among them.
+ */
+private val SETTINGS =
+ linkedMapOf(
+ "trigger_run_id" to Setting.STRING,
+ "conf" to Setting.CONF,
+ "logical_date" to Setting.DATETIME,
+ "run_after" to Setting.DATETIME,
+ "reset_dag_run" to Setting.BOOLEAN,
+ "wait_for_completion" to Setting.BOOLEAN,
+ "poke_interval" to Setting.DURATION,
+ "allowed_states" to Setting.STATES,
+ "failed_states" to Setting.STATES,
+ "skip_when_already_exists" to Setting.BOOLEAN,
+ "fail_when_dag_is_paused" to Setting.BOOLEAN,
+ "note" to Setting.STRING,
+ "deferrable" to Setting.BOOLEAN,
+ )
+
+/**
+ * A task that starts a run of another Dag, in place of a task class.
+ *
+ * Pass one to [DagDef.task] and the Java runtime runs it as Airflow's
+ * `TriggerDagRunOperator` does, so the Dag needs no Python worker for it:
+ *
+ * ```java
+ * dag.task("trigger_downstream",
+ * new TriggerDagRun("downstream_etl")
+ * .config("wait_for_completion", true)
+ * .config("conf", Map.of("rows", 2)));
+ * ```
+ *
+ * In an annotated Dag, a `@Builder.Task` method returns one instead; the
+ * method runs when the Dag is built, not when the task runs.
+ *
+ * The task runs no Java code, takes no arguments and pushes no result other
+ * than the triggered run's ID. It renders no templates, so a value such as
+ * `"{{ ds }}"` reaches the new run unchanged.
+ *
+ * @param dagId `trigger_dag_id`: the Dag to trigger.
+ * @throws IllegalArgumentException if [dagId] is empty.
+ */
+class TriggerDagRun(
+ val dagId: String,
+) {
+ init {
+ require(dagId.isNotEmpty()) { "TriggerDagRun needs the ID of the Dag to
trigger" }
+ }
+
+ internal val settings = linkedMapOf<String, Any>()
+
+ /**
+ * Sets one `TriggerDagRunOperator` setting, named as Python names the
+ * parameter.
+ *
+ * | Key | Value |
+ * | --- | --- |
+ * | `trigger_run_id` | `String`; generated from the trigger time when unset
|
+ * | `conf` | `Map` with string keys and JSON values |
+ * | `logical_date` | [OffsetDateTime] or [Instant]; the trigger time when
unset |
Review Comment:
When only `run_after` is set, the run gets no logical date
(`TriggerRunner.kt:78-83`, same as Python), and `trigger_run_id` on line 98 is
built from `run_after` plus a random 8-character suffix rather than from the
trigger time. Lines 76-77 also say the task pushes only the run ID, but it
pushes `_link_TriggerDagRunLink` too. The code looks right; it's just the docs.
##########
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/TriggerRunner.kt:
##########
@@ -0,0 +1,318 @@
+/*
+ * 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.
+ */
+
+// Mirrors TriggerDagRunOperator.execute and execute_complete in the standard
+// provider, as the Go and TypeScript SDK runtimes do.
Review Comment:
On this base the Go runtime doesn't run trigger tasks itself;
`go-sdk/airflow/bundle.go` hands them to a Python worker, and the Go runtime
version is still open in #74417. Maybe just say TypeScript here?
##########
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Serde.kt:
##########
@@ -165,6 +170,105 @@ private fun serializeTask(
)
}
+/** What Python's serializer writes for a `TriggerDagRunOperator`. */
+private const val TRIGGER_TASK_TYPE = "TriggerDagRunOperator"
+private const val TRIGGER_TASK_MODULE =
"airflow.providers.standard.operators.trigger_dagrun"
+private const val TRIGGER_UI_COLOR = "#ffefeb"
+
+private val TRIGGER_TEMPLATE_FIELDS =
+ listOf(
+ "trigger_dag_id",
+ "trigger_run_id",
+ "logical_date",
+ "conf",
+ "wait_for_completion",
+ "skip_when_already_exists",
+ )
+
+/**
+ * Writes a task declared from a [TriggerDagRun] as Python writes a
+ * `TriggerDagRunOperator`, so the Airflow UI and the Dag dependency graph
+ * treat it as one.
+ *
+ * Python writes an operator's template fields and leaves its other parameters
+ * out, because a Python Dag file holds them. A Java Dag has no such file, so
+ * every setting the author made is written under the name Python's parameter
+ * carries. The runtime runs the task from the Dag in its own bundle and reads
+ * none of this back.
+ */
+private fun writeTriggerDagRun(
+ data: MutableMap<String, Any?>,
+ trigger: TriggerDagRun,
+) {
+ data["task_type"] = TRIGGER_TASK_TYPE
+ data["_task_module"] = TRIGGER_TASK_MODULE
+ data["ui_color"] = TRIGGER_UI_COLOR
+ data["template_fields"] = TRIGGER_TEMPLATE_FIELDS
+ data["template_fields_renderers"] = mapOf("conf" to "py")
+ data["_operator_extra_links"] = mapOf("Triggered DAG" to
"_link_TriggerDagRunLink")
+ data["trigger_dag_id"] = trigger.dagId
+
+ val settings = trigger.settings
+ // Python's logical_date defaults to NOTSET, which lets the operator pick the
+ // trigger time, and writes that sentinel as its name. A template field holds
+ // a datetime as str(datetime), not as a {"__type": "datetime"} object.
+ data["logical_date"] = settings["logical_date"]?.let { pythonStr(it) } ?:
"NOTSET"
+ // Python writes these two template fields whatever they hold.
+ data["wait_for_completion"] = settings["wait_for_completion"] ?: false
+ data["skip_when_already_exists"] = settings["skip_when_already_exists"] ?:
false
+ settings["trigger_run_id"]?.let { data["trigger_run_id"] = it }
+ settings["conf"]?.let { data["conf"] = plainJson(it, "conf", trigger.dagId) }
Review Comment:
A non-JSON value inside `conf` gets past `config()` (which only checks that
keys are strings) and fails here instead, with `trigger.dagId` passed as the
task ID. The import error then reads "Argument 'conf' of task
'downstream_etl'...", naming the target Dag as a task and talking about call
arguments. Could `config()` run the JSON check, so it fails where the KDoc says
it does?
##########
scripts/ci/lang_sdk_serialization/test_dags.yaml:
##########
@@ -305,7 +305,7 @@ dags:
trigger_dag_id: conformance_minimal
trigger_run_id: "triggered_{{ run_id }}"
conf: {rows: 2, ratio: 0.5, nested: {tags: [a, b]}}
- logical_date: "{{ ds }}"
+ logical_date: !datetime 2026-09-30T00:00:00+00:00
Review Comment:
With this line, Static checks on this PR fails in
`check-go-sdk-serialization-conformance`: `conformance_trigger_dag_run: task
trigger_all: logical_date is '2026-09-30T00:00:00+00:00', Python writes
'2026-09-30 00:00:00+00:00'`. If this lands before #74417, that hook stays red
on main for every later go-sdk or fixture change. Can #74417 go first, or this
line move into it so the fixture and the Go fix land together?
##########
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/DagDef.kt:
##########
@@ -404,6 +429,20 @@ class TaskDef(
val id: String,
val definition: Class<out Task>,
) : Endpoint {
+ /**
+ * Declares a task that starts a run of another Dag instead of running Java
+ * code.
+ *
+ * [DagDef.task] is the spelling user code uses; this is what a generated
+ * wiring view calls.
+ *
+ * @param id Task identifier, unique within a [DagDef].
+ * @param trigger What to trigger, and how.
+ */
+ constructor(id: String, trigger: TriggerDagRun) : this(id,
TriggerDagRunPlaceholder::class.java) {
+ this.trigger = trigger
Review Comment:
This keeps the caller's `TriggerDagRun` by reference, and `config()` mutates
it and returns `this`. Reusing one builder for two tasks, or calling `config()`
on it after `dag.task(...)`, silently changes the task that's already
registered. Go's `copyTriggerDagRunSpec` deep-copies for this reason. Could
this snapshot `settings` (with a copy of `conf`) at registration?
##########
java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/execution/TriggerRunnerTest.kt:
##########
@@ -0,0 +1,432 @@
+/*
+ * 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.execution
+
+import org.apache.airflow.sdk.Client
+import org.apache.airflow.sdk.TriggerDagRun
+import org.apache.airflow.sdk.execution.comm.DagRun
+import org.apache.airflow.sdk.execution.comm.DeferTask
+import org.apache.airflow.sdk.execution.comm.RetryTask
+import org.apache.airflow.sdk.execution.comm.StartupDetails
+import org.apache.airflow.sdk.execution.comm.SucceedTask
+import org.apache.airflow.sdk.execution.comm.TIRunContext
+import org.apache.airflow.sdk.execution.comm.TaskInstance
+import org.apache.airflow.sdk.execution.comm.TaskState
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertInstanceOf
+import org.junit.jupiter.api.Assertions.assertNull
+import org.junit.jupiter.api.Assertions.assertTrue
+import org.junit.jupiter.api.DisplayName
+import org.junit.jupiter.api.Test
+import java.time.Duration
+import java.time.OffsetDateTime
+import java.util.UUID
+
+/** One call to [FakeTrigger.triggerDagRun], so a test can read what the task
asked Airflow for. */
+private class TriggerCall(
+ val dagId: String,
+ val runId: String,
+ val logicalDate: OffsetDateTime?,
+ val runAfter: OffsetDateTime?,
+ val conf: Map<String, Any?>?,
+ val resetDagRun: Boolean,
+ val note: String?,
+)
+
+private class FakeTrigger(
+ private val alreadyExists: Boolean = false,
+ private val paused: Boolean = false,
+ private val states: List<String> = emptyList(),
+) : org.apache.airflow.sdk.execution.Client {
+ val triggered = mutableListOf<TriggerCall>()
+ val xComs = mutableListOf<Pair<String, Any>>()
+ var polls = 0
+
+ override fun triggerDagRun(
+ dagId: String,
+ runId: String,
+ logicalDate: OffsetDateTime?,
+ runAfter: OffsetDateTime?,
+ conf: Map<String, Any?>?,
+ resetDagRun: Boolean,
+ note: String?,
+ ): Boolean {
+ triggered += TriggerCall(dagId, runId, logicalDate, runAfter, conf,
resetDagRun, note)
+ return alreadyExists
+ }
+
+ override fun getDagRunState(
+ dagId: String,
+ runId: String,
+ ): String = states[polls++.coerceAtMost(states.lastIndex)]
Review Comment:
This fake ignores `dagId` and `runId`, so polling the wrong run would stay
green. Could it assert they match what `triggerDagRun` received? A few branches
are also only tested on one side: `deferrable` left unset and following the env
default, the queue staying null when triggerer queues are off, and
`fail_when_dag_is_paused` set on an unpaused Dag.
##########
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/TriggerRunner.kt:
##########
@@ -0,0 +1,318 @@
+/*
+ * 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.
+ */
+
+// Mirrors TriggerDagRunOperator.execute and execute_complete in the standard
+// provider, as the Go and TypeScript SDK runtimes do.
+
+package org.apache.airflow.sdk.execution
+
+import org.apache.airflow.sdk.Client
+import org.apache.airflow.sdk.TriggerDagRun
+import org.apache.airflow.sdk.execution.comm.DeferTask
+import org.apache.airflow.sdk.execution.comm.StartupDetails
+import org.apache.airflow.sdk.execution.comm.TaskState
+import java.security.SecureRandom
+import java.time.Duration
+import java.time.Instant
+import java.time.OffsetDateTime
+import java.time.ZoneOffset
+import java.time.format.DateTimeFormatter
+import kotlin.random.asKotlinRandom
+
+/** `DagStateTrigger`, the Python triggerer runs the wait of a deferred task
with. */
+private const val DAG_STATE_TRIGGER =
"airflow.providers.standard.triggers.external_task.DagStateTrigger"
+
+/** `TriggerDagRunLink().xcom_key`: the XCom the "Triggered DAG" extra link
reads. */
+private const val LINK_XCOM_KEY = "_link_TriggerDagRunLink"
+private const val RUN_ID_XCOM_KEY = "trigger_run_id"
+
+/** `TRIGGER_FAIL_REPR`: the `next_method` a failed or timed-out trigger
resumes with. */
+private const val TRIGGER_FAIL = "__fail__"
+private const val EXECUTE_COMPLETE = "execute_complete"
+
+private val DEFAULT_POKE_INTERVAL = Duration.ofSeconds(60)
+
+internal object TriggerRunner {
+ private val logger = Logger(TriggerRunner::class)
+
+ /**
+ * Runs a task declared from a [TriggerDagRun], and reports the state it
+ * leaves the task instance in.
+ *
+ * A task resumed by the triggerer reads its outcome from the event the
+ * trigger fired with, rather than triggering a second run.
+ */
+ fun run(
+ trigger: TriggerDagRun,
+ request: StartupDetails,
+ client: Client,
+ env: (String) -> String? = System::getenv,
+ sleep: (Duration) -> Unit = { Thread.sleep(it.toMillis()) },
+ now: () -> OffsetDateTime = { OffsetDateTime.now(ZoneOffset.UTC) },
+ randomSuffix: () -> String = ::randomRunIdSuffix,
+ ): Any {
+ val nextMethod = request.tiContext?.nextMethod as String?
+ if (nextMethod != null) return resume(trigger, request, nextMethod)
+
+ val runAfter = asOffsetDateTime(trigger.settings["run_after"])
+ // `DagRun.generate_run_id`: a run the task gave only a run_after has no
logical date, and is
+ // named after run_after plus a random suffix. Otherwise the logical date
names it, and is the
+ // trigger time when the task set none.
+ val configuredLogicalDate =
asOffsetDateTime(trigger.settings["logical_date"])
+ val logicalDate =
+ when {
+ configuredLogicalDate != null -> configuredLogicalDate
+ runAfter == null -> now()
+ else -> null
+ }
+ val namesRun = runAfter ?: logicalDate!!
+ val runId =
+ trigger.setting<String>("trigger_run_id")
Review Comment:
`?:` only falls back on null. Python uses `if self.trigger_run_id:`, so an
empty string means "generate one" there, while here `""` goes out as the run
ID. `?.takeIf { it.isNotEmpty() }`, or rejecting `""` in `config()` the way the
constructor rejects an empty `dagId`, would match. Same for `note`.
##########
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/TriggerDagRun.kt:
##########
@@ -0,0 +1,202 @@
+/*
+ * 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 java.time.Duration
+import java.time.Instant
+import java.time.OffsetDateTime
+
+/** A Dag run state, as `allowed_states` and `failed_states` name one. */
+internal val DAG_RUN_STATES = listOf("queued", "running", "success", "failed")
+
+/** Shape of one `TriggerDagRun` setting, for [TriggerDagRun.config] to check
a value against. */
+private enum class Setting {
+ STRING,
+ BOOLEAN,
+ DURATION,
+ DATETIME,
+ STATES,
+ CONF,
+}
+
+/**
+ * Settings named as `TriggerDagRunOperator` names its parameters. The Dag ID
+ * is the constructor argument, so `trigger_dag_id` is not among them.
+ */
+private val SETTINGS =
+ linkedMapOf(
+ "trigger_run_id" to Setting.STRING,
+ "conf" to Setting.CONF,
+ "logical_date" to Setting.DATETIME,
+ "run_after" to Setting.DATETIME,
+ "reset_dag_run" to Setting.BOOLEAN,
+ "wait_for_completion" to Setting.BOOLEAN,
+ "poke_interval" to Setting.DURATION,
+ "allowed_states" to Setting.STATES,
+ "failed_states" to Setting.STATES,
+ "skip_when_already_exists" to Setting.BOOLEAN,
+ "fail_when_dag_is_paused" to Setting.BOOLEAN,
+ "note" to Setting.STRING,
+ "deferrable" to Setting.BOOLEAN,
+ )
+
+/**
+ * A task that starts a run of another Dag, in place of a task class.
+ *
+ * Pass one to [DagDef.task] and the Java runtime runs it as Airflow's
+ * `TriggerDagRunOperator` does, so the Dag needs no Python worker for it:
+ *
+ * ```java
+ * dag.task("trigger_downstream",
+ * new TriggerDagRun("downstream_etl")
+ * .config("wait_for_completion", true)
+ * .config("conf", Map.of("rows", 2)));
+ * ```
+ *
+ * In an annotated Dag, a `@Builder.Task` method returns one instead; the
+ * method runs when the Dag is built, not when the task runs.
+ *
+ * The task runs no Java code, takes no arguments and pushes no result other
+ * than the triggered run's ID. It renders no templates, so a value such as
+ * `"{{ ds }}"` reaches the new run unchanged.
+ *
+ * @param dagId `trigger_dag_id`: the Dag to trigger.
+ * @throws IllegalArgumentException if [dagId] is empty.
+ */
+class TriggerDagRun(
+ val dagId: String,
+) {
+ init {
+ require(dagId.isNotEmpty()) { "TriggerDagRun needs the ID of the Dag to
trigger" }
+ }
+
+ internal val settings = linkedMapOf<String, Any>()
+
+ /**
+ * Sets one `TriggerDagRunOperator` setting, named as Python names the
+ * parameter.
+ *
+ * | Key | Value |
+ * | --- | --- |
+ * | `trigger_run_id` | `String`; generated from the trigger time when unset
|
+ * | `conf` | `Map` with string keys and JSON values |
+ * | `logical_date` | [OffsetDateTime] or [Instant]; the trigger time when
unset |
+ * | `run_after` | [OffsetDateTime] or [Instant] |
+ * | `reset_dag_run` | `Boolean`: clear a run with the same ID instead of
failing |
+ * | `wait_for_completion` | `Boolean`: hold this task open until the run
finishes |
+ * | `poke_interval` | [Duration], whole seconds; 60 seconds when unset |
+ * | `allowed_states` | states that count as success; `success` when unset |
+ * | `failed_states` | states that count as failure; `failed` when unset |
+ * | `skip_when_already_exists` | `Boolean`: skip rather than fail when the
run exists |
+ * | `fail_when_dag_is_paused` | `Boolean`: fail rather than trigger a
paused Dag |
+ * | `note` | `String` recorded against the triggered run |
+ * | `deferrable` | `Boolean`: while waiting, defer instead of holding the
worker |
+ *
+ * A `deferrable` left unset follows Airflow's `[operators]
default_deferrable`.
+ *
+ * @param key One of the keys above.
+ * @param value Value of the shape that key takes.
+ * @return This task, for chaining.
+ * @throws IllegalArgumentException if the key is unknown or the value does
+ * not match it.
+ */
+ fun config(
Review Comment:
The 13 keys here are a closed set, but `.config("poke_interval", 30)` or
`.config("waitForCompletion", true)` compiles and only throws once the Dag is
built. Go (`TriggerDagRunSpec` with `[]DagRunState`) and TS both use a typed
spec. Would typed setters fit better, e.g. `new
TriggerDagRun("etl").waitForCompletion(true).pokeInterval(Duration.ofSeconds(30)).allowedStates(DagRunState.SUCCESS)`?
I see `DagDef.config` and `TaskDef.config` take string keys, but those come
from the schema, and this is cheapest to change before a release.
##########
java-sdk/processor/src/main/kotlin/org/apache/airflow/sdk/BuilderProcessor.kt:
##########
@@ -439,7 +442,13 @@ class BuilderProcessor : AbstractProcessor() {
val taskDef =
CodeBlock
.builder()
- .add($$"new $T($S, $L.class)", TASK_DEF_TYPE, decl.id, classRef)
+ .apply {
+ if (decl.kind == TaskKind.TRIGGER) {
+ add($$"new $T($S, new $T().$L())", TASK_DEF_TYPE, decl.id,
ClassName.get(decl.owner), decl.method.simpleName)
Review Comment:
If a trigger method declares a checked exception (say it reads conf from a
file and throws `IOException`), this generated call has no `throws` and javac
reports "unreported exception" in a file the author never wrote. `depends()`
already gets a named check via `isUnchecked`; the same check on trigger methods
would give a readable error.
##########
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Serde.kt:
##########
@@ -165,6 +170,105 @@ private fun serializeTask(
)
}
+/** What Python's serializer writes for a `TriggerDagRunOperator`. */
+private const val TRIGGER_TASK_TYPE = "TriggerDagRunOperator"
+private const val TRIGGER_TASK_MODULE =
"airflow.providers.standard.operators.trigger_dagrun"
+private const val TRIGGER_UI_COLOR = "#ffefeb"
+
+private val TRIGGER_TEMPLATE_FIELDS =
+ listOf(
+ "trigger_dag_id",
+ "trigger_run_id",
+ "logical_date",
+ "conf",
+ "wait_for_completion",
+ "skip_when_already_exists",
+ )
+
+/**
+ * Writes a task declared from a [TriggerDagRun] as Python writes a
+ * `TriggerDagRunOperator`, so the Airflow UI and the Dag dependency graph
+ * treat it as one.
+ *
+ * Python writes an operator's template fields and leaves its other parameters
+ * out, because a Python Dag file holds them. A Java Dag has no such file, so
+ * every setting the author made is written under the name Python's parameter
+ * carries. The runtime runs the task from the Dag in its own bundle and reads
+ * none of this back.
+ */
+private fun writeTriggerDagRun(
+ data: MutableMap<String, Any?>,
+ trigger: TriggerDagRun,
+) {
+ data["task_type"] = TRIGGER_TASK_TYPE
+ data["_task_module"] = TRIGGER_TASK_MODULE
+ data["ui_color"] = TRIGGER_UI_COLOR
+ data["template_fields"] = TRIGGER_TEMPLATE_FIELDS
+ data["template_fields_renderers"] = mapOf("conf" to "py")
+ data["_operator_extra_links"] = mapOf("Triggered DAG" to
"_link_TriggerDagRunLink")
+ data["trigger_dag_id"] = trigger.dagId
+
+ val settings = trigger.settings
+ // Python's logical_date defaults to NOTSET, which lets the operator pick the
+ // trigger time, and writes that sentinel as its name. A template field holds
+ // a datetime as str(datetime), not as a {"__type": "datetime"} object.
+ data["logical_date"] = settings["logical_date"]?.let { pythonStr(it) } ?:
"NOTSET"
+ // Python writes these two template fields whatever they hold.
+ data["wait_for_completion"] = settings["wait_for_completion"] ?: false
+ data["skip_when_already_exists"] = settings["skip_when_already_exists"] ?:
false
+ settings["trigger_run_id"]?.let { data["trigger_run_id"] = it }
+ settings["conf"]?.let { data["conf"] = plainJson(it, "conf", trigger.dagId) }
+ // A field that is not a template field is read back with
+ // BaseSerialization.deserialize, which returns a datetime only for the
+ // type-encoded form.
+ settings["run_after"]?.let { data["run_after"] = serializeValue(it) }
+ settings["reset_dag_run"]?.takeIf { it == true }?.let {
data["reset_dag_run"] = true }
+ settings["poke_interval"]?.let { data["poke_interval"] = (it as
Duration).seconds.toInt() }
+ settings["allowed_states"]?.let { data["allowed_states"] = it }
+ settings["failed_states"]?.let { data["failed_states"] = it }
+ settings["fail_when_dag_is_paused"]?.takeIf { it == true }?.let {
data["fail_when_dag_is_paused"] = true }
+ settings["note"]?.let { data["note"] = it }
+ settings["deferrable"]?.let { data["deferrable"] = it }
+}
+
+/** `str(datetime)` in UTC, which is how Python writes a datetime in a
template field. */
+private fun pythonStr(value: Any): String {
Review Comment:
This and `pythonIsoformat` in `TriggerRunner.kt` differ only in the
separator. Could `pythonIsoformat` take a `sep` and replace this one, so a fix
to the micros or offset handling can't land in only one of them?
--
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]
