Skip to content

Java SDK: Pick one of two tasks with a condition - #74438

Merged
jason810496 merged 6 commits into
jason/java-sdk/pack-dag-sourcesfrom
jason/java-sdk/native-dag-if
Oct 9, 2026
Merged

jason810496 merged 6 commits into
jason/java-sdk/pack-dag-sourcesfrom
jason/java-sdk/native-dag-if

Conversation

@jason810496

@jason810496 jason810496 commented Oct 8, 2026 •

Copy link
Copy Markdown
Member

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 (current one)
  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

Why

A native Java Dag could only run every task it declared. Nothing let a task's result decide what runs next, so a Dag that loads only when there are rows had to be split in two or fall back to a Python stub.

What changes

DagDef.If and @Builder.If declare a task whose boolean names the side that runs; the other is skipped.

// Interface style. With no task id given, one comes from the class: "hasRows".
dag.If(HasRows.class).config("retries", 2).then(load).orElse(reportEmpty);

public class HasRows implements ConditionTask {
  @Override
  public boolean decide(Context context, Client client) {
    return ((Number) client.getXCom("transform")).longValue() > 0;
  }
}
// Annotation style.
@Builder.If(id = "has_rows")
public boolean hasRows(long transformed) {
  return transformed > 0;
}

@Builder.Deps
static class Wiring implements EtlDeps {
  void depends() {
    hasRows(transformed).then(load(transformed)).orElse(loadEmpty());
  }
}
  • If is capitalized because if is a Java keyword, and @Builder.If reads the same way.
  • The deciding task carries _can_skip_downstream, and the SDK writes the skipmixin_key XCom before the skip, so Airflow skips a cleared side again rather than running it, as Python's SkipMixin does.
  • A condition skips only the side not taken, so a task that runs after both sides needs a trigger rule such as none_failed_min_one_success.
  • SkipDownstreamTasks is reachable through the transport client only, never through the task-facing Client, so a task body cannot skip tasks the Dag did not declare it may.
  • The serialization conformance check now covers the shared branch Dags, and branching is marked supported.

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

@jason810496
jason810496 added this pull request to stack #74441 October 8, 2026 05:01
@jason810496
jason810496 removed this pull request from stack #74441 October 8, 2026 05:06
@jason810496
jason810496 added this pull request to stack #74442 October 8, 2026 05:06
@jason810496
jason810496 force-pushed the jason/java-sdk/native-dag-if branch from 95369c1 to 30e5218 Compare October 8, 2026 05:33
Comment thread java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Condition.kt Outdated
Comment thread java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Condition.kt
Comment thread java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Condition.kt Outdated
Comment thread java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/execution/TaskTest.kt Outdated
Comment thread java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/ConditionTest.kt Outdated
A native Java Dag could only run every task it declared: it had no way to
let a task's result decide what runs next, so a Dag that loads only when
there are rows had to be split in two or fall back to a Python stub.

`DagDef.If` and `@Builder.If` declare a task whose boolean names the side
that runs; the other is skipped. The deciding task carries
`_can_skip_downstream`, so Airflow skips a cleared side again rather than
running it, and the SDK writes the `skipmixin_key` XCom before the skip, as
Python's SkipMixin does. It skips only the side not taken, so a task after
both sides needs a trigger rule such as `none_failed_min_one_success`.

`If` is capitalized because `if` is a Java keyword and `@Builder.If` reads
the same way. With no task id given, one is derived from the condition
class's simple name, so `HasRows.class` becomes `hasRows`.

The runtime reaches `SkipDownstreamTasks` through the transport client
only, never through the task-facing `Client`, so a task body cannot skip
tasks the Dag did not declare it may.
@jason810496
jason810496 force-pushed the jason/java-sdk/native-dag-if branch from 30e5218 to dbe29e4 Compare October 8, 2026 11:46
@jason810496
jason810496 marked this pull request as ready for review October 8, 2026 12:56
@jason810496
jason810496 requested a review from uranusjr as a code owner October 8, 2026 12:56

@jason810496 jason810496 left a comment

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for the review.

Comment thread java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Condition.kt Outdated
Comment thread java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Condition.kt
Comment thread java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Condition.kt Outdated
Comment thread java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/execution/TaskTest.kt Outdated
Comment thread java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/ConditionTest.kt Outdated
@jason810496
jason810496 merged commit d4013cc into main Oct 9, 2026
192 checks passed
@jason810496
jason810496 deleted the jason/java-sdk/native-dag-if branch October 9, 2026 06:30
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants