You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "jason810496 (via GitHub)" <gi...@apache.org> on 2026/10/02 15:27:12 UTC
[PR] Java SDK: Pack each native Dag's source file into the bundle JAR [airflow]
jason810496 opened a new pull request, #74096:
URL: https://github.com/apache/airflow/pull/74096
- related: #71929, #73723
Builds on #71190 ([new commits only](https://github.com/apache/airflow/compare/eb662eac6fc...jason/java-sdk/pack-dag-sources); the commits below them are from #71189 and #71190). Merge after it.
## Why
Airflow can parse native Java Dags (#74036), and its Code view needs the Dag's source. A bundle JAR carried at most one hand-picked file, so a Dag declared in another class had no source of its own. This follows #73723 for TypeScript: pack each Dag's source file once, map every Dag to its file, and always keep the entrypoint as a fallback.
## What changes
- The Gradle plugin's new `packDagSources` task packs the source of `mainClass` and of each class that declares a Dag in Java into the bundle JAR, and indexes them in `sources.json`. Nothing to configure.
- To learn which class declares each Dag, the task runs `mainClass` once at build time. If that run fails, the build logs a warning and packs only the `mainClass` source.
- A Dag built by a `@Builder.Dag` class maps to that class, not to the generated builder. Task handlers for a Python Dag are not listed.
- `.java`, `.kt` and `.scala` sources all work.
The bundle JAR now holds:
```
META-INF/MANIFEST.MF
Main-Class: org.apache.airflow.example.ExampleBundleBuilder
Airflow-Java-SDK-Sources: META-INF/airflow/sources.json
META-INF/airflow/sources.json
META-INF/airflow/sources/org/apache/airflow/example/ExampleBundleBuilder.java
META-INF/airflow/sources/org/apache/airflow/example/nativedag/AnnotationExample.java
META-INF/airflow/sources/org/apache/airflow/example/nativedag/InterfaceExample.java
```
```json
{
"entrypoint_path": "org/apache/airflow/example/ExampleBundleBuilder.java",
"dag_source_paths": {
"java_native_annotation_example": "org/apache/airflow/example/nativedag/AnnotationExample.java",
"java_native_interface_example": "org/apache/airflow/example/nativedag/InterfaceExample.java"
}
}
```
## How to test
```bash
cd java-sdk && ./gradlew :sdk:test :processor:test :plugin:test
```
---
##### Was generative AI tooling used to co-author this PR?
- [x] Yes, with help of Claude Code Opus 5.5 following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Java SDK: Pack each native Dag's source file into the bundle JAR [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #74096:
URL: https://github.com/apache/airflow/pull/74096#discussion_r4214940416
##########
java-sdk/plugin/src/main/kotlin/org/apache/airflow/sdk/plugin/PackDagSources.kt:
##########
@@ -0,0 +1,240 @@
+/*
+ * 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 groovy.json.JsonOutput
+import groovy.json.JsonSlurper
+import org.gradle.api.DefaultTask
+import org.gradle.api.file.ConfigurableFileCollection
+import org.gradle.api.file.DirectoryProperty
+import org.gradle.api.file.RegularFileProperty
+import org.gradle.api.provider.Property
+import org.gradle.api.tasks.CacheableTask
+import org.gradle.api.tasks.Classpath
+import org.gradle.api.tasks.Input
+import org.gradle.api.tasks.InputFiles
+import org.gradle.api.tasks.Nested
+import org.gradle.api.tasks.Optional
+import org.gradle.api.tasks.OutputDirectory
+import org.gradle.api.tasks.OutputFile
+import org.gradle.api.tasks.PathSensitive
+import org.gradle.api.tasks.PathSensitivity
+import org.gradle.api.tasks.TaskAction
+import org.gradle.jvm.toolchain.JavaLauncher
+import java.io.File
+import java.util.concurrent.TimeUnit
+
+internal const val SOURCES_MANIFEST_ATTRIBUTE = "Airflow-Java-SDK-Sources"
+internal const val SOURCES_JSON_PATH = "META-INF/airflow/sources.json"
+internal const val SOURCES_DIR_PATH = "META-INF/airflow/sources"
+
+/**
+ * How long the describe run gets before the build gives up on it and packs
+ * the entrypoint's source alone. A `main` that does not reach
+ * `Server.create(args)` never answers `--describe-sources`, and one that
+ * starts serving instead would otherwise hold the build open.
+ */
+private const val DESCRIBE_TIMEOUT_SECONDS = 120L
Review Comment:
Might be a good idea to make this configurable in `airflowBundle { }`?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Java SDK: Pack each native Dag's source file into the bundle JAR [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #74096:
URL: https://github.com/apache/airflow/pull/74096#discussion_r4214910440
##########
java-sdk/plugin/src/main/kotlin/org/apache/airflow/sdk/plugin/AirflowSdkPlugin.kt:
##########
@@ -107,21 +120,45 @@ class AirflowSdkPlugin : Plugin<Project> {
ext.fatJar.convention(true)
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.classesDirs.from(main.output.classesDirs)
+ task.runtimeClasspath.from(main.output, main.runtimeClasspath)
Review Comment:
```suggestion
task.runtimeClasspath.from(main.runtimeClasspath)
```
`main.output` already includes `classesDirs`, which are also on the describe classpath via `classesDirs`.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Java SDK: Pack each native Dag's source file into the bundle JAR [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #74096:
URL: https://github.com/apache/airflow/pull/74096#discussion_r4214848493
##########
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Server.kt:
##########
@@ -83,10 +91,13 @@ class ApiError(
* The process exits when the coordinator closes the connection (normally after
* one task-instance execution).
*/
-class Server(
- private val comm: InetSocketAddress,
- private val logs: InetSocketAddress,
+class Server private constructor(
+ private val comm: InetSocketAddress?,
+ private val logs: InetSocketAddress?,
+ private val describeSources: File?,
) {
Review Comment:
Instead of making this class do two different modes, I’d just create a separate class. We can use polymorphism to abstract away the interface.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Java SDK: Pack each native Dag's source file into the bundle JAR [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #74096:
URL: https://github.com/apache/airflow/pull/74096#discussion_r4214848493
##########
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Server.kt:
##########
@@ -83,10 +91,13 @@ class ApiError(
* The process exits when the coordinator closes the connection (normally after
* one task-instance execution).
*/
-class Server(
- private val comm: InetSocketAddress,
- private val logs: InetSocketAddress,
+class Server private constructor(
+ private val comm: InetSocketAddress?,
+ private val logs: InetSocketAddress?,
+ private val describeSources: File?,
) {
Review Comment:
Instead of making this class do two different modes depending on nullability, I’d just create a separate class. We can use polymorphism to abstract away the interface. This removes many awkward additional checks needed in the code.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Java SDK: Pack each native Dag's source file into the bundle JAR [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #74096:
URL: https://github.com/apache/airflow/pull/74096#discussion_r4206220567
##########
java-sdk/plugin/src/main/kotlin/org/apache/airflow/sdk/plugin/PackDagSources.kt:
##########
@@ -0,0 +1,222 @@
+/*
+ * 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 groovy.json.JsonOutput
+import groovy.json.JsonSlurper
+import org.gradle.api.DefaultTask
+import org.gradle.api.file.ConfigurableFileCollection
+import org.gradle.api.file.DirectoryProperty
+import org.gradle.api.file.RegularFileProperty
+import org.gradle.api.provider.Property
+import org.gradle.api.tasks.CacheableTask
+import org.gradle.api.tasks.Classpath
+import org.gradle.api.tasks.Input
+import org.gradle.api.tasks.InputFiles
+import org.gradle.api.tasks.Nested
+import org.gradle.api.tasks.Optional
+import org.gradle.api.tasks.OutputDirectory
+import org.gradle.api.tasks.OutputFile
+import org.gradle.api.tasks.PathSensitive
+import org.gradle.api.tasks.PathSensitivity
+import org.gradle.api.tasks.TaskAction
+import org.gradle.jvm.toolchain.JavaLauncher
+import java.io.File
+import java.util.concurrent.TimeUnit
+
+internal const val SOURCES_MANIFEST_ATTRIBUTE = "Airflow-Java-SDK-Sources"
+internal const val SOURCES_JSON_PATH = "META-INF/airflow/sources.json"
+internal const val SOURCES_DIR_PATH = "META-INF/airflow/sources"
+
+/**
+ * How long the describe run gets before the build gives up on it and packs
+ * the entrypoint's source alone. A `main` that does not reach
+ * `Server.create(args)` never answers `--describe-sources`, and one that
+ * starts serving instead would otherwise hold the build open.
+ */
+private const val DESCRIBE_TIMEOUT_SECONDS = 120L
+private const val DRAIN_TIMEOUT_MILLIS = 2_000L
+
+/**
+ * Finds the source file of a class by its `SourceFile` attribute: the class
+ * file's package path plus that name, looked up in each source directory.
+ */
+internal class SourceLocator(
+ private val classesDirs: Iterable<File>,
+ private val sourceDirs: Iterable<File>,
+) {
+ /** Path of the source file relative to its source directory, or `null` if it cannot be found. */
+ fun locate(className: String): String? {
+ val packagePath = className.substringBeforeLast('.', "").replace('.', '/')
+ val classFile =
+ classesDirs
+ .map { File(it, className.replace('.', '/') + ".class") }
+ .firstOrNull { it.isFile } ?: return null
+ val sourceFile = readSourceFile(classFile.readBytes())?.takeIf { it.isNotEmpty() && '/' !in it && '\\' !in it }
+ val relative = listOfNotNull(packagePath.ifEmpty { null }, sourceFile ?: return null).joinToString("/")
+ return relative.takeIf { path -> sourceDirs.any { File(it, path).isFile } }
Review Comment:
This only finds a source at `<package path>/<SourceFile>`. Kotlin's coding conventions recommend dropping the common root package from the directory tree (`package com.acme.dags` in `src/main/kotlin/dags/Orders.kt`), and javac doesn't enforce the layout either, so on those projects every lookup misses, the entrypoint included. Nothing is logged, `sources.json` has no `entrypoint_path` while the manifest still points at it, and the Code view says "This JAR embeds no Dag source. Build it with the Airflow Java SDK Gradle plugin", which is wrong for a JAR this plugin built. Could this fall back to a unique match on the `SourceFile` name under the source dirs, and warn when the entrypoint (or a Dag's class, `info` at L155 today) can't be resolved?
##########
java-sdk/plugin/src/main/kotlin/org/apache/airflow/sdk/plugin/PackDagSources.kt:
##########
@@ -0,0 +1,222 @@
+/*
+ * 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 groovy.json.JsonOutput
+import groovy.json.JsonSlurper
+import org.gradle.api.DefaultTask
+import org.gradle.api.file.ConfigurableFileCollection
+import org.gradle.api.file.DirectoryProperty
+import org.gradle.api.file.RegularFileProperty
+import org.gradle.api.provider.Property
+import org.gradle.api.tasks.CacheableTask
+import org.gradle.api.tasks.Classpath
+import org.gradle.api.tasks.Input
+import org.gradle.api.tasks.InputFiles
+import org.gradle.api.tasks.Nested
+import org.gradle.api.tasks.Optional
+import org.gradle.api.tasks.OutputDirectory
+import org.gradle.api.tasks.OutputFile
+import org.gradle.api.tasks.PathSensitive
+import org.gradle.api.tasks.PathSensitivity
+import org.gradle.api.tasks.TaskAction
+import org.gradle.jvm.toolchain.JavaLauncher
+import java.io.File
+import java.util.concurrent.TimeUnit
+
+internal const val SOURCES_MANIFEST_ATTRIBUTE = "Airflow-Java-SDK-Sources"
+internal const val SOURCES_JSON_PATH = "META-INF/airflow/sources.json"
+internal const val SOURCES_DIR_PATH = "META-INF/airflow/sources"
+
+/**
+ * How long the describe run gets before the build gives up on it and packs
+ * the entrypoint's source alone. A `main` that does not reach
+ * `Server.create(args)` never answers `--describe-sources`, and one that
+ * starts serving instead would otherwise hold the build open.
+ */
+private const val DESCRIBE_TIMEOUT_SECONDS = 120L
+private const val DRAIN_TIMEOUT_MILLIS = 2_000L
+
+/**
+ * Finds the source file of a class by its `SourceFile` attribute: the class
+ * file's package path plus that name, looked up in each source directory.
+ */
+internal class SourceLocator(
+ private val classesDirs: Iterable<File>,
+ private val sourceDirs: Iterable<File>,
+) {
+ /** Path of the source file relative to its source directory, or `null` if it cannot be found. */
+ fun locate(className: String): String? {
+ val packagePath = className.substringBeforeLast('.', "").replace('.', '/')
+ val classFile =
+ classesDirs
+ .map { File(it, className.replace('.', '/') + ".class") }
+ .firstOrNull { it.isFile } ?: return null
+ val sourceFile = readSourceFile(classFile.readBytes())?.takeIf { it.isNotEmpty() && '/' !in it && '\\' !in it }
+ val relative = listOfNotNull(packagePath.ifEmpty { null }, sourceFile ?: return null).joinToString("/")
+ return relative.takeIf { path -> sourceDirs.any { File(it, path).isFile } }
+ }
+
+ fun file(relativePath: String): File = sourceDirs.map { File(it, relativePath) }.first { it.isFile }
+}
+
+/**
+ * The `META-INF/airflow/sources.json` body: `entrypoint_path` and each Dag ID
+ * in `dag_source_paths`.
+ *
+ * Every path is relative to the source directory the file was found in, which
+ * is also where it sits under `META-INF/airflow/sources/` in the JAR, so a
+ * reader resolves one by prefixing that directory.
+ */
+internal fun sourcesJson(
+ entrypointPath: String?,
+ dagSourcePaths: Map<String, String>,
+): String =
+ JsonOutput.prettyPrint(
+ JsonOutput.toJson(
+ linkedMapOf<String, Any>().apply {
+ entrypointPath?.let { put("entrypoint_path", it) }
+ put("dag_source_paths", dagSourcePaths)
Review Comment:
Nothing reads `dag_source_paths` yet. On main, `JavaDagImporter.get_source_code` calls `_find_source_entry(zf)` without the `dag_id` it is given ([_dag_importer.py#L79](https://github.com/apache/airflow/blob/3e302fd257a0cfe8deba2de529bdcae6606b83c8/task-sdk/src/airflow/sdk/coordinators/java/_dag_importer.py#L79)), even though `lang_sdk_processor.py` passes `dag.dag_id`. So with this PR every Java Dag still shows the entrypoint in the Code view. Can this PR pass `dag_id` through, with a test that a Dag mapped to `Reports.java` gets `Reports.java` back? Or link the PR that does.
##########
java-sdk/plugin/src/main/kotlin/org/apache/airflow/sdk/plugin/AirflowSdkPlugin.kt:
##########
@@ -107,21 +120,47 @@ class AirflowSdkPlugin : Plugin<Project> {
ext.fatJar.convention(true)
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.classesDirs.from(main.output.classesDirs)
+ task.runtimeClasspath.from(main.output, 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 ->
task.manifest.attributes(mapOf("Main-Class" to className))
}
}
+ // Only the JARs a bundle is assembled from carry the Dag sources. A
+ // sources or javadoc JAR would claim the attribute without being a
+ // bundle, and the Python side picks the bundle JAR out of a directory
+ // by that attribute.
Review Comment:
The Python side doesn't pick JARs by this attribute. `JavaDagImporter.might_contain_dag` and the coordinator select on `Main-Class`, and `Airflow-Java-SDK-Sources` is only read for the Code view. Since the `doFirst` above stamps `Main-Class` on every `Jar` task, a `-sources.jar` is claimed either way. Could the comment just say the payload belongs only in the JARs a bundle is built from?
##########
java-sdk/plugin/src/main/kotlin/org/apache/airflow/sdk/plugin/PackDagSources.kt:
##########
@@ -0,0 +1,222 @@
+/*
+ * 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 groovy.json.JsonOutput
+import groovy.json.JsonSlurper
+import org.gradle.api.DefaultTask
+import org.gradle.api.file.ConfigurableFileCollection
+import org.gradle.api.file.DirectoryProperty
+import org.gradle.api.file.RegularFileProperty
+import org.gradle.api.provider.Property
+import org.gradle.api.tasks.CacheableTask
+import org.gradle.api.tasks.Classpath
+import org.gradle.api.tasks.Input
+import org.gradle.api.tasks.InputFiles
+import org.gradle.api.tasks.Nested
+import org.gradle.api.tasks.Optional
+import org.gradle.api.tasks.OutputDirectory
+import org.gradle.api.tasks.OutputFile
+import org.gradle.api.tasks.PathSensitive
+import org.gradle.api.tasks.PathSensitivity
+import org.gradle.api.tasks.TaskAction
+import org.gradle.jvm.toolchain.JavaLauncher
+import java.io.File
+import java.util.concurrent.TimeUnit
+
+internal const val SOURCES_MANIFEST_ATTRIBUTE = "Airflow-Java-SDK-Sources"
+internal const val SOURCES_JSON_PATH = "META-INF/airflow/sources.json"
+internal const val SOURCES_DIR_PATH = "META-INF/airflow/sources"
+
+/**
+ * How long the describe run gets before the build gives up on it and packs
+ * the entrypoint's source alone. A `main` that does not reach
+ * `Server.create(args)` never answers `--describe-sources`, and one that
+ * starts serving instead would otherwise hold the build open.
+ */
+private const val DESCRIBE_TIMEOUT_SECONDS = 120L
+private const val DRAIN_TIMEOUT_MILLIS = 2_000L
+
+/**
+ * Finds the source file of a class by its `SourceFile` attribute: the class
+ * file's package path plus that name, looked up in each source directory.
+ */
+internal class SourceLocator(
+ private val classesDirs: Iterable<File>,
+ private val sourceDirs: Iterable<File>,
+) {
+ /** Path of the source file relative to its source directory, or `null` if it cannot be found. */
+ fun locate(className: String): String? {
+ val packagePath = className.substringBeforeLast('.', "").replace('.', '/')
+ val classFile =
+ classesDirs
+ .map { File(it, className.replace('.', '/') + ".class") }
+ .firstOrNull { it.isFile } ?: return null
+ val sourceFile = readSourceFile(classFile.readBytes())?.takeIf { it.isNotEmpty() && '/' !in it && '\\' !in it }
+ val relative = listOfNotNull(packagePath.ifEmpty { null }, sourceFile ?: return null).joinToString("/")
+ return relative.takeIf { path -> sourceDirs.any { File(it, path).isFile } }
+ }
+
+ fun file(relativePath: String): File = sourceDirs.map { File(it, relativePath) }.first { it.isFile }
+}
+
+/**
+ * The `META-INF/airflow/sources.json` body: `entrypoint_path` and each Dag ID
+ * in `dag_source_paths`.
+ *
+ * Every path is relative to the source directory the file was found in, which
+ * is also where it sits under `META-INF/airflow/sources/` in the JAR, so a
+ * reader resolves one by prefixing that directory.
+ */
+internal fun sourcesJson(
+ entrypointPath: String?,
+ dagSourcePaths: Map<String, String>,
+): String =
+ JsonOutput.prettyPrint(
+ JsonOutput.toJson(
+ linkedMapOf<String, Any>().apply {
+ entrypointPath?.let { put("entrypoint_path", it) }
+ put("dag_source_paths", dagSourcePaths)
+ },
+ ),
+ )
+
+/**
+ * Runs the bundle's `mainClass` in describe mode to learn which class declared
+ * each Dag, then collects those classes' source files, and the entry point's,
+ * for the bundle JAR to carry.
+ */
+@CacheableTask
+abstract class PackDagSources : DefaultTask() {
+ @get:Input
+ @get:Optional
+ abstract val mainClass: Property<String>
+
+ @get:Classpath
+ abstract val classesDirs: ConfigurableFileCollection
+
+ @get:Classpath
+ abstract val runtimeClasspath: ConfigurableFileCollection
+
+ @get:InputFiles
+ @get:PathSensitive(PathSensitivity.RELATIVE)
+ abstract val sourceDirs: ConfigurableFileCollection
+
+ @get:Nested
+ @get:Optional
+ abstract val launcher: Property<JavaLauncher>
+
+ @get:OutputFile
+ abstract val describeFile: RegularFileProperty
+
+ @get:OutputDirectory
+ abstract val sourcesDir: DirectoryProperty
+
+ private var describeFailed = false
+
+ init {
+ outputs.doNotCacheIf("the Dag describe run failed") { describeFailed }
+ }
+
+ @TaskAction
+ fun pack() {
+ val describe = describeFile.get().asFile
+ val root = sourcesDir.get().asFile
+ describe.delete()
+ root.deleteRecursively()
+ describe.parentFile.mkdirs()
+
+ val main = mainClass.get()
+ val declaringClasses = describeDags(main, describe)
+ val locator = SourceLocator(classesDirs.files, sourceDirs.files)
+
+ val entrypoint = locator.locate(main)
+ val dagPaths = linkedMapOf<String, String>()
+ declaringClasses.forEach { (dagId, className) ->
+ val path = locator.locate(className)
+ if (path == null) {
+ logger.info("No source file found for class {} of Dag '{}'; it falls back to the entrypoint", className, dagId)
+ } else {
+ dagPaths[dagId] = path
+ }
+ }
+
+ (listOfNotNull(entrypoint) + dagPaths.values).distinct().forEach { path ->
+ val target = File(root, "$SOURCES_DIR_PATH/$path")
+ target.parentFile.mkdirs()
+ locator.file(path).copyTo(target)
+ }
+ File(root, SOURCES_JSON_PATH).apply {
+ parentFile.mkdirs()
+ writeText(sourcesJson(entrypoint, dagPaths) + "\n")
+ }
+ }
+
+ private fun describeDags(
+ main: String,
+ describe: File,
+ ): Map<String, String> {
+ var output = ""
+ val failure =
+ try {
+ val java =
+ launcher.orNull
+ ?.executablePath
+ ?.asFile
+ ?.absolutePath ?: "java"
+ val classpath = (classesDirs + runtimeClasspath).joinToString(File.pathSeparator) { it.absolutePath }
+ val process =
+ ProcessBuilder(java, "-cp", classpath, main, "--describe-sources", describe.absolutePath)
+ .redirectErrorStream(true)
+ .start()
+ // Drained on its own thread: a main that writes more than the pipe
+ // holds would otherwise block before the timeout could fire.
+ val drain = Thread { output = process.inputStream.bufferedReader().readText() }
+ drain.isDaemon = true
+ drain.start()
+ val finished = process.waitFor(DESCRIBE_TIMEOUT_SECONDS, TimeUnit.SECONDS)
+ if (!finished) process.destroyForcibly()
+ drain.join(DRAIN_TIMEOUT_MILLIS)
+ when {
+ !finished -> "it did not finish within $DESCRIBE_TIMEOUT_SECONDS seconds"
+ process.exitValue() != 0 -> "exit code ${process.exitValue()}"
+ else -> null
+ }
+ } catch (e: Exception) {
+ e.message ?: e.javaClass.simpleName
+ }
+ if (failure == null && describe.isFile) {
+ @Suppress("UNCHECKED_CAST")
+ (runCatching { JsonSlurper().parse(describe) }.getOrNull() as? Map<String, Any?>)?.let { parsed ->
+ return parsed.mapNotNull { (id, cls) -> (cls as? String)?.let { id to it } }.toMap(linkedMapOf())
+ }
+ }
+ describeFailed = true
+ val why = failure ?: "it wrote no valid --describe-sources file; does main call Server.serve?"
Review Comment:
The describe run only answers when `main` builds its server with `Server.create(args)`. A `main` that calls `serve` on a `Server(comm, logs)` it constructed itself never sees `--describe-sources`, so this hint points at the wrong call. Maybe "does main pass its args to Server.create?" (the test at `AirflowSdkPluginTest.kt` L188 asserts the current text).
##########
java-sdk/plugin/src/main/kotlin/org/apache/airflow/sdk/plugin/PackDagSources.kt:
##########
@@ -0,0 +1,222 @@
+/*
+ * 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 groovy.json.JsonOutput
+import groovy.json.JsonSlurper
+import org.gradle.api.DefaultTask
+import org.gradle.api.file.ConfigurableFileCollection
+import org.gradle.api.file.DirectoryProperty
+import org.gradle.api.file.RegularFileProperty
+import org.gradle.api.provider.Property
+import org.gradle.api.tasks.CacheableTask
+import org.gradle.api.tasks.Classpath
+import org.gradle.api.tasks.Input
+import org.gradle.api.tasks.InputFiles
+import org.gradle.api.tasks.Nested
+import org.gradle.api.tasks.Optional
+import org.gradle.api.tasks.OutputDirectory
+import org.gradle.api.tasks.OutputFile
+import org.gradle.api.tasks.PathSensitive
+import org.gradle.api.tasks.PathSensitivity
+import org.gradle.api.tasks.TaskAction
+import org.gradle.jvm.toolchain.JavaLauncher
+import java.io.File
+import java.util.concurrent.TimeUnit
+
+internal const val SOURCES_MANIFEST_ATTRIBUTE = "Airflow-Java-SDK-Sources"
+internal const val SOURCES_JSON_PATH = "META-INF/airflow/sources.json"
+internal const val SOURCES_DIR_PATH = "META-INF/airflow/sources"
+
+/**
+ * How long the describe run gets before the build gives up on it and packs
+ * the entrypoint's source alone. A `main` that does not reach
+ * `Server.create(args)` never answers `--describe-sources`, and one that
+ * starts serving instead would otherwise hold the build open.
+ */
+private const val DESCRIBE_TIMEOUT_SECONDS = 120L
+private const val DRAIN_TIMEOUT_MILLIS = 2_000L
+
+/**
+ * Finds the source file of a class by its `SourceFile` attribute: the class
+ * file's package path plus that name, looked up in each source directory.
+ */
+internal class SourceLocator(
+ private val classesDirs: Iterable<File>,
+ private val sourceDirs: Iterable<File>,
+) {
+ /** Path of the source file relative to its source directory, or `null` if it cannot be found. */
+ fun locate(className: String): String? {
+ val packagePath = className.substringBeforeLast('.', "").replace('.', '/')
+ val classFile =
+ classesDirs
+ .map { File(it, className.replace('.', '/') + ".class") }
+ .firstOrNull { it.isFile } ?: return null
+ val sourceFile = readSourceFile(classFile.readBytes())?.takeIf { it.isNotEmpty() && '/' !in it && '\\' !in it }
+ val relative = listOfNotNull(packagePath.ifEmpty { null }, sourceFile ?: return null).joinToString("/")
+ return relative.takeIf { path -> sourceDirs.any { File(it, path).isFile } }
+ }
+
+ fun file(relativePath: String): File = sourceDirs.map { File(it, relativePath) }.first { it.isFile }
+}
+
+/**
+ * The `META-INF/airflow/sources.json` body: `entrypoint_path` and each Dag ID
+ * in `dag_source_paths`.
+ *
+ * Every path is relative to the source directory the file was found in, which
+ * is also where it sits under `META-INF/airflow/sources/` in the JAR, so a
+ * reader resolves one by prefixing that directory.
+ */
+internal fun sourcesJson(
+ entrypointPath: String?,
+ dagSourcePaths: Map<String, String>,
+): String =
+ JsonOutput.prettyPrint(
+ JsonOutput.toJson(
+ linkedMapOf<String, Any>().apply {
+ entrypointPath?.let { put("entrypoint_path", it) }
+ put("dag_source_paths", dagSourcePaths)
+ },
+ ),
+ )
+
+/**
+ * Runs the bundle's `mainClass` in describe mode to learn which class declared
+ * each Dag, then collects those classes' source files, and the entry point's,
+ * for the bundle JAR to carry.
+ */
+@CacheableTask
+abstract class PackDagSources : DefaultTask() {
+ @get:Input
+ @get:Optional
+ abstract val mainClass: Property<String>
+
+ @get:Classpath
+ abstract val classesDirs: ConfigurableFileCollection
+
+ @get:Classpath
+ abstract val runtimeClasspath: ConfigurableFileCollection
+
+ @get:InputFiles
+ @get:PathSensitive(PathSensitivity.RELATIVE)
+ abstract val sourceDirs: ConfigurableFileCollection
+
+ @get:Nested
+ @get:Optional
+ abstract val launcher: Property<JavaLauncher>
+
+ @get:OutputFile
+ abstract val describeFile: RegularFileProperty
+
+ @get:OutputDirectory
+ abstract val sourcesDir: DirectoryProperty
+
+ private var describeFailed = false
+
+ init {
+ outputs.doNotCacheIf("the Dag describe run failed") { describeFailed }
Review Comment:
`doNotCacheIf` only keeps the result out of the build cache. After a failed describe run `describe-sources.json` was never written, and Gradle treats an output that was missing last time and is still missing as unchanged, so the next `jar` with the same inputs reports `packDagSources` up to date, prints no warning, and keeps the entrypoint-only payload (I checked this with a minimal Gradle task). One timeout on a slow CI box then sticks until a source changes. `outputs.upToDateWhen { describeFile.get().asFile.isFile }` would make it rerun.
##########
java-sdk/plugin/src/test/kotlin/org/apache/airflow/sdk/plugin/AirflowSdkPluginTest.kt:
##########
@@ -0,0 +1,230 @@
+/*
+ * 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 groovy.json.JsonSlurper
+import org.gradle.testkit.runner.BuildResult
+import org.gradle.testkit.runner.GradleRunner
+import org.gradle.testkit.runner.TaskOutcome
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertFalse
+import org.junit.jupiter.api.Assertions.assertNull
+import org.junit.jupiter.api.Assertions.assertTrue
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.io.TempDir
+import java.io.File
+import java.util.jar.JarFile
+
+class AirflowSdkPluginTest {
+ private fun main(describeBody: String) =
+ """
+ package com.example;
+
+ public class Main {
+ public static void main(String[] args) throws Exception {
+ if (args.length == 2 && args[0].equals("--describe-sources")) {
+ $describeBody
+ }
+ }
+ }
+ """.trimIndent()
+
+ private val describeDags =
+ """
+ java.nio.file.Files.write(java.nio.file.Paths.get(args[1]), (
+ "{\"orders\":\"com.example.Main\"," +
+ "\"reports\":\"com.example.dags.Reports\"," +
+ "\"reports_backfill\":\"com.example.dags.Reports\"," +
+ "\"ghost\":\"com.example.Ghost\"}").getBytes());
+ """.trimIndent()
+
+ private fun project(
+ dir: File,
+ mainBody: String = describeDags,
+ mainClass: String? = "com.example.Main",
+ ): File {
+ dir.write("settings.gradle", "rootProject.name = 'bundle-test'\n")
+ dir.write(
+ "build.gradle",
+ """
+ plugins { id 'org.apache.airflow.sdk' }
+ airflowBundle {
+ ${mainClass?.let { "mainClass = '$it'" } ?: ""}
+ fatJar = false
Review Comment:
Every TestKit project here sets `fatJar = false`. With the default, `bundle` ships the `shadowJar` output, and no test checks that this JAR carries `META-INF/airflow/` and the `Airflow-Java-SDK-Sources` attribute. Can you add a test that builds `shadowJar`? It needs an `airflow-sdk` JAR with the schema-version attribute on the classpath, so a stub JAR should do.
##########
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/DagSource.kt:
##########
@@ -0,0 +1,66 @@
+/*
+ * 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.internal
+
+import org.apache.airflow.sdk.DagDef
+
+/**
+ * @suppress
+ *
+ * Tracks which class declared a [DagDef], so the bundle can ship that class's
+ * source file. Public so that processor-generated builders can call [declaredBy];
+ * not user-facing API.
+ */
+object DagSource {
+ private const val SDK_PACKAGE = "org.apache.airflow.sdk."
+ private val IGNORED_PREFIXES = listOf(SDK_PACKAGE, "java.", "javax.", "jdk.", "sun.", "kotlin.", "kotlinx.")
+
+ private val walker = StackWalker.getInstance(StackWalker.Option.RETAIN_CLASS_REFERENCE)
+
+ /**
+ * Names [declaring] as the class that declared [dag], replacing what was
+ * captured at construction. A generated builder calls this so the Dag points
+ * at the annotated class rather than the builder.
+ */
+ @JvmStatic
+ fun declaredBy(
+ dag: DagDef,
+ declaring: Class<*>,
+ ): DagDef {
+ dag.declaringClass = outermost(declaring)
+ return dag
+ }
+
+ /** The outermost class of the first caller outside the SDK and the standard libraries. */
+ internal fun capture(): Class<*>? =
+ walker.walk { frames ->
+ frames
+ .map { it.declaringClass }
+ .filter { c -> IGNORED_PREFIXES.none { c.name.startsWith(it) } }
+ .findFirst()
Review Comment:
This records the innermost non-SDK frame, so a Dag built by a factory maps to the factory's class rather than the class that called it. ADR-0015 decision 1 records factory-built Dags against the file whose module body was running, so in the TS reference a helper called from `orders.ts` resolves to `orders.ts`. It also means a factory that lives in a shared library JAR resolves to a class with no source in this project, and all its Dags fall back to the entrypoint. Is the innermost-frame rule intended? If so, could the Java docs say so, since the ADR says every SDK follows the same shape?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Java SDK: Pack each native Dag's source file into the bundle JAR [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 merged PR #74096:
URL: https://github.com/apache/airflow/pull/74096
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Java SDK: Pack each native Dag's source file into the bundle JAR [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #74096:
URL: https://github.com/apache/airflow/pull/74096#discussion_r4218524573
##########
java-sdk/plugin/src/main/kotlin/org/apache/airflow/sdk/plugin/AirflowSdkPlugin.kt:
##########
@@ -107,21 +120,45 @@ class AirflowSdkPlugin : Plugin<Project> {
ext.fatJar.convention(true)
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.classesDirs.from(main.output.classesDirs)
+ task.runtimeClasspath.from(main.output, 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 ->
task.manifest.attributes(mapOf("Main-Class" to className))
}
}
+ // The sources payload belongs only in the JARs a bundle is assembled from,
+ // so a sources or javadoc JAR does not carry it.
+ if (ext.mainClass.isPresent && task.name in BUNDLE_JAR_TASKS) {
+ task.dependsOn(packTask)
Review Comment:
Applied in 62dd96e93bb, together with the wiring change above.
##########
java-sdk/plugin/src/main/kotlin/org/apache/airflow/sdk/plugin/PackDagSources.kt:
##########
@@ -0,0 +1,240 @@
+/*
+ * 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 groovy.json.JsonOutput
+import groovy.json.JsonSlurper
+import org.gradle.api.DefaultTask
+import org.gradle.api.file.ConfigurableFileCollection
+import org.gradle.api.file.DirectoryProperty
+import org.gradle.api.file.RegularFileProperty
+import org.gradle.api.provider.Property
+import org.gradle.api.tasks.CacheableTask
+import org.gradle.api.tasks.Classpath
+import org.gradle.api.tasks.Input
+import org.gradle.api.tasks.InputFiles
+import org.gradle.api.tasks.Nested
+import org.gradle.api.tasks.Optional
+import org.gradle.api.tasks.OutputDirectory
+import org.gradle.api.tasks.OutputFile
+import org.gradle.api.tasks.PathSensitive
+import org.gradle.api.tasks.PathSensitivity
+import org.gradle.api.tasks.TaskAction
+import org.gradle.jvm.toolchain.JavaLauncher
+import java.io.File
+import java.util.concurrent.TimeUnit
+
+internal const val SOURCES_MANIFEST_ATTRIBUTE = "Airflow-Java-SDK-Sources"
+internal const val SOURCES_JSON_PATH = "META-INF/airflow/sources.json"
+internal const val SOURCES_DIR_PATH = "META-INF/airflow/sources"
+
+/**
+ * How long the describe run gets before the build gives up on it and packs
+ * the entrypoint's source alone. A `main` that does not reach
+ * `Server.create(args)` never answers `--describe-sources`, and one that
+ * starts serving instead would otherwise hold the build open.
+ */
+private const val DESCRIBE_TIMEOUT_SECONDS = 120L
+private const val DRAIN_TIMEOUT_MILLIS = 2_000L
+
+/**
+ * Finds the source file of a class by its `SourceFile` attribute: the class
+ * file's package path plus that name, looked up in each source directory, or
+ * else the one source file of that name when the layout does not match the
+ * package.
+ */
+internal class SourceLocator(
+ private val classesDirs: Iterable<File>,
+ private val sourceDirs: Iterable<File>,
+) {
+ /** Every source file under the source dirs by file name, walked once and only if a lookup misses. */
+ private val pathsByName: Map<String, List<String>> by lazy {
+ sourceDirs
+ .flatMap { dir -> dir.walkTopDown().filter(File::isFile).map { it.relativeTo(dir).invariantSeparatorsPath } }
+ .groupBy { it.substringAfterLast('/') }
+ }
+
+ /** Path of the source file relative to its source directory, or `null` if it cannot be found. */
+ fun locate(className: String): String? {
+ val classFile =
+ classesDirs
+ .map { File(it, className.replace('.', '/') + ".class") }
+ .firstOrNull { it.isFile } ?: return null
+ val sourceFile =
+ readSourceFile(classFile.readBytes())?.takeIf { it.isNotEmpty() && '/' !in it && '\\' !in it } ?: return null
+ val packagePath = className.substringBeforeLast('.', "").replace('.', '/')
+ val byPackage = listOfNotNull(packagePath.ifEmpty { null }, sourceFile).joinToString("/")
+ if (sourceDirs.any { File(it, byPackage).isFile }) return byPackage
+ // Neither javac nor Kotlin requires the package to match the directory, so fall back to the
+ // file name when exactly one source tree holds it. An ambiguous name is left unresolved.
+ return pathsByName[sourceFile]?.distinct()?.singleOrNull()
+ }
+
+ fun file(relativePath: String): File = sourceDirs.map { File(it, relativePath) }.first { it.isFile }
+}
+
+/**
+ * The `META-INF/airflow/sources.json` body: `entrypoint_path` and each Dag ID
+ * in `dag_source_paths`.
+ *
+ * Every path is relative to the source directory the file was found in, which
+ * is also where it sits under `META-INF/airflow/sources/` in the JAR, so a
+ * reader resolves one by prefixing that directory.
+ */
+internal fun sourcesJson(
+ entrypointPath: String?,
+ dagSourcePaths: Map<String, String>,
+): String =
+ JsonOutput.prettyPrint(
+ JsonOutput.toJson(
+ linkedMapOf<String, Any>().apply {
+ entrypointPath?.let { put("entrypoint_path", it) }
+ put("dag_source_paths", dagSourcePaths)
+ },
+ ),
+ )
+
+/**
+ * Runs the bundle's `mainClass` in describe mode to learn which class declared
+ * each Dag, then collects those classes' source files, and the entry point's,
+ * for the bundle JAR to carry.
+ */
+@CacheableTask
+abstract class PackDagSources : DefaultTask() {
+ @get:Input
+ @get:Optional
+ abstract val mainClass: Property<String>
+
+ @get:Classpath
+ abstract val classesDirs: ConfigurableFileCollection
+
+ @get:Classpath
+ abstract val runtimeClasspath: ConfigurableFileCollection
+
+ @get:InputFiles
+ @get:PathSensitive(PathSensitivity.RELATIVE)
+ abstract val sourceDirs: ConfigurableFileCollection
+
+ @get:Nested
+ @get:Optional
+ abstract val launcher: Property<JavaLauncher>
+
+ @get:OutputFile
+ abstract val describeFile: RegularFileProperty
+
+ @get:OutputDirectory
+ abstract val sourcesDir: DirectoryProperty
+
+ private var describeFailed = false
+
+ init {
+ outputs.doNotCacheIf("the Dag describe run failed") { describeFailed }
+ // A failed run leaves describeFile missing, which Gradle would otherwise read as unchanged.
+ outputs.upToDateWhen { describeFile.get().asFile.isFile }
+ }
+
+ @TaskAction
+ fun pack() {
+ val describe = describeFile.get().asFile
+ val root = sourcesDir.get().asFile
+ describe.delete()
+ root.deleteRecursively()
+ describe.parentFile.mkdirs()
+
+ val main = mainClass.get()
+ val declaringClasses = describeDags(main, describe)
+ val locator = SourceLocator(classesDirs.files, sourceDirs.files)
+
+ val entrypoint = locator.locate(main)
+ if (entrypoint == null) {
+ logger.warn("No source file found for entrypoint class {}; the Code view will show no source for this JAR", main)
+ }
+ val dagPaths = linkedMapOf<String, String>()
+ declaringClasses.forEach { (dagId, className) ->
+ val path = locator.locate(className)
+ if (path == null) {
+ logger.warn("No source file found for class {} of Dag '{}'; it falls back to the entrypoint", className, dagId)
+ } else {
+ dagPaths[dagId] = path
+ }
+ }
+
+ (listOfNotNull(entrypoint) + dagPaths.values).distinct().forEach { path ->
+ val target = File(root, "$SOURCES_DIR_PATH/$path")
+ target.parentFile.mkdirs()
+ locator.file(path).copyTo(target)
+ }
+ File(root, SOURCES_JSON_PATH).apply {
+ parentFile.mkdirs()
+ writeText(sourcesJson(entrypoint, dagPaths) + "\n")
+ }
+ }
+
+ private fun describeDags(
+ main: String,
+ describe: File,
+ ): Map<String, String> {
+ var output = ""
+ val failure =
+ try {
+ val java =
+ launcher.orNull
+ ?.executablePath
+ ?.asFile
+ ?.absolutePath ?: "java"
+ val classpath = (classesDirs + runtimeClasspath).joinToString(File.pathSeparator) { it.absolutePath }
+ val process =
+ ProcessBuilder(java, "-cp", classpath, main, "--describe-sources", describe.absolutePath)
+ .redirectErrorStream(true)
+ .start()
+ // Drained on its own thread: a main that writes more than the pipe
+ // holds would otherwise block before the timeout could fire.
+ val drain = Thread { output = process.inputStream.bufferedReader().readText() }
+ drain.isDaemon = true
+ drain.start()
+ val finished = process.waitFor(DESCRIBE_TIMEOUT_SECONDS, TimeUnit.SECONDS)
+ if (!finished) process.destroyForcibly()
+ drain.join(DRAIN_TIMEOUT_MILLIS)
+ when {
+ !finished -> "it did not finish within $DESCRIBE_TIMEOUT_SECONDS seconds"
+ process.exitValue() != 0 -> "exit code ${process.exitValue()}"
+ else -> null
+ }
+ } catch (e: Exception) {
+ e.message ?: e.javaClass.simpleName
+ }
+ if (failure == null && describe.isFile) {
+ @Suppress("UNCHECKED_CAST")
+ (runCatching { JsonSlurper().parse(describe) }.getOrNull() as? Map<String, Any?>)?.let { parsed ->
+ return parsed.mapNotNull { (id, cls) -> (cls as? String)?.let { id to it } }.toMap(linkedMapOf())
+ }
+ }
Review Comment:
Done in b18c0aa45c9: the describe file is now the result, and a non-zero exit or a timeout only logs a warning when the file is there and parses.
##########
java-sdk/plugin/src/main/kotlin/org/apache/airflow/sdk/plugin/AirflowSdkPlugin.kt:
##########
@@ -93,12 +98,20 @@ 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
* the `airflow-sdk` JAR instead. The bundle JAR still contains `Main-Class`.
*
*/
+
Review Comment:
Applied in cb3dc2d48f2.
##########
java-sdk/plugin/src/main/kotlin/org/apache/airflow/sdk/plugin/AirflowSdkPlugin.kt:
##########
@@ -63,6 +65,9 @@ abstract class AirflowBundleExtension {
abstract val fatJar: Property<Boolean>
}
+/** The JAR tasks a deployable bundle is assembled from, fat or thin. */
+private val BUNDLE_JAR_TASKS = setOf("jar", "shadowJar")
Review Comment:
Done in 62dd96e93bb: the payload is wired onto the one JAR the bundle is assembled from, inside the same fatJar branch that picks it.
##########
java-sdk/plugin/src/main/kotlin/org/apache/airflow/sdk/plugin/AirflowSdkPlugin.kt:
##########
@@ -107,21 +120,45 @@ class AirflowSdkPlugin : Plugin<Project> {
ext.fatJar.convention(true)
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.classesDirs.from(main.output.classesDirs)
+ task.runtimeClasspath.from(main.output, main.runtimeClasspath)
Review Comment:
Applied in ba08f8e4fab.
##########
java-sdk/plugin/src/main/kotlin/org/apache/airflow/sdk/plugin/PackDagSources.kt:
##########
@@ -0,0 +1,240 @@
+/*
+ * 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 groovy.json.JsonOutput
+import groovy.json.JsonSlurper
+import org.gradle.api.DefaultTask
+import org.gradle.api.file.ConfigurableFileCollection
+import org.gradle.api.file.DirectoryProperty
+import org.gradle.api.file.RegularFileProperty
+import org.gradle.api.provider.Property
+import org.gradle.api.tasks.CacheableTask
+import org.gradle.api.tasks.Classpath
+import org.gradle.api.tasks.Input
+import org.gradle.api.tasks.InputFiles
+import org.gradle.api.tasks.Nested
+import org.gradle.api.tasks.Optional
+import org.gradle.api.tasks.OutputDirectory
+import org.gradle.api.tasks.OutputFile
+import org.gradle.api.tasks.PathSensitive
+import org.gradle.api.tasks.PathSensitivity
+import org.gradle.api.tasks.TaskAction
+import org.gradle.jvm.toolchain.JavaLauncher
+import java.io.File
+import java.util.concurrent.TimeUnit
+
+internal const val SOURCES_MANIFEST_ATTRIBUTE = "Airflow-Java-SDK-Sources"
+internal const val SOURCES_JSON_PATH = "META-INF/airflow/sources.json"
+internal const val SOURCES_DIR_PATH = "META-INF/airflow/sources"
+
+/**
+ * How long the describe run gets before the build gives up on it and packs
+ * the entrypoint's source alone. A `main` that does not reach
+ * `Server.create(args)` never answers `--describe-sources`, and one that
+ * starts serving instead would otherwise hold the build open.
+ */
+private const val DESCRIBE_TIMEOUT_SECONDS = 120L
Review Comment:
Done in a073f81ca68: airflowBundle takes a describeTimeout, and the default is 30 seconds.
##########
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Server.kt:
##########
@@ -83,10 +91,13 @@ class ApiError(
* The process exits when the coordinator closes the connection (normally after
* one task-instance execution).
*/
-class Server(
- private val comm: InetSocketAddress,
- private val logs: InetSocketAddress,
+class Server private constructor(
+ private val comm: InetSocketAddress?,
+ private val logs: InetSocketAddress?,
+ private val describeSources: File?,
) {
Review Comment:
Done in 88934351785: Server is a sealed class, and create returns either SourceDescriber or CoordinatorServer, so the nullable addresses and the checks around them are gone.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Java SDK: Pack each native Dag's source file into the bundle JAR [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #74096:
URL: https://github.com/apache/airflow/pull/74096#discussion_r4214940416
##########
java-sdk/plugin/src/main/kotlin/org/apache/airflow/sdk/plugin/PackDagSources.kt:
##########
@@ -0,0 +1,240 @@
+/*
+ * 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 groovy.json.JsonOutput
+import groovy.json.JsonSlurper
+import org.gradle.api.DefaultTask
+import org.gradle.api.file.ConfigurableFileCollection
+import org.gradle.api.file.DirectoryProperty
+import org.gradle.api.file.RegularFileProperty
+import org.gradle.api.provider.Property
+import org.gradle.api.tasks.CacheableTask
+import org.gradle.api.tasks.Classpath
+import org.gradle.api.tasks.Input
+import org.gradle.api.tasks.InputFiles
+import org.gradle.api.tasks.Nested
+import org.gradle.api.tasks.Optional
+import org.gradle.api.tasks.OutputDirectory
+import org.gradle.api.tasks.OutputFile
+import org.gradle.api.tasks.PathSensitive
+import org.gradle.api.tasks.PathSensitivity
+import org.gradle.api.tasks.TaskAction
+import org.gradle.jvm.toolchain.JavaLauncher
+import java.io.File
+import java.util.concurrent.TimeUnit
+
+internal const val SOURCES_MANIFEST_ATTRIBUTE = "Airflow-Java-SDK-Sources"
+internal const val SOURCES_JSON_PATH = "META-INF/airflow/sources.json"
+internal const val SOURCES_DIR_PATH = "META-INF/airflow/sources"
+
+/**
+ * How long the describe run gets before the build gives up on it and packs
+ * the entrypoint's source alone. A `main` that does not reach
+ * `Server.create(args)` never answers `--describe-sources`, and one that
+ * starts serving instead would otherwise hold the build open.
+ */
+private const val DESCRIBE_TIMEOUT_SECONDS = 120L
Review Comment:
Might be a good idea to make this configurable in `airflowBundle { }`? (We can probably make this a lot shorter if it’s configurable; 120s is awful long in most circumstances.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Java SDK: Pack each native Dag's source file into the bundle JAR [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #74096:
URL: https://github.com/apache/airflow/pull/74096#discussion_r4214882387
##########
java-sdk/plugin/src/main/kotlin/org/apache/airflow/sdk/plugin/AirflowSdkPlugin.kt:
##########
@@ -107,21 +120,45 @@ class AirflowSdkPlugin : Plugin<Project> {
ext.fatJar.convention(true)
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.classesDirs.from(main.output.classesDirs)
+ task.runtimeClasspath.from(main.output, 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 ->
task.manifest.attributes(mapOf("Main-Class" to className))
}
}
+ // The sources payload belongs only in the JARs a bundle is assembled from,
+ // so a sources or javadoc JAR does not carry it.
+ if (ext.mainClass.isPresent && task.name in BUNDLE_JAR_TASKS) {
+ task.dependsOn(packTask)
Review Comment:
```suggestion
```
The next line already carries the dependency with `from`.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Java SDK: Pack each native Dag's source file into the bundle JAR [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #74096:
URL: https://github.com/apache/airflow/pull/74096#discussion_r4214874715
##########
java-sdk/plugin/src/main/kotlin/org/apache/airflow/sdk/plugin/AirflowSdkPlugin.kt:
##########
@@ -93,12 +98,20 @@ 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
* the `airflow-sdk` JAR instead. The bundle JAR still contains `Main-Class`.
*
*/
+
Review Comment:
```suggestion
```
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Java SDK: Pack each native Dag's source file into the bundle JAR [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #74096:
URL: https://github.com/apache/airflow/pull/74096#discussion_r4209908293
##########
java-sdk/plugin/src/main/kotlin/org/apache/airflow/sdk/plugin/PackDagSources.kt:
##########
@@ -0,0 +1,222 @@
+/*
+ * 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 groovy.json.JsonOutput
+import groovy.json.JsonSlurper
+import org.gradle.api.DefaultTask
+import org.gradle.api.file.ConfigurableFileCollection
+import org.gradle.api.file.DirectoryProperty
+import org.gradle.api.file.RegularFileProperty
+import org.gradle.api.provider.Property
+import org.gradle.api.tasks.CacheableTask
+import org.gradle.api.tasks.Classpath
+import org.gradle.api.tasks.Input
+import org.gradle.api.tasks.InputFiles
+import org.gradle.api.tasks.Nested
+import org.gradle.api.tasks.Optional
+import org.gradle.api.tasks.OutputDirectory
+import org.gradle.api.tasks.OutputFile
+import org.gradle.api.tasks.PathSensitive
+import org.gradle.api.tasks.PathSensitivity
+import org.gradle.api.tasks.TaskAction
+import org.gradle.jvm.toolchain.JavaLauncher
+import java.io.File
+import java.util.concurrent.TimeUnit
+
+internal const val SOURCES_MANIFEST_ATTRIBUTE = "Airflow-Java-SDK-Sources"
+internal const val SOURCES_JSON_PATH = "META-INF/airflow/sources.json"
+internal const val SOURCES_DIR_PATH = "META-INF/airflow/sources"
+
+/**
+ * How long the describe run gets before the build gives up on it and packs
+ * the entrypoint's source alone. A `main` that does not reach
+ * `Server.create(args)` never answers `--describe-sources`, and one that
+ * starts serving instead would otherwise hold the build open.
+ */
+private const val DESCRIBE_TIMEOUT_SECONDS = 120L
+private const val DRAIN_TIMEOUT_MILLIS = 2_000L
+
+/**
+ * Finds the source file of a class by its `SourceFile` attribute: the class
+ * file's package path plus that name, looked up in each source directory.
+ */
+internal class SourceLocator(
+ private val classesDirs: Iterable<File>,
+ private val sourceDirs: Iterable<File>,
+) {
+ /** Path of the source file relative to its source directory, or `null` if it cannot be found. */
+ fun locate(className: String): String? {
+ val packagePath = className.substringBeforeLast('.', "").replace('.', '/')
+ val classFile =
+ classesDirs
+ .map { File(it, className.replace('.', '/') + ".class") }
+ .firstOrNull { it.isFile } ?: return null
+ val sourceFile = readSourceFile(classFile.readBytes())?.takeIf { it.isNotEmpty() && '/' !in it && '\\' !in it }
+ val relative = listOfNotNull(packagePath.ifEmpty { null }, sourceFile ?: return null).joinToString("/")
+ return relative.takeIf { path -> sourceDirs.any { File(it, path).isFile } }
+ }
+
+ fun file(relativePath: String): File = sourceDirs.map { File(it, relativePath) }.first { it.isFile }
+}
+
+/**
+ * The `META-INF/airflow/sources.json` body: `entrypoint_path` and each Dag ID
+ * in `dag_source_paths`.
+ *
+ * Every path is relative to the source directory the file was found in, which
+ * is also where it sits under `META-INF/airflow/sources/` in the JAR, so a
+ * reader resolves one by prefixing that directory.
+ */
+internal fun sourcesJson(
+ entrypointPath: String?,
+ dagSourcePaths: Map<String, String>,
+): String =
+ JsonOutput.prettyPrint(
+ JsonOutput.toJson(
+ linkedMapOf<String, Any>().apply {
+ entrypointPath?.let { put("entrypoint_path", it) }
+ put("dag_source_paths", dagSourcePaths)
+ },
+ ),
+ )
+
+/**
+ * Runs the bundle's `mainClass` in describe mode to learn which class declared
+ * each Dag, then collects those classes' source files, and the entry point's,
+ * for the bundle JAR to carry.
+ */
+@CacheableTask
+abstract class PackDagSources : DefaultTask() {
+ @get:Input
+ @get:Optional
+ abstract val mainClass: Property<String>
+
+ @get:Classpath
+ abstract val classesDirs: ConfigurableFileCollection
+
+ @get:Classpath
+ abstract val runtimeClasspath: ConfigurableFileCollection
+
+ @get:InputFiles
+ @get:PathSensitive(PathSensitivity.RELATIVE)
+ abstract val sourceDirs: ConfigurableFileCollection
+
+ @get:Nested
+ @get:Optional
+ abstract val launcher: Property<JavaLauncher>
+
+ @get:OutputFile
+ abstract val describeFile: RegularFileProperty
+
+ @get:OutputDirectory
+ abstract val sourcesDir: DirectoryProperty
+
+ private var describeFailed = false
+
+ init {
+ outputs.doNotCacheIf("the Dag describe run failed") { describeFailed }
Review Comment:
Confirmed, and fixed the way you suggested:
```kotlin
// A failed run leaves describeFile missing, which Gradle would otherwise read as unchanged.
outputs.upToDateWhen { describeFile.get().asFile.isFile }
```
A test builds twice with a `main` that writes nothing and asserts the second run is not `UP-TO-DATE` and warns again.
Fixed in 0cb996db1b6.
##########
java-sdk/plugin/src/main/kotlin/org/apache/airflow/sdk/plugin/PackDagSources.kt:
##########
@@ -0,0 +1,222 @@
+/*
+ * 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 groovy.json.JsonOutput
+import groovy.json.JsonSlurper
+import org.gradle.api.DefaultTask
+import org.gradle.api.file.ConfigurableFileCollection
+import org.gradle.api.file.DirectoryProperty
+import org.gradle.api.file.RegularFileProperty
+import org.gradle.api.provider.Property
+import org.gradle.api.tasks.CacheableTask
+import org.gradle.api.tasks.Classpath
+import org.gradle.api.tasks.Input
+import org.gradle.api.tasks.InputFiles
+import org.gradle.api.tasks.Nested
+import org.gradle.api.tasks.Optional
+import org.gradle.api.tasks.OutputDirectory
+import org.gradle.api.tasks.OutputFile
+import org.gradle.api.tasks.PathSensitive
+import org.gradle.api.tasks.PathSensitivity
+import org.gradle.api.tasks.TaskAction
+import org.gradle.jvm.toolchain.JavaLauncher
+import java.io.File
+import java.util.concurrent.TimeUnit
+
+internal const val SOURCES_MANIFEST_ATTRIBUTE = "Airflow-Java-SDK-Sources"
+internal const val SOURCES_JSON_PATH = "META-INF/airflow/sources.json"
+internal const val SOURCES_DIR_PATH = "META-INF/airflow/sources"
+
+/**
+ * How long the describe run gets before the build gives up on it and packs
+ * the entrypoint's source alone. A `main` that does not reach
+ * `Server.create(args)` never answers `--describe-sources`, and one that
+ * starts serving instead would otherwise hold the build open.
+ */
+private const val DESCRIBE_TIMEOUT_SECONDS = 120L
+private const val DRAIN_TIMEOUT_MILLIS = 2_000L
+
+/**
+ * Finds the source file of a class by its `SourceFile` attribute: the class
+ * file's package path plus that name, looked up in each source directory.
+ */
+internal class SourceLocator(
+ private val classesDirs: Iterable<File>,
+ private val sourceDirs: Iterable<File>,
+) {
+ /** Path of the source file relative to its source directory, or `null` if it cannot be found. */
+ fun locate(className: String): String? {
+ val packagePath = className.substringBeforeLast('.', "").replace('.', '/')
+ val classFile =
+ classesDirs
+ .map { File(it, className.replace('.', '/') + ".class") }
+ .firstOrNull { it.isFile } ?: return null
+ val sourceFile = readSourceFile(classFile.readBytes())?.takeIf { it.isNotEmpty() && '/' !in it && '\\' !in it }
+ val relative = listOfNotNull(packagePath.ifEmpty { null }, sourceFile ?: return null).joinToString("/")
+ return relative.takeIf { path -> sourceDirs.any { File(it, path).isFile } }
Review Comment:
Right, and the failure mode is the bad kind: nothing logged, no `entrypoint_path`, and a Code view telling the user the JAR embeds no source for a JAR this plugin built. `locate` now falls back to a unique match on the `SourceFile` name, over an index built once and only when the package path misses:
```kotlin
private val pathsByName: Map<String, List<String>> by lazy {
sourceDirs
.flatMap { dir -> dir.walkTopDown().filter(File::isFile).map { it.relativeTo(dir).invariantSeparatorsPath } }
.groupBy { it.substringAfterLast('/') }
}
```
An ambiguous name is left unresolved rather than guessed. The entrypoint miss now warns, and the per-Dag miss moved from `info` to `warn`. A TestKit case covers `package com.example.dags;` living in `src/main/java/dags/Reports.java`.
Fixed in 0cb996db1b6.
##########
java-sdk/plugin/src/test/kotlin/org/apache/airflow/sdk/plugin/AirflowSdkPluginTest.kt:
##########
@@ -0,0 +1,230 @@
+/*
+ * 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 groovy.json.JsonSlurper
+import org.gradle.testkit.runner.BuildResult
+import org.gradle.testkit.runner.GradleRunner
+import org.gradle.testkit.runner.TaskOutcome
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertFalse
+import org.junit.jupiter.api.Assertions.assertNull
+import org.junit.jupiter.api.Assertions.assertTrue
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.io.TempDir
+import java.io.File
+import java.util.jar.JarFile
+
+class AirflowSdkPluginTest {
+ private fun main(describeBody: String) =
+ """
+ package com.example;
+
+ public class Main {
+ public static void main(String[] args) throws Exception {
+ if (args.length == 2 && args[0].equals("--describe-sources")) {
+ $describeBody
+ }
+ }
+ }
+ """.trimIndent()
+
+ private val describeDags =
+ """
+ java.nio.file.Files.write(java.nio.file.Paths.get(args[1]), (
+ "{\"orders\":\"com.example.Main\"," +
+ "\"reports\":\"com.example.dags.Reports\"," +
+ "\"reports_backfill\":\"com.example.dags.Reports\"," +
+ "\"ghost\":\"com.example.Ghost\"}").getBytes());
+ """.trimIndent()
+
+ private fun project(
+ dir: File,
+ mainBody: String = describeDags,
+ mainClass: String? = "com.example.Main",
+ ): File {
+ dir.write("settings.gradle", "rootProject.name = 'bundle-test'\n")
+ dir.write(
+ "build.gradle",
+ """
+ plugins { id 'org.apache.airflow.sdk' }
+ airflowBundle {
+ ${mainClass?.let { "mainClass = '$it'" } ?: ""}
+ fatJar = false
Review Comment:
Added. It builds a stub `airflow-sdk` JAR carrying the schema-version attribute into a local repo inside the test project, runs `shadowJar`, and asserts both `META-INF/airflow/sources.json` and the `Airflow-Java-SDK-Sources` attribute.
Fixed in 0cb996db1b6.
##########
java-sdk/plugin/src/main/kotlin/org/apache/airflow/sdk/plugin/PackDagSources.kt:
##########
@@ -0,0 +1,222 @@
+/*
+ * 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 groovy.json.JsonOutput
+import groovy.json.JsonSlurper
+import org.gradle.api.DefaultTask
+import org.gradle.api.file.ConfigurableFileCollection
+import org.gradle.api.file.DirectoryProperty
+import org.gradle.api.file.RegularFileProperty
+import org.gradle.api.provider.Property
+import org.gradle.api.tasks.CacheableTask
+import org.gradle.api.tasks.Classpath
+import org.gradle.api.tasks.Input
+import org.gradle.api.tasks.InputFiles
+import org.gradle.api.tasks.Nested
+import org.gradle.api.tasks.Optional
+import org.gradle.api.tasks.OutputDirectory
+import org.gradle.api.tasks.OutputFile
+import org.gradle.api.tasks.PathSensitive
+import org.gradle.api.tasks.PathSensitivity
+import org.gradle.api.tasks.TaskAction
+import org.gradle.jvm.toolchain.JavaLauncher
+import java.io.File
+import java.util.concurrent.TimeUnit
+
+internal const val SOURCES_MANIFEST_ATTRIBUTE = "Airflow-Java-SDK-Sources"
+internal const val SOURCES_JSON_PATH = "META-INF/airflow/sources.json"
+internal const val SOURCES_DIR_PATH = "META-INF/airflow/sources"
+
+/**
+ * How long the describe run gets before the build gives up on it and packs
+ * the entrypoint's source alone. A `main` that does not reach
+ * `Server.create(args)` never answers `--describe-sources`, and one that
+ * starts serving instead would otherwise hold the build open.
+ */
+private const val DESCRIBE_TIMEOUT_SECONDS = 120L
+private const val DRAIN_TIMEOUT_MILLIS = 2_000L
+
+/**
+ * Finds the source file of a class by its `SourceFile` attribute: the class
+ * file's package path plus that name, looked up in each source directory.
+ */
+internal class SourceLocator(
+ private val classesDirs: Iterable<File>,
+ private val sourceDirs: Iterable<File>,
+) {
+ /** Path of the source file relative to its source directory, or `null` if it cannot be found. */
+ fun locate(className: String): String? {
+ val packagePath = className.substringBeforeLast('.', "").replace('.', '/')
+ val classFile =
+ classesDirs
+ .map { File(it, className.replace('.', '/') + ".class") }
+ .firstOrNull { it.isFile } ?: return null
+ val sourceFile = readSourceFile(classFile.readBytes())?.takeIf { it.isNotEmpty() && '/' !in it && '\\' !in it }
+ val relative = listOfNotNull(packagePath.ifEmpty { null }, sourceFile ?: return null).joinToString("/")
+ return relative.takeIf { path -> sourceDirs.any { File(it, path).isFile } }
+ }
+
+ fun file(relativePath: String): File = sourceDirs.map { File(it, relativePath) }.first { it.isFile }
+}
+
+/**
+ * The `META-INF/airflow/sources.json` body: `entrypoint_path` and each Dag ID
+ * in `dag_source_paths`.
+ *
+ * Every path is relative to the source directory the file was found in, which
+ * is also where it sits under `META-INF/airflow/sources/` in the JAR, so a
+ * reader resolves one by prefixing that directory.
+ */
+internal fun sourcesJson(
+ entrypointPath: String?,
+ dagSourcePaths: Map<String, String>,
+): String =
+ JsonOutput.prettyPrint(
+ JsonOutput.toJson(
+ linkedMapOf<String, Any>().apply {
+ entrypointPath?.let { put("entrypoint_path", it) }
+ put("dag_source_paths", dagSourcePaths)
+ },
+ ),
+ )
+
+/**
+ * Runs the bundle's `mainClass` in describe mode to learn which class declared
+ * each Dag, then collects those classes' source files, and the entry point's,
+ * for the bundle JAR to carry.
+ */
+@CacheableTask
+abstract class PackDagSources : DefaultTask() {
+ @get:Input
+ @get:Optional
+ abstract val mainClass: Property<String>
+
+ @get:Classpath
+ abstract val classesDirs: ConfigurableFileCollection
+
+ @get:Classpath
+ abstract val runtimeClasspath: ConfigurableFileCollection
+
+ @get:InputFiles
+ @get:PathSensitive(PathSensitivity.RELATIVE)
+ abstract val sourceDirs: ConfigurableFileCollection
+
+ @get:Nested
+ @get:Optional
+ abstract val launcher: Property<JavaLauncher>
+
+ @get:OutputFile
+ abstract val describeFile: RegularFileProperty
+
+ @get:OutputDirectory
+ abstract val sourcesDir: DirectoryProperty
+
+ private var describeFailed = false
+
+ init {
+ outputs.doNotCacheIf("the Dag describe run failed") { describeFailed }
+ }
+
+ @TaskAction
+ fun pack() {
+ val describe = describeFile.get().asFile
+ val root = sourcesDir.get().asFile
+ describe.delete()
+ root.deleteRecursively()
+ describe.parentFile.mkdirs()
+
+ val main = mainClass.get()
+ val declaringClasses = describeDags(main, describe)
+ val locator = SourceLocator(classesDirs.files, sourceDirs.files)
+
+ val entrypoint = locator.locate(main)
+ val dagPaths = linkedMapOf<String, String>()
+ declaringClasses.forEach { (dagId, className) ->
+ val path = locator.locate(className)
+ if (path == null) {
+ logger.info("No source file found for class {} of Dag '{}'; it falls back to the entrypoint", className, dagId)
+ } else {
+ dagPaths[dagId] = path
+ }
+ }
+
+ (listOfNotNull(entrypoint) + dagPaths.values).distinct().forEach { path ->
+ val target = File(root, "$SOURCES_DIR_PATH/$path")
+ target.parentFile.mkdirs()
+ locator.file(path).copyTo(target)
+ }
+ File(root, SOURCES_JSON_PATH).apply {
+ parentFile.mkdirs()
+ writeText(sourcesJson(entrypoint, dagPaths) + "\n")
+ }
+ }
+
+ private fun describeDags(
+ main: String,
+ describe: File,
+ ): Map<String, String> {
+ var output = ""
+ val failure =
+ try {
+ val java =
+ launcher.orNull
+ ?.executablePath
+ ?.asFile
+ ?.absolutePath ?: "java"
+ val classpath = (classesDirs + runtimeClasspath).joinToString(File.pathSeparator) { it.absolutePath }
+ val process =
+ ProcessBuilder(java, "-cp", classpath, main, "--describe-sources", describe.absolutePath)
+ .redirectErrorStream(true)
+ .start()
+ // Drained on its own thread: a main that writes more than the pipe
+ // holds would otherwise block before the timeout could fire.
+ val drain = Thread { output = process.inputStream.bufferedReader().readText() }
+ drain.isDaemon = true
+ drain.start()
+ val finished = process.waitFor(DESCRIBE_TIMEOUT_SECONDS, TimeUnit.SECONDS)
+ if (!finished) process.destroyForcibly()
+ drain.join(DRAIN_TIMEOUT_MILLIS)
+ when {
+ !finished -> "it did not finish within $DESCRIBE_TIMEOUT_SECONDS seconds"
+ process.exitValue() != 0 -> "exit code ${process.exitValue()}"
+ else -> null
+ }
+ } catch (e: Exception) {
+ e.message ?: e.javaClass.simpleName
+ }
+ if (failure == null && describe.isFile) {
+ @Suppress("UNCHECKED_CAST")
+ (runCatching { JsonSlurper().parse(describe) }.getOrNull() as? Map<String, Any?>)?.let { parsed ->
+ return parsed.mapNotNull { (id, cls) -> (cls as? String)?.let { id to it } }.toMap(linkedMapOf())
+ }
+ }
+ describeFailed = true
+ val why = failure ?: "it wrote no valid --describe-sources file; does main call Server.serve?"
Review Comment:
Reworded to "does main pass its args to Server.create?", and the assertion in `AirflowSdkPluginTest` follows it.
Fixed in 0cb996db1b6.
##########
java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/DagSource.kt:
##########
@@ -0,0 +1,66 @@
+/*
+ * 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.internal
+
+import org.apache.airflow.sdk.DagDef
+
+/**
+ * @suppress
+ *
+ * Tracks which class declared a [DagDef], so the bundle can ship that class's
+ * source file. Public so that processor-generated builders can call [declaredBy];
+ * not user-facing API.
+ */
+object DagSource {
+ private const val SDK_PACKAGE = "org.apache.airflow.sdk."
+ private val IGNORED_PREFIXES = listOf(SDK_PACKAGE, "java.", "javax.", "jdk.", "sun.", "kotlin.", "kotlinx.")
+
+ private val walker = StackWalker.getInstance(StackWalker.Option.RETAIN_CLASS_REFERENCE)
+
+ /**
+ * Names [declaring] as the class that declared [dag], replacing what was
+ * captured at construction. A generated builder calls this so the Dag points
+ * at the annotated class rather than the builder.
+ */
+ @JvmStatic
+ fun declaredBy(
+ dag: DagDef,
+ declaring: Class<*>,
+ ): DagDef {
+ dag.declaringClass = outermost(declaring)
+ return dag
+ }
+
+ /** The outermost class of the first caller outside the SDK and the standard libraries. */
+ internal fun capture(): Class<*>? =
+ walker.walk { frames ->
+ frames
+ .map { it.declaringClass }
+ .filter { c -> IGNORED_PREFIXES.none { c.name.startsWith(it) } }
+ .findFirst()
Review Comment:
Intended, and now documented rather than changed. The JVM has no module body for ADR-0015 decision 1 to key on, so a Dag is recorded against the class that constructed it. `java.rst` says so, including that a factory-built Dag maps to the factory's class and that a factory in a dependency JAR has no source in the project and falls back to the `mainClass` source. ADR-0015 decision 5 notes why a JVM SDK resolves differently.
I deliberately left `DagSource.declaredBy` out of the docs: it is reachable, but it sits in the `internal` package with `@suppress` on its KDoc and exists for processor-generated builders, so documenting it as a workaround would be promoting it to API. Say the word if you would rather it became supported.
Documented in 0cb996db1b6.
##########
java-sdk/plugin/src/main/kotlin/org/apache/airflow/sdk/plugin/AirflowSdkPlugin.kt:
##########
@@ -107,21 +120,47 @@ class AirflowSdkPlugin : Plugin<Project> {
ext.fatJar.convention(true)
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.classesDirs.from(main.output.classesDirs)
+ task.runtimeClasspath.from(main.output, 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 ->
task.manifest.attributes(mapOf("Main-Class" to className))
}
}
+ // Only the JARs a bundle is assembled from carry the Dag sources. A
+ // sources or javadoc JAR would claim the attribute without being a
+ // bundle, and the Python side picks the bundle JAR out of a directory
+ // by that attribute.
Review Comment:
Right, `Main-Class` is what both `might_contain_dag` and the coordinator select on, and the `doFirst` stamps it on every `Jar` task anyway. The comment now just says the payload belongs only in the JARs a bundle is assembled from.
Fixed in 0cb996db1b6.
##########
java-sdk/plugin/src/main/kotlin/org/apache/airflow/sdk/plugin/PackDagSources.kt:
##########
@@ -0,0 +1,222 @@
+/*
+ * 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 groovy.json.JsonOutput
+import groovy.json.JsonSlurper
+import org.gradle.api.DefaultTask
+import org.gradle.api.file.ConfigurableFileCollection
+import org.gradle.api.file.DirectoryProperty
+import org.gradle.api.file.RegularFileProperty
+import org.gradle.api.provider.Property
+import org.gradle.api.tasks.CacheableTask
+import org.gradle.api.tasks.Classpath
+import org.gradle.api.tasks.Input
+import org.gradle.api.tasks.InputFiles
+import org.gradle.api.tasks.Nested
+import org.gradle.api.tasks.Optional
+import org.gradle.api.tasks.OutputDirectory
+import org.gradle.api.tasks.OutputFile
+import org.gradle.api.tasks.PathSensitive
+import org.gradle.api.tasks.PathSensitivity
+import org.gradle.api.tasks.TaskAction
+import org.gradle.jvm.toolchain.JavaLauncher
+import java.io.File
+import java.util.concurrent.TimeUnit
+
+internal const val SOURCES_MANIFEST_ATTRIBUTE = "Airflow-Java-SDK-Sources"
+internal const val SOURCES_JSON_PATH = "META-INF/airflow/sources.json"
+internal const val SOURCES_DIR_PATH = "META-INF/airflow/sources"
+
+/**
+ * How long the describe run gets before the build gives up on it and packs
+ * the entrypoint's source alone. A `main` that does not reach
+ * `Server.create(args)` never answers `--describe-sources`, and one that
+ * starts serving instead would otherwise hold the build open.
+ */
+private const val DESCRIBE_TIMEOUT_SECONDS = 120L
+private const val DRAIN_TIMEOUT_MILLIS = 2_000L
+
+/**
+ * Finds the source file of a class by its `SourceFile` attribute: the class
+ * file's package path plus that name, looked up in each source directory.
+ */
+internal class SourceLocator(
+ private val classesDirs: Iterable<File>,
+ private val sourceDirs: Iterable<File>,
+) {
+ /** Path of the source file relative to its source directory, or `null` if it cannot be found. */
+ fun locate(className: String): String? {
+ val packagePath = className.substringBeforeLast('.', "").replace('.', '/')
+ val classFile =
+ classesDirs
+ .map { File(it, className.replace('.', '/') + ".class") }
+ .firstOrNull { it.isFile } ?: return null
+ val sourceFile = readSourceFile(classFile.readBytes())?.takeIf { it.isNotEmpty() && '/' !in it && '\\' !in it }
+ val relative = listOfNotNull(packagePath.ifEmpty { null }, sourceFile ?: return null).joinToString("/")
+ return relative.takeIf { path -> sourceDirs.any { File(it, path).isFile } }
+ }
+
+ fun file(relativePath: String): File = sourceDirs.map { File(it, relativePath) }.first { it.isFile }
+}
+
+/**
+ * The `META-INF/airflow/sources.json` body: `entrypoint_path` and each Dag ID
+ * in `dag_source_paths`.
+ *
+ * Every path is relative to the source directory the file was found in, which
+ * is also where it sits under `META-INF/airflow/sources/` in the JAR, so a
+ * reader resolves one by prefixing that directory.
+ */
+internal fun sourcesJson(
+ entrypointPath: String?,
+ dagSourcePaths: Map<String, String>,
+): String =
+ JsonOutput.prettyPrint(
+ JsonOutput.toJson(
+ linkedMapOf<String, Any>().apply {
+ entrypointPath?.let { put("entrypoint_path", it) }
+ put("dag_source_paths", dagSourcePaths)
Review Comment:
Good catch, `dag_source_paths` was written and never read. `get_source_code` now passes the ID it is given:
```python
info = _find_source_entry(zf, dag_id)
```
Tests cover a Dag mapped to its own file getting that file back, an unmapped Dag, and `dag_id=None`, with the entrypoint fallback where expected.
Fixed in 0cb996db1b6.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Java SDK: Pack each native Dag's source file into the bundle JAR [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #74096:
URL: https://github.com/apache/airflow/pull/74096#discussion_r4214835361
##########
java-sdk/plugin/src/main/kotlin/org/apache/airflow/sdk/plugin/PackDagSources.kt:
##########
@@ -0,0 +1,240 @@
+/*
+ * 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 groovy.json.JsonOutput
+import groovy.json.JsonSlurper
+import org.gradle.api.DefaultTask
+import org.gradle.api.file.ConfigurableFileCollection
+import org.gradle.api.file.DirectoryProperty
+import org.gradle.api.file.RegularFileProperty
+import org.gradle.api.provider.Property
+import org.gradle.api.tasks.CacheableTask
+import org.gradle.api.tasks.Classpath
+import org.gradle.api.tasks.Input
+import org.gradle.api.tasks.InputFiles
+import org.gradle.api.tasks.Nested
+import org.gradle.api.tasks.Optional
+import org.gradle.api.tasks.OutputDirectory
+import org.gradle.api.tasks.OutputFile
+import org.gradle.api.tasks.PathSensitive
+import org.gradle.api.tasks.PathSensitivity
+import org.gradle.api.tasks.TaskAction
+import org.gradle.jvm.toolchain.JavaLauncher
+import java.io.File
+import java.util.concurrent.TimeUnit
+
+internal const val SOURCES_MANIFEST_ATTRIBUTE = "Airflow-Java-SDK-Sources"
+internal const val SOURCES_JSON_PATH = "META-INF/airflow/sources.json"
+internal const val SOURCES_DIR_PATH = "META-INF/airflow/sources"
+
+/**
+ * How long the describe run gets before the build gives up on it and packs
+ * the entrypoint's source alone. A `main` that does not reach
+ * `Server.create(args)` never answers `--describe-sources`, and one that
+ * starts serving instead would otherwise hold the build open.
+ */
+private const val DESCRIBE_TIMEOUT_SECONDS = 120L
+private const val DRAIN_TIMEOUT_MILLIS = 2_000L
+
+/**
+ * Finds the source file of a class by its `SourceFile` attribute: the class
+ * file's package path plus that name, looked up in each source directory, or
+ * else the one source file of that name when the layout does not match the
+ * package.
+ */
+internal class SourceLocator(
+ private val classesDirs: Iterable<File>,
+ private val sourceDirs: Iterable<File>,
+) {
+ /** Every source file under the source dirs by file name, walked once and only if a lookup misses. */
+ private val pathsByName: Map<String, List<String>> by lazy {
+ sourceDirs
+ .flatMap { dir -> dir.walkTopDown().filter(File::isFile).map { it.relativeTo(dir).invariantSeparatorsPath } }
+ .groupBy { it.substringAfterLast('/') }
+ }
+
+ /** Path of the source file relative to its source directory, or `null` if it cannot be found. */
+ fun locate(className: String): String? {
+ val classFile =
+ classesDirs
+ .map { File(it, className.replace('.', '/') + ".class") }
+ .firstOrNull { it.isFile } ?: return null
+ val sourceFile =
+ readSourceFile(classFile.readBytes())?.takeIf { it.isNotEmpty() && '/' !in it && '\\' !in it } ?: return null
+ val packagePath = className.substringBeforeLast('.', "").replace('.', '/')
+ val byPackage = listOfNotNull(packagePath.ifEmpty { null }, sourceFile).joinToString("/")
+ if (sourceDirs.any { File(it, byPackage).isFile }) return byPackage
+ // Neither javac nor Kotlin requires the package to match the directory, so fall back to the
+ // file name when exactly one source tree holds it. An ambiguous name is left unresolved.
+ return pathsByName[sourceFile]?.distinct()?.singleOrNull()
+ }
+
+ fun file(relativePath: String): File = sourceDirs.map { File(it, relativePath) }.first { it.isFile }
+}
+
+/**
+ * The `META-INF/airflow/sources.json` body: `entrypoint_path` and each Dag ID
+ * in `dag_source_paths`.
+ *
+ * Every path is relative to the source directory the file was found in, which
+ * is also where it sits under `META-INF/airflow/sources/` in the JAR, so a
+ * reader resolves one by prefixing that directory.
+ */
+internal fun sourcesJson(
+ entrypointPath: String?,
+ dagSourcePaths: Map<String, String>,
+): String =
+ JsonOutput.prettyPrint(
+ JsonOutput.toJson(
+ linkedMapOf<String, Any>().apply {
+ entrypointPath?.let { put("entrypoint_path", it) }
+ put("dag_source_paths", dagSourcePaths)
+ },
+ ),
+ )
+
+/**
+ * Runs the bundle's `mainClass` in describe mode to learn which class declared
+ * each Dag, then collects those classes' source files, and the entry point's,
+ * for the bundle JAR to carry.
+ */
+@CacheableTask
+abstract class PackDagSources : DefaultTask() {
+ @get:Input
+ @get:Optional
+ abstract val mainClass: Property<String>
+
+ @get:Classpath
+ abstract val classesDirs: ConfigurableFileCollection
+
+ @get:Classpath
+ abstract val runtimeClasspath: ConfigurableFileCollection
+
+ @get:InputFiles
+ @get:PathSensitive(PathSensitivity.RELATIVE)
+ abstract val sourceDirs: ConfigurableFileCollection
+
+ @get:Nested
+ @get:Optional
+ abstract val launcher: Property<JavaLauncher>
+
+ @get:OutputFile
+ abstract val describeFile: RegularFileProperty
+
+ @get:OutputDirectory
+ abstract val sourcesDir: DirectoryProperty
+
+ private var describeFailed = false
+
+ init {
+ outputs.doNotCacheIf("the Dag describe run failed") { describeFailed }
+ // A failed run leaves describeFile missing, which Gradle would otherwise read as unchanged.
+ outputs.upToDateWhen { describeFile.get().asFile.isFile }
+ }
+
+ @TaskAction
+ fun pack() {
+ val describe = describeFile.get().asFile
+ val root = sourcesDir.get().asFile
+ describe.delete()
+ root.deleteRecursively()
+ describe.parentFile.mkdirs()
+
+ val main = mainClass.get()
+ val declaringClasses = describeDags(main, describe)
+ val locator = SourceLocator(classesDirs.files, sourceDirs.files)
+
+ val entrypoint = locator.locate(main)
+ if (entrypoint == null) {
+ logger.warn("No source file found for entrypoint class {}; the Code view will show no source for this JAR", main)
+ }
+ val dagPaths = linkedMapOf<String, String>()
+ declaringClasses.forEach { (dagId, className) ->
+ val path = locator.locate(className)
+ if (path == null) {
+ logger.warn("No source file found for class {} of Dag '{}'; it falls back to the entrypoint", className, dagId)
+ } else {
+ dagPaths[dagId] = path
+ }
+ }
+
+ (listOfNotNull(entrypoint) + dagPaths.values).distinct().forEach { path ->
+ val target = File(root, "$SOURCES_DIR_PATH/$path")
+ target.parentFile.mkdirs()
+ locator.file(path).copyTo(target)
+ }
+ File(root, SOURCES_JSON_PATH).apply {
+ parentFile.mkdirs()
+ writeText(sourcesJson(entrypoint, dagPaths) + "\n")
+ }
+ }
+
+ private fun describeDags(
+ main: String,
+ describe: File,
+ ): Map<String, String> {
+ var output = ""
+ val failure =
+ try {
+ val java =
+ launcher.orNull
+ ?.executablePath
+ ?.asFile
+ ?.absolutePath ?: "java"
+ val classpath = (classesDirs + runtimeClasspath).joinToString(File.pathSeparator) { it.absolutePath }
+ val process =
+ ProcessBuilder(java, "-cp", classpath, main, "--describe-sources", describe.absolutePath)
+ .redirectErrorStream(true)
+ .start()
+ // Drained on its own thread: a main that writes more than the pipe
+ // holds would otherwise block before the timeout could fire.
+ val drain = Thread { output = process.inputStream.bufferedReader().readText() }
+ drain.isDaemon = true
+ drain.start()
+ val finished = process.waitFor(DESCRIBE_TIMEOUT_SECONDS, TimeUnit.SECONDS)
+ if (!finished) process.destroyForcibly()
+ drain.join(DRAIN_TIMEOUT_MILLIS)
+ when {
+ !finished -> "it did not finish within $DESCRIBE_TIMEOUT_SECONDS seconds"
+ process.exitValue() != 0 -> "exit code ${process.exitValue()}"
+ else -> null
+ }
+ } catch (e: Exception) {
+ e.message ?: e.javaClass.simpleName
+ }
+ if (failure == null && describe.isFile) {
+ @Suppress("UNCHECKED_CAST")
+ (runCatching { JsonSlurper().parse(describe) }.getOrNull() as? Map<String, Any?>)?.let { parsed ->
+ return parsed.mapNotNull { (id, cls) -> (cls as? String)?.let { id to it } }.toMap(linkedMapOf())
+ }
+ }
Review Comment:
I think we should accept the result as long as the file exists. The JVM process might fail with whatever reason at teardown (or not even exit at all and get killed after the timeout) if the user does some weird stuff, but as long as the file is there, we should be able to tell at least _that_ part of the process worked, which is what we really care about. We can log a warning (error?) for this case.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Java SDK: Pack each native Dag's source file into the bundle JAR [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #74096:
URL: https://github.com/apache/airflow/pull/74096#discussion_r4214871735
##########
java-sdk/plugin/src/main/kotlin/org/apache/airflow/sdk/plugin/AirflowSdkPlugin.kt:
##########
@@ -63,6 +65,9 @@ abstract class AirflowBundleExtension {
abstract val fatJar: Property<Boolean>
}
+/** The JAR tasks a deployable bundle is assembled from, fat or thin. */
+private val BUNDLE_JAR_TASKS = setOf("jar", "shadowJar")
Review Comment:
This duplicates the decision the bundle task already makes. The two can drift, and a user building the bundle from a custom jar task can't get the payload.
Wire the payload onto the jar task consumed by the bundle task, in the same `if (ext.fatJar.get())` branch instead.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org