From 517bea55547643ac9f6d9b467dfc8decfc93b4c2 Mon Sep 17 00:00:00 2001 From: Andrew Chang Date: Tue, 25 Aug 2026 15:04:07 +0300 Subject: [PATCH] Warn on suspicious Dag and task IDs when a Java SDK Bundle is built The Airflow server validates IDs authoritatively and the allowed character set has changed before, so the SDK warns instead of failing. Per review, the check lives in a dedicated module and runs where Dags are assembled into a Bundle, which both the interface based and annotation based syntaxes reach; the annotation processor only ever saw the annotation side. --- .../kotlin/org/apache/airflow/sdk/Bundle.kt | 9 ++ .../kotlin/org/apache/airflow/sdk/DagDef.kt | 9 +- .../org/apache/airflow/sdk/IdValidation.kt | 71 ++++++++++ .../org/apache/airflow/sdk/BundleTest.kt | 31 +++++ .../apache/airflow/sdk/IdValidationTest.kt | 127 ++++++++++++++++++ 5 files changed, 244 insertions(+), 3 deletions(-) create mode 100644 java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/IdValidation.kt create mode 100644 java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/IdValidationTest.kt diff --git a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Bundle.kt b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Bundle.kt index 6cd549270a8a4..5445db5c5ee93 100644 --- a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Bundle.kt +++ b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Bundle.kt @@ -25,6 +25,11 @@ package org.apache.airflow.sdk * Build a [Bundle] by implementing [BundleBuilder], then pass it to * [Server.serve] to start accepting task-execution requests. * + * Dag and task IDs the Airflow server would reject (longer than 250 + * characters, or containing anything other than letters, digits, dashes, + * dots, and underscores) produce a best-effort warning in the task logs; + * the server remains the authoritative validator. + * * @property dags All registered Dags keyed by [DagDef.id]. * @throws IllegalArgumentException if any two Dags share the same ID. */ @@ -32,6 +37,10 @@ class Bundle( dags: Iterable, ) { internal val dags: Map = dags.associateByDagId() + + init { + IdValidation.warnOnSuspiciousIds(this.dags.values) + } } private fun Iterable.associateByDagId(): Map { 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 e3ba700a0fa1c..d93182b76ccd8 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 @@ -36,13 +36,16 @@ import kotlin.Throws * .addTask("load", Load.class); * ``` * - * @param id Dag identifier. Must contain only ASCII alphanumeric characters, - * dashes, dots, or underscores; must be unique within a [Bundle]. + * Dag and task IDs are validated authoritatively by the Airflow server. This + * class accepts any ID; IDs the server would reject produce a best-effort + * warning when the Dags are assembled into a [Bundle]. + * + * @param id Dag identifier; must be unique within a [Bundle]. * * @see Builder.Dag */ class DagDef( - val id: String, // TODO: charset check? + val id: String, ) { internal val tasks = linkedMapOf() diff --git a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/IdValidation.kt b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/IdValidation.kt new file mode 100644 index 0000000000000..d9f573c631781 --- /dev/null +++ b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/IdValidation.kt @@ -0,0 +1,71 @@ +/* + * 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.Logger + +/** + * Best-effort checks of Dag and task IDs against the rules the Airflow server + * enforces (`airflow.utils.helpers.validate_key`). Warn-only: the server + * validates authoritatively, and checks like the `..` one depend on server + * configuration this process cannot see. + */ +internal object IdValidation { + private const val MAX_ID_LENGTH = 250 + private val ID_REGEX = Regex("""^[\p{L}\p{N}_.-]+$""") + + private val logger = Logger(IdValidation::class) + + fun warnOnSuspiciousIds(dags: Iterable) { + for (dag in dags) { + warnOnSuspiciousId("Dag id", dag.id, mapOf("dag_id" to dag.id)) + for (taskId in dag.tasks.keys) { + warnOnSuspiciousId("Task id", taskId, mapOf("dag_id" to dag.id, "task_id" to taskId)) + } + } + } + + private fun warnOnSuspiciousId( + label: String, + id: String, + arguments: Map, + ) { + val length = id.codePointCount(0, id.length) + if (length > MAX_ID_LENGTH) { + logger.warning( + "$label is longer than $MAX_ID_LENGTH characters; the Airflow server will reject it", + arguments + ("length" to length), + ) + } + if (!ID_REGEX.matches(id)) { + logger.warning( + "$label must be made of alphanumeric characters, dashes, dots, and underscores; " + + "the Airflow server will reject it", + arguments, + ) + } else if (id.contains("..")) { + logger.warning( + "$label contains '..'; the Airflow server will reject it " + + "unless [core] allow_double_dot_in_ids is enabled", + arguments, + ) + } + } +} diff --git a/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/BundleTest.kt b/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/BundleTest.kt index 57050754e887e..5d23e95c07631 100644 --- a/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/BundleTest.kt +++ b/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/BundleTest.kt @@ -19,11 +19,26 @@ package org.apache.airflow.sdk +import org.apache.airflow.sdk.execution.Level +import org.apache.airflow.sdk.execution.LogCapture import org.junit.jupiter.api.Assertions +import org.junit.jupiter.api.BeforeEach import org.junit.jupiter.api.DisplayName import org.junit.jupiter.api.Test internal class BundleTest { + private class NoOp : Task { + override fun execute( + context: Context, + client: Client, + ) = Unit + } + + @BeforeEach + fun setUp() { + LogCapture.drain() + } + @Test @DisplayName("Should index dags by dagId") fun shouldIndexDagsByDagId() { @@ -32,6 +47,22 @@ internal class BundleTest { val bundle = Bundle(listOf(dag)) Assertions.assertEquals(mapOf("dag" to dag), bundle.dags) + Assertions.assertEquals(emptyList(), LogCapture.drain()) + } + + @Test + @DisplayName("Should warn on dag and task ids the server would reject") + fun shouldWarnOnSuspiciousIds() { + Bundle(listOf(DagDef("bad dag").addTask("bad task", NoOp::class.java))) + + val warnings = LogCapture.drain() + Assertions.assertEquals(2, warnings.size) + Assertions.assertEquals(setOf(Level.WARNING), warnings.map { it.level }.toSet()) + Assertions.assertEquals(mapOf("dag_id" to "bad dag"), warnings[0].arguments) + Assertions.assertEquals( + mapOf("dag_id" to "bad dag", "task_id" to "bad task"), + warnings[1].arguments, + ) } @Test diff --git a/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/IdValidationTest.kt b/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/IdValidationTest.kt new file mode 100644 index 0000000000000..845562579af70 --- /dev/null +++ b/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/IdValidationTest.kt @@ -0,0 +1,127 @@ +/* + * 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.CapturedLogMessage +import org.apache.airflow.sdk.execution.Level +import org.apache.airflow.sdk.execution.LogCapture +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.BeforeEach +import org.junit.jupiter.api.DisplayName +import org.junit.jupiter.api.Test + +private const val LOGGER_NAME = "org.apache.airflow.sdk.IdValidation" + +private fun dagTooLongWarning( + id: String, + length: Int, +) = CapturedLogMessage( + Level.WARNING, + LOGGER_NAME, + "Dag id is longer than 250 characters; the Airflow server will reject it", + mapOf("dag_id" to id, "length" to length), +) + +private fun dagCharsetWarning(id: String) = + CapturedLogMessage( + Level.WARNING, + LOGGER_NAME, + "Dag id must be made of alphanumeric characters, dashes, dots, and underscores; " + + "the Airflow server will reject it", + mapOf("dag_id" to id), + ) + +private fun dagDoubleDotWarning(id: String) = + CapturedLogMessage( + Level.WARNING, + LOGGER_NAME, + "Dag id contains '..'; the Airflow server will reject it " + + "unless [core] allow_double_dot_in_ids is enabled", + mapOf("dag_id" to id), + ) + +internal class IdValidationTest { + private class NoOp : Task { + override fun execute( + context: Context, + client: Client, + ) = Unit + } + + @BeforeEach + fun setUp() { + LogCapture.drain() + } + + @Test + @DisplayName("dag id warnings — exact events across every branch") + fun dagIdWarnings() { + val astral = "𠀀" + val tooLongAndInvalid = "a".repeat(250) + " b" + val cases: List>> = + listOf( + "simple" to emptyList(), + "with-dash" to emptyList(), + "with.dot" to emptyList(), + "with_underscore" to emptyList(), + "0numeric" to emptyList(), + "café_dag" to emptyList(), + "任務" to emptyList(), + "a".repeat(250) to emptyList(), + astral.repeat(250) to emptyList(), + "a".repeat(251) to listOf(dagTooLongWarning("a".repeat(251), 251)), + "任".repeat(251) to listOf(dagTooLongWarning("任".repeat(251), 251)), + astral.repeat(251) to listOf(dagTooLongWarning(astral.repeat(251), 251)), + "with space" to listOf(dagCharsetWarning("with space")), + "with/slash" to listOf(dagCharsetWarning("with/slash")), + "with:colon" to listOf(dagCharsetWarning("with:colon")), + "with\ttab" to listOf(dagCharsetWarning("with\ttab")), + "a..b c" to listOf(dagCharsetWarning("a..b c")), + "a..b" to listOf(dagDoubleDotWarning("a..b")), + tooLongAndInvalid to + listOf(dagTooLongWarning(tooLongAndInvalid, 252), dagCharsetWarning(tooLongAndInvalid)), + ) + cases.forEach { (id, expected) -> + IdValidation.warnOnSuspiciousIds(listOf(DagDef(id))) + assertEquals(expected, LogCapture.drain(), "id=$id") + } + } + + @Test + @DisplayName("a task warning carries its dag id") + fun taskWarningCarriesDagId() { + val dag = DagDef("my_dag").addTask("bad task", NoOp::class.java) + + IdValidation.warnOnSuspiciousIds(listOf(dag)) + + assertEquals( + listOf( + CapturedLogMessage( + Level.WARNING, + LOGGER_NAME, + "Task id must be made of alphanumeric characters, dashes, dots, and underscores; " + + "the Airflow server will reject it", + mapOf("dag_id" to "my_dag", "task_id" to "bad task"), + ), + ), + LogCapture.drain(), + ) + } +}