Java SDK: answer TaskHandlerParseRequest with registered handlers

[jason810496]

Stack (bottom to top), on native #74037 (stack #74170): #73973, #73974, #73975, #73977, #73976, #74067, #74135, #74140, #74141, #73971, #74030, #74031, #74032, #74136, #74137, #74138, #74139

Why

The Dag processor will check each stub task against the task handler that the artifact a worker would run registers (ADR-0011, ADR-0012). #73974 added the probe and the coordinator hooks, and #73975 made Go bundles answer. Today a Java bundle fails on a TaskHandlerParseRequest as an unexpected first frame.

This PR makes a Java bundle answer with every handler registered through @Builder.TaskHandler or Bundle.register(dagId, taskId, cls) and how each handler's parameters bind. It also lets JavaCoordinator find the JAR a stub task would run and start it for the probe. Nothing calls the probe yet: #74135 does.

What changes

A Java bundle registers task handlers as before:

public static class ReportInput implements TaskInput {
  @ArgName("run_label") public String label;
  public long transformed;
}

@Builder.TaskHandler(dag = "etl", task = "transform")
public long transform(Client client, long extracted, List<Double> weights) { ... }

@Builder.TaskHandler(dag = "etl", task = "report")
public void report(ReportInput input) { ... }

When the Dag processor probes it, the bundle replies with every task handler it registers, keyed by Dag id. task_handlers depends only on the JAR, never on the request, and fileloc repeats the request's file.

TaskHandlerParsingResult(
    fileloc="/bundles/java/etl.jar",
    task_handlers={"etl": [
        TaskHandlerDeclaration(task_id="transform", binding="positional", params=[
            TaskHandlerParam(name="extracted", value_schema={"type": "integer", "format": "int64"}),
            TaskHandlerParam(name="weights", value_schema={"anyOf": [
                {"type": "array", "items": {"anyOf": [{"type": "number", "format": "double"}, {"type": "null"}]}},
                {"type": "null"}]}),
        ]),
        TaskHandlerDeclaration(task_id="report", binding="named", params=[
            TaskHandlerParam(name="run_label", exact_name=True,
                             value_schema={"anyOf": [{"type": "string"}, {"type": "null"}]}),
            TaskHandlerParam(name="transformed", value_schema={"type": "integer", "format": "int64"}),
        ]),
    ]},
)

Java runtime

• Task handlers only. Only @Builder.TaskHandler classes and Bundle.register(dagId, taskId, cls) are declared, in registration order, and no handler runs. The tasks of a Dag declared in Java (DagDef, @Builder.Dag) are never task handlers, so the reply leaves them out. A test pins this, and the Server KDoc says so.
• Dag tasks are refused. The processor marks each class it generates for a @Builder.Dag task with @GeneratedDagTask, and Bundle.register(dagId, taskId, cls) rejects such a class. Its task belongs to a Dag declared in Java, so the error tells the author to declare a task handler with @Builder.TaskHandler. A test fails without the check, and the Bundle.register KDoc and java-sdk/README.md say so.
• Binding. Flat data parameters are positional, with their Java names. A TaskInput (as the only data parameter, or as an InputTask<I>) is named, and @ArgName fields match exactly. A handler that reads no argument is named with no params, so the Dag processor only warns about an argument passed to it.
• Parameter names. Java drops parameter names at compile time unless built with -parameters. So the annotation processor gives each class it generates for a @Builder.TaskHandler method a static AIRFLOW_TASK_PARAMS field with its flat parameters' names and declared types, or its TaskInput type. @Builder.Dag task classes get no such field.
• value_schema is the declared Java type in the vocabulary build_arg_bindings emits for a Python annotation:
  • only primitives exclude null;
  • arrays, collections and maps describe their items, and a POJO is {"type": "object"} without its fields;
  • String, the boxed numbers, BigInteger, BigDecimal, UUID, Duration and the java.time date and time types (OffsetDateTime, ZonedDateTime, LocalDateTime, Instant, LocalDate, LocalTime) each get a schema;
  • byte[], Object, Optional, Date, URI and the java.time types not listed above (such as OffsetTime and Period) get none;
  • an enum lists its constant names, sorted, because Class.getFields has no defined order;
  • it does not widen for the decoder's coercions (it also turns "5" into a long), as pydantic's schemas do for its lax mode.
• Reply shape. The generated models cannot hold task_handlers, so the reply is a plain map. An SDK test checks its keys, required fields and binding values against the vendored supervisor schema.
• Reply handshake. The runtime waits for the parent's acknowledgement before it exits, so the parent has the result first.

