Warn on suspicious Dag and task IDs at Java SDK build time

java-sdk/processor/src/main/kotlin/org/apache/airflow/sdk/BuilderProcessor.kt

@@ -34,6 +34,7 @@ import javax.annotation.processing.RoundEnvironment
 import javax.annotation.processing.SupportedAnnotationTypes
 import javax.annotation.processing.SupportedSourceVersion
 import javax.lang.model.SourceVersion
+import javax.lang.model.element.Element
 import javax.lang.model.element.ExecutableElement
 import javax.lang.model.element.Modifier
 import javax.lang.model.element.TypeElement
@@ -92,6 +93,8 @@ class BuilderProcessor : AbstractProcessor() {
 
   private fun buildDag(el: TypeElement): TypeSpec {
     val ann = el.getAnnotation(Builder.Dag::class.java)!!
+    val dagId = ann.id.ifBlank { el.simpleName.toString() }
+    processingEnv.warnOnSuspiciousId(el, "Dag id \"$dagId\"", dagId)
 
     val builderClass =
       TypeSpec
@@ -103,21 +106,23 @@ class BuilderProcessor : AbstractProcessor() {
         .methodBuilder("build")
         .addModifiers(Modifier.PUBLIC, Modifier.STATIC)
         .returns(ClassName.get(Dag::class.java))
-        .addStatement($$"var dag = new $T($S)", ClassName.get(Dag::class.java), ann.id.ifBlank { el.simpleName })
+        .addStatement($$"var dag = new $T($S)", ClassName.get(Dag::class.java), dagId)
 
     for (inner in el.enclosedElements) {
       if (inner !is ExecutableElement) continue
       if (inner.isVarArgs) throw IllegalArgumentException("Cannot create task from vararg function ${inner.simpleName}")
 
-      val ann = inner.getAnnotation(Builder.Task::class.java) ?: continue
+      val taskAnn = inner.getAnnotation(Builder.Task::class.java) ?: continue
+      val taskId = taskAnn.id.ifBlank { inner.simpleName.toString() }
+      processingEnv.warnOnSuspiciousId(inner, "Task id \"$taskId\" in dag \"$dagId\"", taskId)
       val innerName = inner.simpleName.toString().replaceFirstChar(Char::uppercase)
 
       val task = buildTask(innerName, inner, el)
       builderClass.addType(task.spec)
 
       buildMethod.addStatement(
         $$"dag.addTask($S, $L.class)",
-        ann.id.ifBlank { inner.simpleName },
+        taskId,
         innerName,
       )
     }
@@ -194,6 +199,37 @@ private fun ProcessingEnvironment.isType(
   c: ClassName,
 ): Boolean = typeUtils.isSameType(t, elementUtils.getTypeElement(c.canonicalName()).asType())
 
+private const val MAX_ID_LENGTH = 250
+private val ID_REGEX = Regex("""^[\p{L}\p{N}_.-]+$""")
+
+private fun ProcessingEnvironment.warnOnSuspiciousId(
+  element: Element,
+  label: String,
+  id: String,
+) {
+  val length = id.codePointCount(0, id.length)
+  if (length > MAX_ID_LENGTH) {
+    messager.printMessage(
+      Diagnostic.Kind.WARNING,
+      "$label is longer than $MAX_ID_LENGTH characters ($length); the Airflow server will reject it",
+      element,
+    )
+  }
+  if (!ID_REGEX.matches(id)) {
+    messager.printMessage(
+      Diagnostic.Kind.WARNING,
+      "$label must be made of alphanumeric characters, dashes, dots, and underscores; the Airflow server will reject it",
+      element,
+    )
+  } else if (id.contains("..")) {
+    messager.printMessage(
+      Diagnostic.Kind.WARNING,
+      "$label contains '..'; the Airflow server will reject it unless [core] allow_double_dot_in_ids is enabled",
+      element,
+    )
+  }
+}
+
 private data class RequiredXCom(
   val paramType: TypeMirror,
   val paramName: String,

java-sdk/processor/src/test/kotlin/org/apache/airflow/sdk/BuilderTest.kt

@@ -19,10 +19,12 @@
 
 package org.apache.airflow.sdk
 
+import com.google.testing.compile.Compilation
 import com.google.testing.compile.CompilationSubject.assertThat
 import com.google.testing.compile.Compiler
 import com.google.testing.compile.JavaFileObjectSubject
 import com.google.testing.compile.JavaFileObjects
+import org.junit.jupiter.api.Assertions.assertEquals
 import org.junit.jupiter.api.DisplayName
 import org.junit.jupiter.api.Test
 
@@ -38,6 +40,21 @@ private fun JavaFileObjectSubject.hasSourceEquivalentTo(
   JavaFileObjects.forSourceString(qual, source),
 )
 
+private fun Compilation.serverRejectWarnings(): List<String> =
+  warnings().map { it.getMessage(null) }.filter { "the Airflow server will reject it" in it }
+
+private fun dagCharsetWarning(id: String) =
+  "Dag id \"$id\" must be made of alphanumeric characters, dashes, dots, and underscores; " +
+    "the Airflow server will reject it"
+
+private fun dagTooLongWarning(
+  id: String,
+  length: Int,
+) = "Dag id \"$id\" is longer than 250 characters ($length); the Airflow server will reject it"
+
+private fun dagDoubleDotWarning(id: String) =
+  "Dag id \"$id\" contains '..'; the Airflow server will reject it unless [core] allow_double_dot_in_ids is enabled"
+
 class BuilderTest {
   @Test
   @DisplayName("generate builder for dag class")
@@ -364,4 +381,82 @@ class BuilderTest {
       "Cannot create task from vararg function t1",
     )
   }
+
+  @Test
+  @DisplayName("dag id warnings — exact messages across every branch")
+  fun dagIdWarnings() {
+    val astral = "𠀀"
+    val tooLongAndInvalid = "a".repeat(250) + " b"
+    val cases: List<Pair<String, List<String>>> =
+      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) ->
+      val compilation =
+        compile(
+          """
+        package org.apache.airflow.example;
+        import org.apache.airflow.sdk.Builder;
+        @Builder.Dag(id = "$id") public class TestExample {}
+        """,
+        )
+      assertThat(compilation).succeeded()
+      assertThat(compilation).generatedSourceFile("org.apache.airflow.example.TestExampleBuilder")
+      assertEquals(expected, compilation.serverRejectWarnings(), "id=$id")
+    }
+  }
+
+  @Test
+  @DisplayName("a task warning names its dag")
+  fun taskWarningNamesItsDag() {
+    val compilation =
+      compile(
+        """
+        package org.apache.airflow.example;
+        import org.apache.airflow.sdk.Builder;
+        @Builder.Dag(id = "my_dag")
+        public class TestExample { @Builder.Task(id = "bad task") public void t1() {} }
+        """,
+      )
+    assertThat(compilation).succeeded()
+    assertEquals(
+      listOf(
+        "Task id \"bad task\" in dag \"my_dag\" must be made of alphanumeric characters, dashes, dots, and underscores; the Airflow server will reject it",
+      ),
+      compilation.serverRejectWarnings(),
+    )
+  }
+
+  @Test
+  @DisplayName("a blank id falls back to the element name and does not warn")
+  fun blankIdFallsBackToElementName() {
+    val compilation =
+      compile(
+        """
+        package org.apache.airflow.example;
+        import org.apache.airflow.sdk.Builder;
+        @Builder.Dag public class TestExample { @Builder.Task public void t1() {} }
+        """,
+      )
+    assertThat(compilation).succeeded()
+    assertEquals(emptyList<String>(), compilation.serverRejectWarnings())
+  }
 }

java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Dag.kt

@@ -30,13 +30,19 @@ import kotlin.Throws
  * where the annotation processor generates the wiring for you. Only use this
  * class directly if you need to do low-level plumbing.
  *
- * @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. Ids
+ * declared through [Builder.Dag]/[Builder.Task] that the server would reject
+ * (longer than 250 characters, or containing anything other than letters,
+ * digits, dashes, dots, and underscores) produce a best-effort build-time
+ * warning from the annotation processor; this low-level API does not check
+ * them.
+ *
+ * @param id Dag identifier; must be unique within a [Bundle].
  *
  * @see Builder.Dag
  */
 class Dag(
-  val id: String, // TODO: charset check?
+  val id: String,
 ) {
   internal var tasks = mutableMapOf<String, Class<out Task>>()