Add e2e test suite for Airflow event-driven DAGs with Apache Kafka

[jason810496]

Why

Airflow 3 introduces event-driven DAGs with Assets, AssetWatchers, and MessageQueueTrigger to enable reactive, message-driven pipelines. This PR adds a comprehensive e2e
test suite to verify that Kafka-based event routing works correctly, including proper handling of malformed messages and dead-letter queues.

What

• Dag pattern: Implemented a producer-consumer pattern where producer sends 9 messages (8 valid orders + 1 malformed) to a Kafka topic, and consumer triggers 9 times via AssetWatcher
• Test verification: Consumer Dag uses retries (3), short-circuit on malformed messages, and routes failures to a dead-letter queue topic
• CI integration: Extended selective-checks to detect event-driven test file changes and trigger the suite; added workflow inputs/outputs for test mode selection
• Kafka integration: The kafka.yaml and update_run.sh are copied from breeze --integration kafka

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

• Yes (please specify the tool below) Claude Code following the guidelines

[Copilot]

Pull request overview

Note

Copilot was unable to run its full agentic suite in this review.

Adds an event-driven (Kafka-backed) E2E test suite for Airflow 3 Assets/AssetWatchers and wires it into Breeze selective checks + CI workflows so the suite runs when relevant files change.

Changes:

• Introduces Kafka producer/consumer example DAGs and a new E2E test validating triggers, retries/DLQ behavior, and Kafka offsets.
• Adds Kafka docker-compose + init script and updates the E2E harness to spin up Kafka and create required topics.
• Extends Breeze selective checks and GitHub workflows to support an event_driven E2E test mode.

Reviewed changes

Copilot reviewed 14 out of 15 changed files in this pull request and generated 7 comments.

Show a summary per file

File	Description
dev/breeze/src/airflow_breeze/utils/selective_checks.py	Adds a new selective-checks file group + toggle for event-driven E2E tests.
dev/breeze/src/airflow_breeze/commands/testing_commands.py	Extends --e2e-test-mode choices with event_driven.
dev/breeze/doc/images/output_testing_airflow-e2e-tests.txt	Updates generated CLI help output artifact.
dev/breeze/doc/images/output_testing_airflow-e2e-tests.svg	Updates generated CLI help output artifact (includes event_driven).
airflow-e2e-tests/tests/airflow_e2e_tests/event_driven_tests/test_event_driven.py	Adds Kafka event-driven E2E test validating DAG triggers + Kafka offsets.
airflow-e2e-tests/tests/airflow_e2e_tests/event_driven_tests/init.py	Adds package init for the new test module.
airflow-e2e-tests/tests/airflow_e2e_tests/e2e_test_utils/clients.py	Adds list_dag_runs helper used by the new test.
airflow-e2e-tests/tests/airflow_e2e_tests/dags/example_event_driven.py	Adds example producer/consumer event-driven DAGs (Assets + Kafka trigger + DLQ).
airflow-e2e-tests/tests/airflow_e2e_tests/constants.py	Adds Kafka docker directory path constant.
airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py	Adds event_driven mode setup, Kafka file copying, and topic creation.
airflow-e2e-tests/docker/kafka/update_run.sh	Adds broker init script for Kafka container startup.
airflow-e2e-tests/docker/kafka.yml	Adds Kafka broker docker-compose definition for E2E runs.
.github/workflows/ci-amd-arm.yml	Plumbs through the new selective-check output for event-driven E2E tests.
.github/workflows/airflow-e2e-tests.yml	Updates workflow input docs to include event_driven mode.
.github/workflows/additional-prod-image-tests.yml	Adds a new job to run event-driven E2E tests when selected.

[Copilot]

Pull request overview

Copilot reviewed 15 out of 16 changed files in this pull request and generated 6 comments.

[jason810496]

The CI failure require additional PR to fix the AwaitTrigger first:

 airflow-triggerer-1 | 2026-05-02T15:25:00.831436Z [error ] Trigger ID 1 exited with error 'AwaitMessageTrigger' object has no attribute '_task_instance' [airflow.jobs.triggerer_job_runner] error_detail=[{'exc_type': 'AttributeError', 'exc_value': "'AwaitMessageTrigger' object has no attribute '_task_instance'", 'exc_notes': [], 'syntax_error': None, 'is_cause': False, 'frames': [{'filename': '/home/airflow/.local/lib/python3.10/site-packages/airflow/jobs/triggerer_job_runner.py', 'lineno': 1238, 'name': 'cleanup_finished_triggers'}, {'filename': '/home/airflow/.local/lib/python3.10/site-packages/greenback/_impl.py', 'lineno': 119, 'name': 'greenback_shim'}, {'filename': '/home/airflow/.local/lib/python3.10/site-packages/greenback/_impl.py', 'lineno': 208, 'name': '_greenback_shim'}, {'filename': '/home/airflow/.local/lib/python3.10/site-packages/greenback/_impl.py', 'lineno': 84, 'name': 'trampoline'}, {'filename': '/home/airflow/.local/lib/python3.10/site-packages/outcome/_impl.py', 'lineno': 185, 'name': 'send'}, {'filename': '/home/airflow/.local/lib/python3.10/site-packages/airflow/jobs/triggerer_job_runner.py', 'lineno': 1387, 'name': 'run_trigger'}, {'filename': '/home/airflow/.local/lib/python3.10/site-packages/airflow/triggers/base.py', 'lineno': 104, 'name': 'task_instance'}], 'is_group': False, 'exceptions': []}] loc=triggerer_job_runner.py:929
