Java SDK: Add edge labels to native Dags

[FrankYang0529]

• Deps.Flow.label(flow, text) wraps one end of an edge. Each edge of a fan-out can then carry its own label. Declaring an existing edge again only adds the label to that edge.
• A later label on the same edge replaces the earlier label.

---

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

• Yes - Claude Code

[jason810496]

Nice, thanks.

I pushed 255a2b1 on top with a few fixes from review:

• Labels are now applied in the order they were set, so the label set last wins, as in Python's DAG.set_edge_info. Before, a task edge's own label always beat its group edge's label, and relabeling an existing
  group edge didn't take effect.
• Updated EdgeLabelTest to match. I checked both cases against Python's Label, and the edge_info matches.
• Corrected the KDoc and docs note: Python's Label also changes the edge when one end is in no task group.

[uranusjr]

Python uses the >> Label >> syntax because this naturally reads like having a label in the arrow. This does not read as well in other languages without >>, and makes even less sense with before.

I think we should come up with a better syntax instead just copying Python with >> and << substituted.

[jason810496]

I think we should come up with a better syntax instead just copying Python with >> and << substituted.

I considered the following syntax before:

loaded.before(reportEmpty, "when empty");
cleaned.after(loaded, "always");

However, the wiring should supports fan-in and fan-out.

loaded.before(reportEmpty, "when empty", reportSuspicious, "when suspicious"); // this is not a working syntax to support fan-in.
// the only sutiable way might looks like:
loaded.before([reportEmpty, reportSuspicious], ["when empty", "when suspicious"]); 
loaded.before([reportEmpty,"when empty"], [reportSuspicious,, "when suspicious"]); 

This is why the current syntax goes with current shape, having a dedicated Flow.label wrapper to representing the edge with label:

loaded.before(Flow.label(reportEmpty, "when empty"), Flow.label(reportSuspicious, "when suspicious"));
cleaned.after(Flow.label(loaded, "always"));

The Go-SDK went with the same shape already and there was't objection back then.
I don't have a better user interface design at this moment.

[FrankYang0529]

Thanks for the comments and suggestions.

I agree with Jason that Java checks every varargs argument against the same type. A before(Object...) signature would accept the call above, but it would also accept nodes and labels in any order. Only a runtime check could then catch a misplaced label.

With each label still paired with its node, two changes may make the Java code read better. WDYT?

1. Add a labeled method to the node so that each call reads more like an arrow with a label on it:

   loaded.before(reportEmpty.labeled("when empty"), reportSuspicious.labeled("when suspicious"));
   cleaned.after(loaded.labeled("always"));

A bad part is that labeled becomes a reserved name for task methods inside a task group, because the generated view of a group inherits every method of the node type.

2. Let Then, Else and Case take a label directly, because labels are used most often on branches:

   dag.If(HasRows.class).Then(load, "rows found").Else(reportEmpty, "no rows");
   dag.Switch(PickPath.class).Case(handleLong, "long").Case(handleShort, "short");

[jason810496]

Hi @uranusjr,
Please let me know if there's any further concern from merging this PR when you have a moment, thanks.

[uranusjr]

I feel it’d make more sense to put the label before the task. Maybe something like

loaded.before(Label.of("when empty").on(reportEmpty), Label.of("when suspicious").on(reportSuspicious))

I’m not sure if Python currently has both a >> label >> [b, c] and a >> [label_for_b >> b, label_for_c >> c] (label before or after the flow forks). If this is a thing, it also makes sense to also have

loaded.labeled("forks").before(reportEmpty, reportSuspicious)

It’s more difficult to come up with a syntax for after (since that one is backwards by design), but the general idea is to keep the label text between the tasks to reflect its position in the rendered graph.

[FrankYang0529]

Thanks for the suggestion. The label term between tasks is more clear. I switched to following syntax:

loaded.before(Label.of("when empty").on(reportEmpty), Label.of("when suspicious").on(reportSuspicious));

About Python syntax:

• a >> Label("forks") >> [b, c] labels both edges.
• a >> [Label("x") >> b, Label("y") >> c] draws both edges but labels neither, and raises no error. When Label("x") >> b runs, the label has no upstream yet, and the expression returns b. So the outer expression is just a >> [b, c].

on can take several tasks, so the first form maps to:

loaded.before(Label.of("forks").on(reportEmpty, reportSuspicious)); // a >> Label("forks") >> [b, c]

To keep the text between the tasks, I did not add loaded.labeled("forks"). after also works with the same syntax:

cleaned.after(Label.of("always").on(loaded)); // load >> Label("always") >> cleanup

The Go SDK (airflow.Label(node, text)) and the TS SDK (label(node, text)) both put the task first. Should those two SDKs follow the Java syntax?

[jason810496]

The Go SDK (airflow.Label(node, text)) and the TS SDK (label(node, text)) both put the task first. Should those two SDKs follow the Java syntax?

The TS SDK side edge label just merged yesterday btw.
Yes, let's align the Java side syntax please, thanks a lot.

[github-actions]

Backport successfully created: v3-4-test

Note: As of Merging PRs targeted for Airflow 3.X
the committer who merges the PR is responsible for backporting the PRs that are bug fixes (generally speaking) to the maintenance branches.

In matter of doubt please ask in #release-management Slack channel.

Status	Branch	Result
✅	v3-4-test