[jason810496]

Merge order:

1. Java SDK: Group a native Dag's tasks with task groups #74230 — Group a native Dag's tasks with task groups (current one)
2. Java SDK: Serialize native Dags to DagSerialization v3 #71190 — Serialize native Dags to DagSerialization v3
3. 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

Example

With annotations, a @Builder.TaskGroup class holds the tasks of one group, and nesting one in
another nests the groups:

@Builder.Dag(id = "java_etl", schedule = "@daily")
public class EtlPipeline {

  @Builder.Task
  public long extract() { return 42L; }

  @Builder.TaskGroup                      // the group "Staging", after the class
  static class Staging {
    @Builder.Task
    public long stage(long rows) { return rows; }   // the task "Staging.stage"

    @Builder.TaskGroup(id = "checks")
    static class Checks {
      @Builder.Task
      public void nulls(long staged) { }            // "Staging.checks.nulls"
    }
  }

  @Builder.Deps
  static class Wiring implements EtlPipelineDeps {
    void depends() {
      var rows = extract();
      var staged = staging().stage(rows);
      staging().checks().nulls(staged);
      extract().before(staging());        // the whole group runs after extract
    }
  }
}

The generated view nests the same way, so a group is both the namespace of what it holds and a point
in the flow:

public interface EtlPipelineDeps extends Deps {
  default TaskRef<Long> extract() { ... }
  default Staging staging() { return new Staging() {}; }

  interface Staging extends Group {
    @Override default String groupId() { return "Staging"; }
    default TaskRef<Long> stage(Arg<? extends Long> rows) { ... }
    default Checks checks() { return new Checks() {}; }

    interface Checks extends Group {
      @Override default String groupId() { return "Staging.checks"; }
      default TaskRef<Void> nulls(Arg<? extends Long> staged) { ... }
    }
  }
}

On the interface surface, taskGroup declares a group on the Dag or inside another group:

var staging = dag.taskGroup("staging");
var stage = staging.task("stage", Stage.class);                       // task "staging.stage"
staging.taskGroup("checks").task("nulls", Nulls.class).after(stage);  // "staging.checks.nulls"
extract.before(staging);

How

• Everything declared in a group carries the group's ID as a prefix, so stage in Staging is the
  task Staging.stage.
• A group class is static, non-private, and needs a no-argument constructor, because the generated
  task bodies instantiate it. Its ID defaults to the class name, as a Dag's does.
• Task method names scope to their own group, so two groups can each declare run(). The generated
  task classes are named after the class tree, such as Staging_Checks_Nulls.
• Support before, after and Flow.of.

Serialization of task groups, and the task-group capability flag, land in #71190, which is where a
native Dag is serialized at all.

prefix_group_id = false and the display options (tooltip, colours, display name) are left for a
follow-up.

---

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

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

[jason810496]

Thanks for the review.

[henry3260]

Just want to help unblock the technical side here, not a formal review.

[jason810496]

Thanks for the review.