apache / apache/airflow

AIP-108: Re-enable Dag Processor coordinator schema-version pinning

Open
#66,934 0 comments 0 reactions 1 assignee Claimed by @jason810496 View on GitHub
AIP-85: Dag Importer area:coordinator area:core on hold
Dominant language
Python
Stars
46.9k
Forks
17.8k
Avg merge
2d 9h
Merged PRs (30d)
472

Description

### Background

In `airflow-core/src/airflow/dag_processing/processor.py` (around lines 582–590), the Dag Processor's `_on_child_started` has the coordinator-pinning block commented out behind a TODO:

```python
# TODO(jason810496): Uncomment the coordinator-pinning block below once the
# Dag Processor <-> Coordinator interaction pattern is settled and
# ``CoordinatorManager.for_dag_file`` is reintroduced (it was removed in
# astronomer/airflow#1578, which narrowed AIP-108 scope to task execution only).
# When a coordinator handles this file, the pin lets the supervisor's IPC
# migrator downgrade outgoing head-shape bodies and upgrade incoming
# bodies against the lang-SDK's frozen message schema version.
# if (coordinator := get_coordinator_manager().for_dag_file(bundle_name, path)) is not None:
# self.lang_sdk_msg_schema_version = coordinator.target_msg_schema_version(msg)
self.send_msg(msg, request_id=0)
```

The block was disabled because `CoordinatorManager.for_dag_file` was removed in astronomer/airflow#1578, which narrowed AIP-108's initial scope to task execution only. As a result, the Dag Processor currently cannot pin its outgoing IPC frames to a foreign-runtime (lang-SDK) message schema version, so the supervisor's migrator cannot downgrade outgoing head-shape bodies / upgrade incoming bodies for Dag-parsing exchanges with non-Python lang-SDK runtimes.

### What needs to happen

1. Settle the Dag Processor <-> Coordinator interaction pattern for the parsing path (which file extensions / queues route to which coordinator at parse time, and what `StartupDetails`-equivalent message the parser sends).
2. Reintroduce a resolution entry point on `CoordinatorManager` (e.g. `for_dag_file(bundle_name, path)`) or its successor that returns the coordinator handling a given Dag file.
3. Uncomment / re-wire the pinning block in `_on_child_started`, setting `self.lang_sdk_msg_schema_version` from `coordinator.target_msg_schema_version(msg)` before `send_msg`.
4. Add unit tests covering the three cases the existing `TestDagFileParseRequestCoordinatorMigration` parametrization assumes (no coordinator → no pin; coordinator with `target_msg_schema_version` returning a known version → pin set; unknown / invalid manifest version → graceful failure path).
5. Remove or rewrite the existing tests in `test_processor.py` that currently assert on `proc.lang_sdk_msg_schema_version` so they reflect the final behavior.

### Acceptance criteria

- The TODO at `airflow-core/src/airflow/dag_processing/processor.py:582-590` is removed.
- A Dag file routed to a non-Python coordinator results in `lang_sdk_msg_schema_version` being set on the Dag-processor subprocess before `StartupDetails` (or its parsing equivalent) is sent.
- `CoordinatorManager` exposes a public method for Dag-file → coordinator resolution, used by both the parser and (where appropriate) the supervisor.
- Tests in `airflow-core/tests/.../test_processor.py` cover the pinning path end-to-end.

### Context

- Original PR where the TODO was introduced: astronomer/airflow#1577 (mirrored upstream in the AIP-108 work)
- Removal of `CoordinatorManager.for_dag_file`: astronomer/airflow#1578
- Related review thread that requested this tracking issue: https://github.com/astronomer/airflow/pull/1577#discussion_r3240218006

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.