Vendored supervisor schema

java-sdk/sdk/schema/schema.json was refreshed only when its api_version changed, by downloading the published file. A version is published only at release, so the in-progress 2026-10-30 returns 404 and messages added during the cycle never reached the Java models.

• The Gradle task :sdk:syncSupervisorSchema now copies the Task SDK snapshot when it declares the configured version. It downloads the published file as before otherwise.
• Code generation reads only the committed schema.json, and the snapshot is copied only with java-sdk/gradlew -p java-sdk :sdk:syncSupervisorSchema, then commit the file. No build step runs the task, so a later change to the Task SDK snapshot neither rewrites the file nor fails ktlint. A read-only :sdk:checkSupervisorSchemaVersion fails generation when the file is not at the configured version.
• The refresh brings in the task handler messages, and main's changes since the last vendoring (DagSourceCode, UpdateDagRunNote, retry_reason, multi_team and more required fields). It also drops map_index and element_index from XComArgBinding, which the Task SDK snapshot never declared; ArgBinding.kt still reads them from the raw map, so decoding does not change. The Java SDK and its tests need no other change.
• The sync-java-sdk-supervisor-schema hook stays a version-only check at the pre-commit stage and runs the task only when the configured version changes. Only the hook's docstring and prek description changed.

Coordinator

