Java SDK: Trigger another Dag's run from a native Java Dag

[jason810496]

• depends on: Java SDK: Choose one of several tasks with a branch #74439
• related: Java SDK: Pack each native Dag's source file into the bundle JAR #74096, Java SDK: Pick one of two tasks with a condition #74438, Go SDK: run TriggerDagRun tasks in the Go runtime #74417

Merge order:

1. Java SDK: Pack each native Dag's source file into the bundle JAR #74096 — Pack each native Dag's source file into the bundle JAR
2. Java SDK: Pick one of two tasks with a condition #74438 — Pick one of two tasks with a condition
3. Java SDK: Choose one of several tasks with a branch #74439 — Choose one of several tasks with a branch
4. Java SDK: Trigger another Dag's run from a native Java Dag #74440 — Trigger another Dag's run from a native Java Dag (current one)

Why

A native Java Dag could not start another Dag's run. TriggerDagRunOperator is Python, and a Java bundle has no Python Dag file to declare one in, so a Java Dag that fans out to another Dag needed a Python counterpart for that one task.

What changes

A TriggerDagRun passed to dag.task declares a task with no Java body; the Java runtime runs it as TriggerDagRunOperator does. Its settings are named as Python names the operator's parameters.

// Interface style.
dag.task("trigger_downstream",
    new TriggerDagRun("downstream_etl")
        .config("wait_for_completion", true)
        .config("poke_interval", Duration.ofSeconds(30))
        .config("conf", Map.of("rows", 2)));

// Annotation style: an ordinary @Builder.Task method that returns a TriggerDagRun.
// The serialized Dag has to carry what is triggered, so the method runs when the Dag
// is built rather than when the task runs, and therefore takes no parameters.
@Builder.Task(id = "trigger_downstream")
public TriggerDagRun triggerDownstream() {
  return new TriggerDagRun("downstream_etl").config("wait_for_completion", true);
}

• The task serializes as Python's own TriggerDagRunOperator — its template fields, UI colour and extra link — plus a dag_dependencies entry, so the UI draws the dependency between the two Dags. Python writes only an operator's template fields and leaves the rest to the Dag file; a Java Dag has no such file, so every setting the author made is written too.
• The runtime records the triggered run's link and ID as XComs, skips or fails on a run that already exists, and waits for completion either by polling or, with deferrable, by deferring to the Python triggerer's DagStateTrigger and resuming from the event it fires. This is the first Java runtime to emit DeferTask and a skipped TaskState.

One shared fixture line

scripts/ci/lang_sdk_serialization/test_dags.yaml now gives the shared trigger Dag a real logical date instead of a Jinja template, because the SDK takes one as a date-time. #74417 makes the same change for the Go SDK; whichever lands second drops that line, and until then the Go serialization conformance check reports that single difference.

---

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

• Yes, with help of Claude Code Opus 5.5 following the guidelines

[jason810496]

Thanks for the review.

[guan404ming]

Thanks!