Everything can be set up in a Breeze shell. Go is installed automatically; pass --sdk java to install Java to "breeze shell" and "breeze start-airflow".

Tasks executed by the Java or Go SDK are wired up through three pieces:

  • A Python stub DAG file defines the DAG shape that the scheduler sees. Each language task is declared with @task.stub(queue="<name>"). The function body can simply be ... ; the stub is never actually executed in Python, so the body is always ignored.
  • A language bundle that registers the same DAG ID and task IDs. This is what actually runs.
  • Airflow configuration maps the queue name to a coordinator class and the bundle location.

Pre-setup: Python stub DAGs

Both SDK sections share a single stub DAG file. The queue name "sdk" is wired to a different coordinator in each section. Change the environment variable AIRFLOW_SDK_QUEUE_TO_COORDINATOR to select which SDK is under test.

Save as /files/dags/aip108_e2e.py :

from __future__ import annotations

from airflow.sdk import dag, task


@dag(schedule=None)
def aip108_e2e():
    @task
    def python_upstream():
        return "hello_from_python"

    @task.stub(queue="sdk")
    def succeed_task(): ...

    @task.stub(queue="sdk")
    def variable_task(): ...

    @task.stub(queue="sdk")
    def connection_task(): ...

    @task.stub(queue="sdk")
    def xcom_task(): ...

    @task.stub(queue="sdk")
    def context_task(): ...

    @task.stub(queue="sdk")
    def taskflow_task(): ...

    @task
    def python_downstream(ti=None):
        value = ti.xcom_pull(task_ids="xcom_task", key="return_value")
        if value is None:
            raise RuntimeError("expected an XCom value from xcom_task")
        print("Cross-language XCom verified:", value)

    py_up = python_upstream()

    succeed_task()
    variable_task()
    connection_task()
    context_task()

    xcom = xcom_task()
    py_up >> xcom >> python_downstream()
    py_up >> taskflow_task()


@dag(schedule=None)
def aip108_fail_e2e():
    @task.stub(queue="sdk")
    def fail_task(): ...

    @task.stub(queue="sdk")
    def panic_task(): ...

    fail_task()
    panic_task()


aip108_e2e()
aip108_fail_e2e()

1. Testing the Java SDK

Prerequisites

Requirement

Version

JDK

11

Gradle

8+

Airflow worker

HEAD

Create a new Gradle project

mkdir /files/aip108-java-dag && cd /files/aip108-java-dag
gradle init --type java-application --dsl kotlin