• JavaCoordinator._find_task_handler_artifact(bundle_path=..., dag_id=...) is the worker's own scan (_JarInfo.find, main_class included), so the Dag processor checks the JAR a worker would run. The Dag id plays no part, because a task of any Dag runs the same Main-Class.
• It returns the resolved path of the JAR that sets that Main-Class. The scan can stop on another JAR, such as the airflow-sdk JAR that gives a thin JAR its schema version, so _JarInfo now records the path. A caller computes the path in the Dag bundle relative to bundle_path.resolve().
• JavaCoordinator._build_parse_task_handler_command(path=...) is a task's own command, so the probe answers for the classpath and Main-Class a worker would run. Without main_class that is the scan's pick (Coordinator: detect ambiguous entrypoints and duplicate dag_ids at import time #71134).
• Probe a Lang-SDK artifact for its task handlers #73974's gate applies: the probe starts only a runtime whose schema version is at least 2026-10-30, and the Java SDK reports it.
• A JAR built from main's Java SDK before this change also reports 2026-10-30 but cannot answer, so its probe fails. Check stub tasks against task handlers when a Dag file is parsed #74135 then logs a warning, and rebuilding the bundle fixes it.

Bundles

The example, the e2e test bundle, the Scala Spark example and the k8s example registered the Java tasks of Python-owned Dags as a Dag declared in Java, which the reply never reports. They now register those tasks as task handlers. Execution already falls back to task handlers, so runs do not change.

Docs

• The contributor guide (30_new_language_sdk.rst) says JavaCoordinator implements both hooks.
• java-sdk/README.md describes the parse and the rejection of a @Builder.Dag task class. A task of a Python Dag is a @Builder.TaskHandler or a hand-written Task class registered with Bundle.register(dagId, taskId, cls).

Builds on #73974 (hooks, parse_task_handler, the gate, the probe) and #73973 (the models), and on native #74036 (_build_command, _JarInfo with its sorted walk, read_main_attributes). No user docs: #74135 adds them.

How to test

java-sdk/gradlew -p java-sdk build
uv run --project scripts pytest scripts/tests/ci/prek/test_sync_java_sdk_supervisor_schema.py -q
uv run --project task-sdk --with-editable shared/secrets_masker pytest task-sdk/tests/task_sdk/coordinators -q
uv run --project airflow-core --with-editable shared/secrets_masker pytest airflow-core/tests/unit/dag_processing/ -q
# Builds java-sdk/example as a thin bundle and probes it:
AIRFLOW_LANG_SDK_REAL_PROBE_TESTS=1 uv run --project airflow-core --with-editable shared/secrets_masker pytest airflow-core/tests/unit/dag_processing/test_task_handler_processor_java.py -q
SKIP=lychee-docker,regenerate-java-sdk-verification-metadata prek run --from-ref <parent> --to-ref HEAD --stage pre-commit
prek run mypy-task-sdk --all-files
prek run mypy-airflow-core --all-files

Ran:

• Java SDK, JDK 11 (Temurin 11.0.32.1): java-sdk/gradlew -p java-sdk build --rerun-tasks passed with 202 tests against 190 before (processor 24, sdk 150, jpl 10, jul 6, log4j2 5, slf4j 7). :sdk:checkSupervisorSchemaVersion passes (api_version=2026-10-30), the build leaves schema.json untouched, and cmp shows it identical to the monorepo snapshot.
• scripts: the sync hook tests, 5 passed. Main had none.
• Scratch repo with the ktlint, spotless and sync hooks, java-sdk and both schemas, after adding a $defs entry to the snapshot at the same api_version: prek run ktlint --all-files failed with files were modified before this change, because the build copied the snapshot over schema.json. Now all three hooks pass with --all-files, a Gradle build leaves the file untouched, and java-sdk/gradlew -p java-sdk :sdk:syncSupervisorSchema copies the snapshot when asked. When the configured version and the snapshot move together, a build fails on the check and names that command, and the sync hook refreshes the file.
• task-sdk coordinators/: 412 passed, run serially. java/test_coordinator.py goes from 57 to 74 tests: TestMainJar (3 new, 5 updated), TestFindTaskHandlerArtifact (8) and TestBuildParseTaskHandlerCommand (6, including the probe exec).
• airflow-core dag_processing/: 747 passed, 10 skipped (the 8 that need a backend, the FAB auth manager or the system test Dags, and the two real probes without their variable).
• The real Java probe with AIRFLOW_LANG_SDK_REAL_PROBE_TESTS=1: 1 passed in 18 s with a warm Gradle cache, on JDK 11. It finds probe-example.jar (the airflow-sdk JAR beside it sets the schema version) and the probe answers with the three Dags that register task handlers. The native java_native_interface_example Dag is left out.
• The four fixtures build against the working-tree SDK through an included build, so nothing is published. The e2e and k8s bundles build on JDK 11 (25 JARs each). The Scala Spark example builds a thin bundle of 167 JARs on JDK 17. Without main_class the scan stops on javassist.CtClass, because five dependency JARs set a Main-Class, so such a bundle needs main_class. With it set, the find returns compile-scala.jar and the probe answers scala_spark_example with spark_extract, spark_transform and spark_load, each named with no params.
• Each commit passes its own tests when checked out alone.
• Each new test fails without its change:
  • The sync hook's logic is unchanged, so its tests pin the existing version-only check and also pass against the base script. The check task and the on-request copy have no unit test, and the scratch-repo run above covers them.
  • Restoring the old BuilderProcessor.kt fails 11 processor tests: three for the handler params field and eight for the marker. Dropping only the marker fails the eight.
  • Removing the check from Bundle.register fails shouldRejectClassGeneratedForDagTask.
  • ValueSchemaTest does not compile without ValueSchema.kt, and dropping .sorted() fails the scalar test.
  • Restoring the old Server.kt fails both Server parse tests, the second on its check of the error (Unexpected initial frame instead of Coordinator comm closed), not on a timeout. Adding the Dags of the bundle to the reply fails two TaskHandlerParseTest tests and the Server answer test. Renaming exact_name fails the schema check, and swallowing the missing acknowledgement fails failsTaskHandlerParseWithoutAcknowledgement.
  • 22 of the new or updated coordinator tests fail on the old coordinator. Recording the path on the schema-version branch instead of the Main-Class branch fails three, an unresolved path fails the symlink test, building the probe command with _build_jar_command fails the two-JARs test, and scanning the directory of the found JAR instead of the scan roots fails the app/ and libs/ layout test.
  • With the old Server.kt the real probe returns no handlers and The Lang-SDK runtime exited with code 1 without a parse result. With the old coordinator the find raises NotImplementedError, and with the old example java_interface_example is missing.
• prek over the whole range at the pre-commit stage passed with lychee-docker and regenerate-java-sdk-verification-metadata skipped (66 hooks passed, the rest had no files to check, including check-supervisor-schemas-versions). The Task SDK schema, airflow-core/src, the Go and TS SDKs and the Gradle plugin are untouched. mypy-task-sdk, mypy-airflow-core, mypy-scripts and the sync hook pass with --all-files.

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Claude Code (Opus 5.5) following the guidelines

[jason810496]

This was not really merged. The stack was reordered so the TypeScript layer now sits on top of this Java layer, and once both branches were pushed, this PR's commits were already in its base branch, so GitHub marked it merged automatically and deleted its head branch. No merge commit was made. The Java layer continues in #74317, in the new stack #74318.