jason810496 commented on code in PR #74440:
URL: https://github.com/apache/airflow/pull/74440#discussion_r4220439677
##########
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:
Done in a8d8d11c734: both examples now trigger a new
java_native_target_example that triggers nothing, so the loop is gone.
##########
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:
Done in 4f263fcfb7b: a trigger task is rejected where it is named as a case,
and the message says to put the trigger downstream of an ordinary task the
switch chooses.
##########
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:
Done in 5d013803137: the settings are snapshotted when the task is declared,
with tests for a reused builder and for a config call after registration.
##########
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:
Done in e638c382553, without waiting on #74417: logical_date is a template
field, so Python writes whatever the author gave it, and the fixture now spells
it the way Python spells a datetime. An SDK that takes a date-time and one that
passes the field through as a string then write the same text, so the Java, Go
and TypeScript conformance hooks all pass on this PR as it stands, and #74417
can still move the fixture to a typed datetime afterwards.
##########
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:
Done in c4a82f66f43: the fake asserts it polls the run it started, and the
three one-sided branches are covered.
##########
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:
Done in 192db48d10c, with a runTask test for the pushed run id and one where
the client throws and the task ends failed.
##########
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:
Done in ddd5fd71fbd, though not with typed setters: config now runs through
the same Field table and shared validator that DagDef.config and TaskDef.config
use, so there is one spelling and one message shape across the SDK, and a wrong
value still fails when the Dag is built, exactly as a mistyped retries value
does.
##########
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:
Done in 375bef298ff: it names the TypeScript runtime only.
##########
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:
Done in e13fd59e43b, for all three.
##########
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:
Done in fb0b40b541c: pythonIsoformat takes a separator and pythonStr is gone.
##########
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:
Done in 1cd51bf192b, with the same unchecked-exception check depends gets
and a BuilderTest for it.
##########
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:
Done in ddd5fd71fbd: the JSON check runs in config now, so it fails where
the KDoc says and no longer names the target Dag as a task.
##########
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:
Done in ddd5fd71fbd: an empty trigger run id or note is rejected in config,
the way an empty dag id already is, so the fallback stays correct.
--
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]
