jason810496 · Oct 9, 2026 · Oct 2, 2026 · Oct 2, 2026 · Oct 6, 2026 · Oct 7, 2026
diff --git a/airflow-core/adr/lang-sdk/0015-per-dag-source-in-bundle-artifact.md b/airflow-core/adr/lang-sdk/0015-per-dag-source-in-bundle-artifact.md
@@ -94,6 +94,8 @@ TypeScript is the reference ([#73723](https://github.com/apache/airflow/pull/737
 packers adopt the same shape — per-Dag source resolution, de-duplication by path, and an
 always-embedded entrypoint fallback — so a bundle reader treats every language the same way and the
 Code tab behaves identically regardless of which SDK produced the artifact.
+A JVM SDK has no module body, so it resolves a Dag by the class that declared it, and a Dag built by a
+factory maps to the factory's class.
 
 Go has no module evaluation, so the Go SDK records the file that calls `airflow.Dag`. A Dag built in
 a factory function maps to the factory's file, not to the file that calls the factory.

diff --git a/airflow-core/docs/authoring-and-scheduling/language-sdks/java.rst b/airflow-core/docs/authoring-and-scheduling/language-sdks/java.rst
@@ -1144,6 +1144,10 @@ into the bundle JAR, so the Airflow UI can show the source of a native Java Dag
 :ref:`java-sdk/native-dag-parsing`). To find those classes, the plugin runs ``mainClass`` once at build
 time. If that run fails, the build logs a warning and packs only the ``mainClass`` source.
 
+A Dag is recorded against the class that constructed it. A Dag built by a factory therefore maps to the
+factory's class, and one built by a factory in a dependency JAR has no source in the project and falls
+back to the ``mainClass`` source.
+
 .. note::
 
   You only need the ``annotationProcessor`` entry if you use the annotation-based API. It is not needed for

@@ -91,6 +91,10 @@ Now `cd example` into the example project, and
   ../gradlew bundle
   ```
 
+  The bundle JAR also carries the source file of each Dag declared in Java
+  (and of the entrypoint) under `META-INF/airflow/`, so Airflow can show a
+  Dag's source.
+
 * Put the [DAG with stub tasks](./example/src/resources/dags) to somewhere Airflow can find.
 
 * Ensure the `java` command is available in the same environments the Airflow

@@ -25,6 +25,12 @@ plugins {
 
 dependencies {
     implementation("com.gradleup.shadow:shadow-gradle-plugin:9.1.0") // Last supporting Java 11.
+
+    testImplementation(kotlin("test"))
+}
+
+tasks.withType<Test> {
+    useJUnitPlatform()
 }
 
 gradlePlugin {

@@ -22,13 +22,16 @@ package org.apache.airflow.sdk.plugin
 import com.github.jengelman.gradle.plugins.shadow.tasks.ShadowJar
 import org.gradle.api.Plugin
 import org.gradle.api.Project
+import org.gradle.api.plugins.JavaPluginExtension
 import org.gradle.api.provider.Property
 import org.gradle.api.tasks.Copy
 import org.gradle.api.tasks.Input
 import org.gradle.api.tasks.SourceSetContainer
 import org.gradle.api.tasks.bundling.Jar
+import org.gradle.jvm.toolchain.JavaToolchainService
 import java.lang.reflect.Modifier
 import java.net.URLClassLoader
+import java.time.Duration
 import java.util.jar.JarFile
 import kotlin.jvm.java
 
@@ -39,6 +42,7 @@ import kotlin.jvm.java
  * airflowBundle {
  *     mainClass = "com.example.ExampleBundleBuilder"
  *     // fatJar = false  // opt out of shadow JAR creation
+ *     // describeTimeout = Duration.ofSeconds(60)
  * }
  * ```
  */
@@ -61,6 +65,10 @@ abstract class AirflowBundleExtension {
    */
   @get:Input
   abstract val fatJar: Property<Boolean>
+
+  /** How long the describe run may take before the build gives up and packs only the entrypoint's source. */
+  @get:Input
+  abstract val describeTimeout: Property<Duration>
 }
 
 /**
@@ -93,6 +101,13 @@ abstract class AirflowBundleExtension {
  * identify which version of the Supervisor Schema it should use to communicate
  * with the built JAR.
  *
+ * When `mainClass` is set, the bundle JAR also carries the source file of each
+ * Dag declared in Java and of the entrypoint, so Airflow can show a Dag's source.
+ * They are collected by the `packDagSources` task, which runs `mainClass` once in
+ * a describe mode, and are listed in `META-INF/airflow/sources.json`, named by the
+ * `Airflow-Java-SDK-Sources` manifest attribute. If that run fails, a warning is
+ * logged and only the entrypoint's source is packed.
+ *
  * If `fatJar` is explicitly set to `false`, the `bundle` task builds a bare JAR
  * containing only the Dag bundle, and collect all dependency JARs into the target
  * directory instead. In this mode, `Airflow-Supervisor-Schema-Version` lives in
@@ -105,8 +120,31 @@ class AirflowSdkPlugin : Plugin<Project> {
 
     val ext = project.extensions.create("airflowBundle", AirflowBundleExtension::class.java)
     ext.fatJar.convention(true)
+    ext.describeTimeout.convention(Duration.ofSeconds(30))
 
     project.afterEvaluate {
+      val main = project.extensions.getByType(SourceSetContainer::class.java).getByName("main")
+
+      val packTask =
+        project.tasks.register("packDagSources", PackDagSources::class.java) { task ->
+          task.group = "build"
+          task.description = "Collects the source file of each Dag and the entrypoint to pack into the bundle JAR."
+          task.dependsOn(project.tasks.named("classes"))
+          task.onlyIf { ext.mainClass.isPresent }
+          task.mainClass.set(ext.mainClass)
+          task.describeTimeout.set(ext.describeTimeout)
+          task.classesDirs.from(main.output.classesDirs)
+          task.runtimeClasspath.from(main.runtimeClasspath)
+          task.sourceDirs.from(main.allSource.srcDirs)
+          task.launcher.convention(
+            project.extensions
+              .getByType(JavaToolchainService::class.java)
+              .launcherFor(project.extensions.getByType(JavaPluginExtension::class.java).toolchain),
+          )
+          task.describeFile.set(project.layout.buildDirectory.file("airflow/describe-sources.json"))
+          task.sourcesDir.set(project.layout.buildDirectory.dir("airflow/sources"))
+        }
+
       project.tasks.withType(Jar::class.java).configureEach { task ->
         task.doFirst {
           ext.mainClass.orNull?.let { className ->
@@ -117,11 +155,7 @@ class AirflowSdkPlugin : Plugin<Project> {
 
       val classFiles =
         project.objects.fileCollection().from(
-          project.extensions
-            .getByType(SourceSetContainer::class.java)
-            .getByName("main")
-            .output
-            .classesDirs,
+          main.output.classesDirs,
           project.configurations.getByName("runtimeClasspath"),
         )
 
@@ -162,8 +196,17 @@ class AirflowSdkPlugin : Plugin<Project> {
           }
         }
 
+      fun packSourcesInto(jarTask: String) {
+        if (!ext.mainClass.isPresent) return
+        project.tasks.named(jarTask, Jar::class.java).configure { task ->
+          task.from(packTask.flatMap { it.sourcesDir })
+          task.manifest.attributes(mapOf(SOURCES_MANIFEST_ATTRIBUTE to SOURCES_JSON_PATH))
+        }
+      }
+
       if (ext.fatJar.get()) {
         project.plugins.apply("com.gradleup.shadow")
+        packSourcesInto("shadowJar")
 
         val schemaVersionProvider =
           project.providers.provider {
@@ -205,6 +248,8 @@ class AirflowSdkPlugin : Plugin<Project> {
           task.into(project.layout.buildDirectory.dir("bundle"))
         }
       } else {
+        packSourcesInto("jar")
+
         // bundle copies the thin JAR and all runtime dependency JARs into
         // build/bundle/, mirroring what installDist puts in lib/.
         project.tasks.register("bundle", Copy::class.java) { task ->

@@ -0,0 +1,85 @@
+/*
+ * 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.plugin
+
+import java.io.ByteArrayInputStream
+import java.io.DataInputStream
+import java.io.IOException
+
+private const val CLASS_MAGIC = 0xCAFEBABE.toInt()
+
+/**
+ * Reads the `SourceFile` attribute of a class file, which javac, kotlinc and scalac all
+ * write, or returns `null` if the class has none or the bytes are not a class file.
+ */
+internal fun readSourceFile(classFile: ByteArray): String? =
+  try {
+    DataInputStream(ByteArrayInputStream(classFile)).use(::readSourceFile)
+  } catch (_: IOException) {
+    null
+  }
+
+private fun readSourceFile(input: DataInputStream): String? {
+  if (input.readInt() != CLASS_MAGIC) return null
+  input.skipBytes(4) // minor and major version
+
+  val utf8 = arrayOfNulls<String>(input.readUnsignedShort())
+  var index = 1
+  while (index < utf8.size) {
+    when (val tag = input.readUnsignedByte()) {
+      1 -> utf8[index] = input.readUTF()
+      3, 4, 9, 10, 11, 12, 17, 18 -> input.skipBytes(4)
+      5, 6 -> {
+        input.skipBytes(8)
+        index++ // long and double take two slots
+      }
+      7, 8, 16, 19, 20 -> input.skipBytes(2)
+      15 -> input.skipBytes(3)
+      else -> throw IOException("Unknown constant pool tag $tag")
+    }
+    index++
+  }
+
+  input.skipBytes(6) // access flags, this class, super class
+  input.skipBytes(2 * input.readUnsignedShort()) // interfaces
+  repeat(2) { skipMembers(input) } // fields, then methods
+
+  repeat(input.readUnsignedShort()) {
+    val name = utf8.getOrNull(input.readUnsignedShort())
+    val length = input.readInt()
+    if (name == "SourceFile" && length == 2) return utf8.getOrNull(input.readUnsignedShort())
+    input.skipBytes(length)
+  }
+  return null
+}
+
+private fun skipMembers(input: DataInputStream) {
+  repeat(input.readUnsignedShort()) {
+    input.skipBytes(6) // access flags, name, descriptor
+    skipAttributes(input)
+  }
+}
+
+private fun skipAttributes(input: DataInputStream) {
+  repeat(input.readUnsignedShort()) {
+    input.skipBytes(2)
+    input.skipBytes(input.readInt())
+  }
+}