Fail instead of silently falling behind when a Lang-SDK vendored Dag schema drifts from airflow-core (#74091)

A core Dag-serialization schema change shipped with the Go SDK's vendored copy
left behind, uncaught by any hook: the sync hooks for Go and Java were manual
only, and the one check that runs on every commit watched the vendored copies,
not the Python source, so it never saw the schema change at all. The same
drift already caused a two-release-old bug in the Go SDK's generated models
(#73954).

Go's and Java's sync hooks now run on every commit and are triggered by the
Python source too, the way the TypeScript SDK's already was, refreshing the
vendored copy and failing when they had to. Java's Gradle task keeps the same
task name and stays silent for generateDagDsl's own dependency, so an
ordinary in-repo build still refreshes a schema someone is mid-edit on
without breaking; only the prek hook's invocation passes the flag that makes
it fail.

java-sdk/sdk/build.gradle.kts

@@ -225,6 +225,13 @@ abstract class SyncDagSchemaTask : DefaultTask() {
     @get:Internal
     abstract val targetFile: RegularFileProperty
 
+    // False for an ordinary in-repo build, which must keep refreshing the copy silently
+    // so it does not break on a schema a developer is actively editing. True for the
+    // prek hook, which passes -PfailOnDagSchemaDrift so a commit that leaves the
+    // vendored copy behind fails instead of shipping unnoticed.
+    @get:Internal
+    abstract val failOnDagSchemaDrift: Property<Boolean>
+
     @TaskAction
     fun sync() {
         val src = sourceFile.get().asFile
@@ -239,6 +246,12 @@ abstract class SyncDagSchemaTask : DefaultTask() {
         }
         logger.lifecycle("Refreshing vendored dag-schema.json from ${src.path}")
         src.copyTo(dst, overwrite = true)
+        if (failOnDagSchemaDrift.getOrElse(false)) {
+            throw GradleException(
+                "Vendored dag-schema.json was out of date and has been refreshed from ${src.path}. " +
+                    "Review the diff and commit it.",
+            )
+        }
     }
 }
 
@@ -661,6 +674,8 @@ val syncDagSchema by tasks.registering(SyncDagSchemaTask::class) {
     description = "Refresh the vendored Dag serialization schema from the monorepo copy when present."
     sourceFile = layout.projectDirectory.file("../../airflow-core/src/airflow/serialization/schema.json")
     targetFile = dagSchemaInput
+    // -PfailOnDagSchemaDrift carries no value, so presence (not content) is the signal.
+    failOnDagSchemaDrift = providers.gradleProperty("failOnDagSchemaDrift").map { true }.orElse(false)
 }
 
 tasks.register<GenerateDagDslTask>("generateDagDsl") {