Replace the generated app/build.gradle.kts with the following. The SDK snapshot is available in the Apache snapshots repository (https://lists.apache.org/thread/w8wtbc9r397pglbklsc26qjsb2dqwn5h).

// app/build.gradle.kts
plugins {
  application
}

repositories {
  mavenCentral()
  maven {
    url = uri("https://repository.apache.org/content/repositories/snapshots/")
    mavenContent { snapshotsOnly() }
  }
}

dependencies {
  annotationProcessor("org.apache.airflow:airflow-sdk:1.0.0-SNAPSHOT")
  implementation("org.apache.airflow:airflow-sdk:1.0.0-SNAPSHOT")
}

application {
  mainClass = "com.example.airflow.MyBundle"
}

tasks.jar {
  manifest {
    attributes("Main-Class" to "com.example.airflow.MyBundle")
  }
}

// settings.gradle.kts
include("app")

Write the tasks

The annotation processor generates MyDagBuilder from MyDag.java at compile time. MyBundle.java is the entry point that wires everything together.

Create app/src/main/java/com/example/airflow/MyDag.java :

package com.example.airflow;

import org.apache.airflow.sdk.*;

@Builder.Dag(id = "aip108_e2e")
public class MyDag {

  @Builder.Task(id = "succeed_task")
  public void succeedTask() {
    System.out.println("succeed_task: returning normally");
  }

  @Builder.Task(id = "variable_task")
  public void variableTask(Client client) {
    Object value = client.getVariable("aip108_test_var");
    System.out.println("variable_task: got variable value = " + value);
    if (value == null) throw new RuntimeException("variable 'aip108_test_var' not found");
  }

  @Builder.Task(id = "connection_task")
  public void connectionTask(Client client) {
    var conn = client.getConnection("aip108_test_conn");
    System.out.println("connection_task: host = " + conn.host);
  }

  @Builder.Task(id = "xcom_task")
  public String xcomTask(Client client, @Builder.XCom(task = "python_upstream") Object upstream) {
    System.out.println("xcom_task: pulled upstream XCom = " + upstream);
    System.out.println("xcom_task: pushed XCom 'hello_from_java'");
    return "hello_from_java";
  }

  // Injecting a Context parameter is the Java equivalent of Python's
  // get_current_context(): the runtime fills it from the task's StartupDetails.
  @Builder.Task(id = "context_task")
  public void contextTask(Context context) {
    System.out.println("context_task: dag_id = " + context.dagRun.dagId
        + ", run_id = " + context.ti.runId
        + ", task_id = " + context.ti.taskId
        + ", try_number = " + context.ti.tryNumber);
  }

  // TaskFlow-style typed XCom input: the upstream return value is injected as a
  // typed parameter (cast to String here) instead of calling client.getXCom.
  @Builder.Task(id = "taskflow_task")
  public void taskflowTask(@Builder.XCom(task = "python_upstream") String fromPython) {
    System.out.println("taskflow_task: typed upstream XCom = " + fromPython);
  }
}

Create app/src/main/java/com/example/airflow/MyFailDag.java :

package com.example.airflow;

import org.apache.airflow.sdk.Builder;

@Builder.Dag(id = "aip108_fail_e2e")
public class MyFailDag {

  @Builder.Task(id = "fail_task")
  public void failTask() {
    throw new RuntimeException("deliberate failure - expected in test JV-02");
  }

  @Builder.Task(id = "panic_task")
  public void panicTask() {
    throw new Error("deliberate panic - expected in test JV-03");
  }
}

Create app/src/main/java/com/example/airflow/MyBundle.java :

package com.example.airflow;

import java.util.List;
import org.apache.airflow.sdk.*;

public class MyBundle implements BundleBuilder {

  @Override
  public Iterable<Dag> getDags() {
    return List.of(MyDagBuilder.build(), MyFailDagBuilder.build());
  }

  public static void main(String[] args) {
    Server.create(args).serve(new MyBundle().build());
  }
}

Build and deploy JARs

Build the app JAR and stage all runtime dependencies:

cd /files/aip108-java-dag
./gradlew :app:installDist
# Output: app/build/install/app/lib/  (app.jar + all runtime dependency JARs)

Verify the manifests contain the required attributes. Main-Class must be in the app JAR; Airflow-Supervisor-Schema-Version is carried by the SDK JAR:

unzip -p app/build/install/app/lib/app.jar META-INF/MANIFEST.MF
# Expected: Main-Class: com.example.airflow.MyBundle

unzip -p app/build/install/app/lib/airflow-sdk-*.jar META-INF/MANIFEST.MF
# Expected: Airflow-Supervisor-Schema-Version: <date string>

Copy all JARs to a location the Airflow worker can read:

mkdir -p /files/java-jars
cp app/build/install/app/lib/*.jar /files/java-jars/

Configure Airflow

Add to $AIRFLOW_HOME/airflow.cfg (or set the equivalent AIRFLOW__ environment variables):

[sdk]
coordinators = {
  "java": {
    "classpath": "airflow.sdk.coordinators.java.JavaCoordinator",
    "kwargs": {
      "jars_root": ["/files/java-jars"]
    }
  }
}

queue_to_coordinator = {"sdk": "java"}

Integration test cases

Test Case ID

Test Objective

Test Steps

Prerequisite

Expected Outcome

Execution Status

JV-01

(P0) Java task that returns normally reaches TI state success

Trigger aip108_e2e; wait for succeed_task TI.

JAR deployed; coordinator configured

TI state = success; task log contains "succeed_task: returning normally"


JV-02

(P0) Java task that throws RuntimeException reaches TI state failed

Trigger aip108_fail_e2e; wait for fail_task TI.

—

TI state = failed; task log contains "deliberate failure" and a Java stack trace; DAG run of aip108_e2e is unaffected


JV-03

(P0) Java task that throws Error reaches TI state failed (JVM crashes; worker detects exit)

Trigger aip108_fail_e2e; wait for panic_task TI.

—

TI state = failed; task log contains "deliberate panic"; JVM process exits abnormally (the SDK runner only catches Exception, so the Error is uncaught; the worker detects the exit and marks the TI failed)


JV-04

(P0) Java task reads its runtime context via an injected Context parameter (get_current_context equivalent)

Trigger the full aip108_e2e DAG run; wait for context_task TI.

—

TI state = success; task log contains "context_task: dag_id = aip108_e2e", "task_id = context_task", and a non-empty run_id


JV-05

(P0) Java task reads a Variable via the coordinator

Set aip108_test_var = hello in Airflow Admin (or via airflow variables set aip108_test_var hello); trigger variable_task.

Variable present

TI state = success; task log contains "got variable value = hello"


JV-06

(P0) Java task reads a Connection via the coordinator

Create connection aip108_test_conn (type HTTP, host example.com); trigger connection_task.

Connection present

TI state = success; task log contains "host = example.com"


JV-07

(P0) Java task pushes XCom; downstream Python task reads it

Trigger the full aip108_e2e DAG run; wait for python_downstream.

—

python_downstream TI state = success; log contains "Cross-language XCom verified: hello_from_java"


JV-08

(P0) Java task reads XCom written by an upstream Python task

Trigger the full DAG run; wait for xcom_task.

—

Task log contains "pulled upstream XCom = hello_from_python"


JV-09

(P0) Sequential client round-trips within a task are matched by request/response ID

Trigger aip108_e2e; wait for xcom_task TI. The annotation-generated wrapper issues GetXCom (upstream pull via @Builder.XCom) then SetXCom (return value) as two back-to-back comm socket exchanges.

—

Both round-trips return correct values; no cross-contamination; TI state = success; log contains "pulled upstream XCom = hello_from_python" and "pushed XCom 'hello_from_java'"


JV-10

(P1) Missing task in bundle results in TI state removed

Edit aip108_e2e.py to add @task.stub(queue="sdk") def ghost_task(): ... without adding the corresponding task to Java; trigger ghost_task.

—

TI state = removed; JVM exits 0; no crash or hang


JV-11

(P1) Missing variable raises an error in the task

Remove variable aip108_test_var; trigger variable_task.

Variable absent

TI state = failed; task log contains "variable 'aip108_test_var' not found"


JV-12

(P1) Missing connection raises an error in the task

Remove connection aip108_test_conn; trigger connection_task.

Connection absent

TI state = failed; task log contains "connection 'aip108_test_conn' not found"


JV-13

(P1) JAR not on classpath -> coordinator raises at startup

Set jars_root to an empty directory; trigger any Java task.

Empty jars_root

Worker logs a FileNotFoundError naming the missing JAR; TI state = failed


JV-14

(P1) No JAR with Airflow-Supervisor-Schema-Version in jars_root -> coordinator rejects

Point jars_root at that directory and trigger any Java task.

jars_root with app.jar (no airflow-sdk-*.jar)

Coordinator raises FileNotFoundError mentioning Airflow-Supervisor-Schema-Version; TI state = failed; JVM never launched


2. Testing the Go SDK

Prerequisites

Requirement

Version

Go

1.24+

airflow-go-pack

HEAD (see below)

Airflow worker

HEAD

Install airflow-go-pack:

cd <airflow-repo-root>/go-sdk
go install ./cmd/airflow-go-pack
# Binary lands in $(go env GOPATH)/bin/airflow-go-pack

Create a new Go module

mkdir /files/aip108-go-dag && cd /files/aip108-go-dag
go mod init example.com/aip108-go-dag

The Go SDK has no tagged release yet, but it is a normal nested module (github.com/apache/airflow/go-sdk) in a public repo, so go get resolves it by commit. Pin the same commit your Airflow worker is built from so the bundle's supervisor_schema_version matches what the coordinator expects:

go get github.com/apache/airflow/go-sdk@<commit-hash>

go get records a pseudo-version (v0.0.0-<date>-<commit>) in go.mod and the checksum in go.sum, so the build is reproducible without a local checkout. If instead you are iterating on uncommitted local SDK changes, swap the go get for a replace at the local tree (go mod edit -replace=github.com/apache/airflow/go-sdk=<airflow-repo-root>/go-sdk).

After writing main.go (below), run go mod tidy to pull the indirect dependencies.

Write the tasks

A Go bundle is a single package main. RegisterDags is the source of truth for the bundle's dag_id/task_ids, and AddTaskWithName pins each task to the task_id declared in the Python stub. Task functions are plain Go funcs; the runtime injects arguments by type (context.Context, *slog.Logger, an sdk.Client or a narrower client interface, or a tagged struct for typed XCom input). Returning a non-nil error fails the task; a first return value is pushed as its return-value XCom. sdk.CurrentContext(ctx) exposes the runtime context (GO-13) and xcom:"<task_id>" struct tags pull typed upstream XComs (GO-14).

Both DAGs live in one bundle. Create main.go:

package main

import (
	"context"
	"fmt"
	"log"
	"log/slog"

	v1 "github.com/apache/airflow/go-sdk/bundle/bundlev1"
	"github.com/apache/airflow/go-sdk/bundle/bundlev1/bundlev1server"
	"github.com/apache/airflow/go-sdk/sdk"
)

var (
	bundleName    = "aip108_go_e2e"
	bundleVersion = "0.0"
)

type myBundle struct{}

var _ v1.BundleProvider = (*myBundle)(nil)

func (m *myBundle) GetBundleVersion() v1.BundleInfo {
	return v1.BundleInfo{Name: bundleName, Version: &bundleVersion}
}

func (m *myBundle) RegisterDags(dagbag v1.Registry) error {
	e2e := dagbag.AddDag("aip108_e2e")
	e2e.AddTaskWithName("succeed_task", succeedTask)
	e2e.AddTaskWithName("variable_task", variableTask)
	e2e.AddTaskWithName("connection_task", connectionTask)
	e2e.AddTaskWithName("xcom_task", xcomTask)
	e2e.AddTaskWithName("context_task", contextTask)
	e2e.AddTaskWithName("taskflow_task", taskflowTask)

	fail := dagbag.AddDag("aip108_fail_e2e")
	fail.AddTaskWithName("fail_task", failTask)
	fail.AddTaskWithName("panic_task", panicTask)
	return nil
}

func main() {
	if err := bundlev1server.Serve(&myBundle{}); err != nil {
		log.Fatal(err)
	}
}

func succeedTask(log *slog.Logger) error {
	log.Info("succeed_task: returning nil")
	return nil
}

func variableTask(ctx context.Context, client sdk.VariableClient, log *slog.Logger) error {
	val, err := client.GetVariable(ctx, "aip108_test_var")
	if err != nil {
		return err
	}
	log.InfoContext(ctx, "got variable", "value", val)
	return nil
}

func connectionTask(ctx context.Context, client sdk.ConnectionClient, log *slog.Logger) error {
	conn, err := client.GetConnection(ctx, "aip108_test_conn")
	if err != nil {
		return err
	}
	log.InfoContext(ctx, "got connection", "host", conn.Host)
	return nil
}

func xcomTask(ctx context.Context, client sdk.XComClient, log *slog.Logger) (any, error) {
	rc := sdk.CurrentContext(ctx)
	upstream, err := client.GetXCom(ctx, rc.TI.DagID, rc.TI.RunID, "python_upstream", nil, "return_value", nil)
	if err != nil {
		return nil, err
	}
	log.InfoContext(ctx, "pulled upstream XCom", "value", upstream)
	return "hello_from_go", nil
}

func contextTask(ctx context.Context, log *slog.Logger) error {
	rc := sdk.CurrentContext(ctx)
	log.InfoContext(ctx, "context_task",
		"dag_id", rc.TI.DagID,
		"run_id", rc.TI.RunID,
		"task_id", rc.TI.TaskID,
		"try_number", rc.TI.TryNumber)
	return nil
}

type UpstreamInput struct {
	FromPython string `xcom:"python_upstream"`
}

func taskflowTask(ctx context.Context, log *slog.Logger, in UpstreamInput) error {
	log.InfoContext(ctx, "taskflow typed input", "from_python", in.FromPython)
	return nil
}

func failTask() error {
	return fmt.Errorf("deliberate failure - expected in test GO-02")
}

func panicTask() error {
	panic("deliberate panic - expected in test GO-03")
}

Build the bundle

airflow-go-pack builds the package, execs the binary to capture its manifest, and appends the source + manifest + AFBNDL01 trailer. The root command is the packer (there is no pack subcommand); --output names the result:

A packed bundle is a plain executable identified by its AFBNDL01 trailer, not by a filename suffix, so it carries no file extension (the packer adds .exe only on Windows). Name the output accordingly:

cd /files/aip108-go-dag
airflow-go-pack . --output /tmp/aip108-go-e2e
# Wrote bundle /tmp/aip108-go-e2e (sdk=go/<version>, dags=2)

Inspect the bundle to confirm both DAGs and their tasks are registered. sdk.version reflects the linked SDK build: a v0.0.0-<date>-<commit> pseudo-version for a go get-pinned build, or (devel) for a local replace build:

airflow-go-pack inspect /tmp/aip108-go-e2e

airflow_bundle_metadata_version: "1.0"
sdk:
  language: "go"
  version: "(devel)"
  supervisor_schema_version: "2026-06-16"
source: "main.go"
dags:
  aip108_e2e:
    tasks:
      - "succeed_task"
      - "variable_task"
      - "connection_task"
      - "xcom_task"
      - "context_task"
      - "taskflow_task"
  aip108_fail_e2e:
    tasks:
      - "fail_task"
      - "panic_task"

Copy the bundle to a location the Airflow worker can read (still no extension):

mkdir -p /files/go-bundles
cp /tmp/aip108-go-e2e /files/go-bundles/

Cross-platform builds (develop on macOS, deploy on Linux)

The bundle is a native executable, so it must be built for the worker's OS/arch, not the developer's. On an Apple-silicon Mac targeting a linux/amd64 worker, pass -goos/-goarch to the packer:

airflow-go-pack --goos linux --goarch amd64 . --output /tmp/aip108-go-e2e

airflow-go-pack runs the freshly built binary to capture its manifest, so it cross-builds the deployable artifact and builds a throwaway host-arch binary solely for that introspection step. Use the -goos/-goarch flags rather than the GOOS/GOARCH environment variables: env vars would instead cross-build a host-unrunnable packer (and break the introspection step). To pack a pre-built cross binary instead, capture its manifest on the target platform (./bundle --airflow-metadata > meta.yaml) and pass --executable ./bundle --source main.go --airflow-metadata meta.yaml.

Configure Airflow

Add to $AIRFLOW_HOME/airflow.cfg :

[sdk]
coordinators = {
    "go": {
        "classpath": "airflow.sdk.coordinators.executable.ExecutableCoordinator",
        "kwargs": {
            "executables_root": ["/files/go-bundles"]
        }
    }
}

queue_to_coordinator = {"sdk": "go"}

Integration test cases

Test Case ID

Test Objective

Test Steps

Prerequisite

Expected Outcome

Execution Status

GO-01

(P0) Go task that returns nil reaches TI state success

Trigger aip108_e2e; wait for succeed_task TI.

Bundle deployed; coordinator configured

TI state = success; task log contains "succeed_task: returning nil"


GO-02

(P0) Go task that returns an error reaches TI state failed

Trigger aip108_fail_e2e; wait for fail_task TI.

—

TI state = failed; task log contains "deliberate failure"; DAG run of aip108_e2e is unaffected


GO-03

(P0) Panicking Go task reaches TI state failed (panic recovered)

Trigger aip108_fail_e2e; wait for panic_task TI.

—

TI state = failed; task log contains "deliberate panic"; bundle exits 0 (panic is caught by the runner, not the OS)


GO-04

(P0) Go task reads its runtime context via sdk.CurrentContext (get_current_context equivalent)

Trigger the full aip108_e2e DAG run; wait for context_task TI.

—

TI state = success; task log contains "context_task" with dag_id=aip108_e2e, task_id=context_task, and a non-empty run_id


GO-05

(P0) Go task reads a Variable via the coordinator

Set aip108_test_var = hello (airflow variables set aip108_test_var hello); trigger variable_task.

Variable present

TI state = success; task log contains "got variable" value=hello


GO-06

(P0) Go task reads a Connection via the coordinator

Create connection aip108_test_conn (type HTTP, host example.com); trigger connection_task.

Connection present

TI state = success; task log contains "host=example.com"


GO-07

(P0) Go task pushes XCom; downstream Python task reads it

Trigger the full aip108_e2e DAG run; wait for python_downstream.

—

python_downstream TI state = success; log contains "Cross-language XCom verified: hello_from_go"


GO-08

(P0) Go task reads XCom written by an upstream Python task

Trigger the full DAG run; wait for xcom_task.

—

Task log contains "pulled upstream XCom" value=hello_from_python


GO-09

(P0) Concurrent requests from Go task are matched by response ID

Trigger xcom_task: it issues a GetXCom (upstream pull) immediately followed by the return-value PushXCom, exercising request/response ID correlation over the comm socket.

—

Both round-trips return correct values; no cross-contamination; xcom_task TI state = success


GO-10

(P1) Missing task in bundle results in TI state removed

Edit aip108_e2e.py to add @task.stub(queue="sdk") def ghost_task(): ... without adding the corresponding task to Go; trigger ghost_task.

—

TI state = removed; bundle exits 0; no crash or hang


GO-11

(P0) Env var shadows coordinator for Variable reads

Set env var AIRFLOW_VAR_AIP108_TEST_VAR=from_env on the worker; trigger variable_task.

Env var set

TI state = success; task log shows "from_env" without a supervisor round-trip


GO-12

(P1) Missing variable raises an error in the task

Remove variable aip108_test_var; trigger variable_task.

Variable absent

TI state = failed; task log contains "Task failed" with variable not found: "aip108_test_var"


GO-13

(P1) Missing connection raises an error in the task

Remove connection aip108_test_conn; trigger connection_task.

Connection absent

TI state = failed; task log contains "Task failed" with connection not found: "aip108_test_conn"


GO-14

(P1) Bundle binary not on executables_root -> coordinator raises

Set executables_root to an empty directory; trigger any Go task.

Empty executables_root

Worker logs FileNotFoundError naming the missing bundle; TI state = failed


GO-15

(P1) Bundle with missing supervisor_schema_version -> coordinator rejects

Build a bundle by hand without the schema version field in the manifest YAML; place it in executables_root.

Bad bundle present

Coordinator raises FileNotFoundError mentioning the missing field; TI state = failed


GO-16

(P1) Tampered bundle (hash mismatch) -> coordinator rejects

Flip one byte in the packed bundle binary; place it in executables_root.

Tampered bundle (see below)

Coordinator raises an integrity error; TI state = failed; original bundle not affected


GO-16 setup: how to tamper the bundle

Back up the original, then flip one byte at a known offset well within the binary payload (offset 100 is safe; the metadata footer is at the end of the file).

Back the original up outside executables_root. The coordinator scans every executable under that directory, so a valid backup left alongside the tampered file would just be picked instead, masking the rejection.

# Back up the original outside the scan directory so it can be restored later
cp /files/go-bundles/aip108-go-e2e /tmp/aip108-go-e2e.orig

# Flip byte 100 in-place
python3 -c "
from pathlib import Path
data = bytearray(Path('/files/go-bundles/aip108-go-e2e').read_bytes())
data[100] ^= 0xFF
Path('/files/go-bundles/aip108-go-e2e').write_bytes(data)
"

Trigger any Go task and verify the coordinator rejects it. Then restore:

mv /tmp/aip108-go-e2e.orig /files/go-bundles/aip108-go-e2e



  • No labels