airflow-triggerer-1 | 2026-05-02T15:25:00.832067Z [error ] Trigger exited without sending an event. Dependent tasks will be failed. [airflow.jobs.triggerer_job_runner] loc=triggerer_job_runner.py:929 name='ID 1'
airflow-triggerer-1 | 2026-05-02T15:25:02.858280Z [error ] Trigger ID 1 exited with error 'AwaitMessageTrigger' object has no attribute '_task_instance' [airflow.jobs.triggerer_job_runner] error_detail=[{'exc_type': 'AttributeError', 'exc_value': "'AwaitMessageTrigger' object has no attribute '_task_instance'", 'exc_notes': [], 'syntax_error': None, 'is_cause': False, 'frames': [{'filename': '/home/airflow/.local/lib/python3.10/site-packages/airflow/jobs/triggerer_job_runner.py', 'lineno': 1238, 'name': 'cleanup_finished_triggers'}, {'filename': '/home/airflow/.local/lib/python3.10/site-packages/greenback/_impl.py', 'lineno': 119, 'name': 'greenback_shim'}, {'filename': '/home/airflow/.local/lib/python3.10/site-packages/greenback/_impl.py', 'lineno': 208, 'name': '_greenback_shim'}, {'filename': '/home/airflow/.local/lib/python3.10/site-packages/greenback/_impl.py', 'lineno': 84, 'name': 'trampoline'}, {'filename': '/home/airflow/.local/lib/python3.10/site-packages/outcome/_impl.py', 'lineno': 185, 'name': 'send'}, {'filename': '/home/airflow/.local/lib/python3.10/site-packages/airflow/jobs/triggerer_job_runner.py', 'lineno': 1387, 'name': 'run_trigger'}, {'filename': '/home/airflow/.local/lib/python3.10/site-packages/airflow/triggers/base.py', 'lineno': 104, 'name': 'task_instance'}], 'is_group': False, 'exceptions': []}] loc=triggerer_job_runner.py:929

[parkhojeong]

i reviewed about term usage :)

[Copilot]

Pull request overview

Copilot reviewed 15 out of 16 changed files in this pull request and generated 2 comments.

Comments suppressed due to low confidence (1)

dev/breeze/src/airflow_breeze/utils/run_tests.py:170

• run_docker_compose_tests() now adds -sv to pytest args for all docker-compose based suites. This forces verbose + no-capture output in CI (and duplicates -s for the quick-start path), which can significantly increase logs and make failures harder to scan. Consider keeping the previous default and only enabling -s/-v when explicitly requested (e.g., via a CLI flag / env var) or only for specific suites that need streaming output.

 # Always with color
pytest_args = ["--color=yes", "-sv"]
if test_type == "task-sdk-integration":
test_path = Path("tests") / "task_sdk_tests"
cwd = TASK_SDK_INTEGRATION_TESTS_ROOT_PATH.as_posix()
elif test_type == "airflow-e2e-tests":
test_path = Path("tests") / "airflow_e2e_tests" / f"{test_mode}_tests"
cwd = AIRFLOW_E2E_TESTS_ROOT_PATH.as_posix()
elif test_type == "airflow-ctl-integration":
test_path = Path("tests") / "airflowctl_tests"
cwd = AIRFLOW_CTL_TESTS_ROOT_PATH.as_posix()
else:
test_path = Path("tests") / "docker_tests" / "test_docker_compose_quick_start.py"
cwd = DOCKER_TESTS_ROOT_PATH.as_posix()
pytest_args.append("-s") # -s to see print outputs as they come

[Copilot]

Pull request overview

Copilot reviewed 16 out of 17 changed files in this pull request and generated no new comments.

Comments suppressed due to low confidence (1)

dev/breeze/src/airflow_breeze/utils/run_tests.py:169

• run_docker_compose_tests() now unconditionally adds -sv for all docker-compose-based test types (task-sdk integration, airflow-e2e-tests, airflow-ctl integration, and even the quick-start). This changes log volume/capture behavior globally and also makes the later pytest_args.append('-s') for the quick-start redundant. Consider scoping -s/-v to the specific test type(s) that need streaming output (e.g. only airflow-e2e-tests), or wiring this to an explicit Breeze flag/env var so CI log size and output format don’t change unexpectedly.

 # Always with color
pytest_args = ["--color=yes", "-sv"]
if test_type == "task-sdk-integration":
test_path = Path("tests") / "task_sdk_tests"
cwd = TASK_SDK_INTEGRATION_TESTS_ROOT_PATH.as_posix()
elif test_type == "airflow-e2e-tests":
test_path = Path("tests") / "airflow_e2e_tests" / f"{test_mode}_tests"
cwd = AIRFLOW_E2E_TESTS_ROOT_PATH.as_posix()
elif test_type == "airflow-ctl-integration":
test_path = Path("tests") / "airflowctl_tests"
cwd = AIRFLOW_CTL_TESTS_ROOT_PATH.as_posix()
else:
test_path = Path("tests") / "docker_tests" / "test_docker_compose_quick_start.py"
cwd = DOCKER_TESTS_ROOT_PATH.as_posix()
pytest_args.append("-s") # -s to see print outputs as they come

[jscheffl]

Oh, cool!

FYI @renat-sagut

[jscheffl]

Looks great! I assume you master to make CI green. Comments are non blocking just some thoughts. But also easy to apply.