Skip to content

Let a Dag importer find the coordinator that parses its files - #74042

Merged
jason810496 merged 4 commits into
mainfrom
jason/lang-sdk-e2e/03c-coordinator-dag-parsing
Oct 5, 2026
Merged

jason810496 merged 4 commits into
mainfrom
jason/lang-sdk-e2e/03c-coordinator-dag-parsing

Conversation

@jason810496

@jason810496 jason810496 commented Oct 1, 2026 •

Copy link
Copy Markdown
Member

Stack (bottom to top): #74042, #74035, #74043, #74036, #74037, #73845, #73846, #73847

Why

To parse native Dags, the Dag importer that claims a file has to find the coordinator that parses it. It looks in [sdk] coordinators, so a runtime is still declared in one place.

What changes

  • CoordinatorDagImporter is the base class for a runtime's Dag importer. A subclass sets coordinator_classpath, and get_parsing_coordinator() returns the coordinator of that class that parses its files in its Dag bundle.
  • With no coordinator of that class, the importer is not registered. With one, it parses the importer's files in every Dag bundle. With several, the new [sdk] dag_bundle_to_coordinator must map the bundle to one of them, or get_parsing_coordinator() raises InvalidCoordinatorError.
  • DagImporterRegistry.from_config(bundle_name) registers each importer in COORDINATOR_DAG_IMPORTERS that has a coordinator. The tuple is empty until Parse native Java Dags and show their source #74036 and Parse native TypeScript Dags with the Node coordinator #74037 add the Java and Node importers.
  • SubprocessCoordinator.parse_dag(...) replaces the calling process with the runtime that parses one Dag file.
  • A Lang-SDK runtime gets the resolved [api] base_url, [operators] default_deferrable and [triggerer] queues_enabled in its environment, like the log levels.

With two JavaCoordinators, jdk-11 and jdk-17, tasks still follow their queue and the new option picks which one parses each Dag bundle:

[sdk]
queue_to_coordinator = {"legacy-java": "jdk-11", "modern-java": "jdk-17"}
dag_bundle_to_coordinator = {"dags-folder": "jdk-17"}

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

Comment thread task-sdk/src/airflow/sdk/execution_time/coordinator.py Outdated
Comment thread task-sdk/src/airflow/sdk/execution_time/coordinator.py Outdated
Comment thread task-sdk/src/airflow/sdk/execution_time/coordinator.py Outdated
Comment thread task-sdk/src/airflow/sdk/importers/base.py Outdated
Comment thread task-sdk/tests/task_sdk/execution_time/test_coordinator.py Outdated
@jason810496
jason810496 force-pushed the jason/lang-sdk-e2e/03c-coordinator-dag-parsing branch 2 times, most recently from 107f1eb to e3fc440 Compare October 2, 2026 04:07
@jason810496
jason810496 force-pushed the jason/lang-sdk-e2e/03c-coordinator-dag-parsing branch from e3fc440 to 3018722 Compare October 2, 2026 05:44
@pierrejeambrun
pierrejeambrun force-pushed the jason/lang-sdk-e2e/03c-coordinator-dag-parsing branch from 3018722 to d21aea9 Compare October 2, 2026 09:00
@jason810496
jason810496 force-pushed the jason/lang-sdk-e2e/03c-coordinator-dag-parsing branch from d21aea9 to 42c00d4 Compare October 2, 2026 11:41
@jason810496
jason810496 marked this pull request as ready for review October 2, 2026 12:28

@jason810496 jason810496 left a comment

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for the reivew.

Comment thread task-sdk/src/airflow/sdk/execution_time/coordinator.py Outdated
Comment thread task-sdk/src/airflow/sdk/execution_time/coordinator.py Outdated
Comment thread task-sdk/src/airflow/sdk/execution_time/coordinator.py Outdated
Comment thread task-sdk/src/airflow/sdk/importers/base.py Outdated
Comment thread task-sdk/tests/task_sdk/execution_time/test_coordinator.py Outdated
@jason810496
jason810496 force-pushed the jason/lang-sdk-e2e/03c-coordinator-dag-parsing branch from 42c00d4 to 63a5cd3 Compare October 2, 2026 14:07
@jason810496
jason810496 force-pushed the jason/lang-sdk-e2e/03c-coordinator-dag-parsing branch from 63a5cd3 to b830155 Compare October 2, 2026 15:09
@jason810496
jason810496 removed this pull request from stack #74044 October 3, 2026 17:58
@jason810496
jason810496 force-pushed the jason/lang-sdk-e2e/03c-coordinator-dag-parsing branch from b830155 to b9fcd45 Compare October 3, 2026 17:58
@jason810496
jason810496 added this pull request to stack #74170 October 3, 2026 17:59
@jason810496
jason810496 force-pushed the jason/lang-sdk-e2e/03c-coordinator-dag-parsing branch from b9fcd45 to 3dfc4d4 Compare October 4, 2026 01:06
@jason810496
jason810496 force-pushed the jason/lang-sdk-e2e/03c-coordinator-dag-parsing branch from 3dfc4d4 to 1e177e6 Compare October 4, 2026 08:51
@jason810496 jason810496 changed the title Let a coordinator parse the Dag files of the bundles it serves Let a Dag importer find the coordinator that parses its files Oct 4, 2026
Base automatically changed from jason/lang-sdk-e2e/03b-serialized-dag-validation to main October 5, 2026 03:38
A CoordinatorDagImporter names its coordinator class in
coordinator_classpath and finds that coordinator in [sdk] coordinators.
With several coordinators of the class, the new optional
[sdk] dag_bundle_to_coordinator picks the one for a Dag bundle. A Dag
bundle's importer registry registers the importers listed in
COORDINATOR_DAG_IMPORTERS that have a coordinator.

SubprocessCoordinator.parse_dag replaces the calling process with the
runtime that parses one Dag file, and the runtime gets the resolved
[api] base_url, [operators] default_deferrable and [triggerer]
queues_enabled in its environment.
@jason810496
jason810496 force-pushed the jason/lang-sdk-e2e/03c-coordinator-dag-parsing branch 2 times, most recently from aef2336 to 6fd1704 Compare October 5, 2026 05:20
Comment thread task-sdk/src/airflow/sdk/coordinators/_subprocess.py Outdated
Comment thread task-sdk/src/airflow/sdk/execution_time/coordinator.py

@uranusjr uranusjr left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Two minor points needing resolution. Generally looks good to me.

jason810496 and others added 2 commits October 5, 2026 06:33
Python's signal module never exposes SIGXFZ, so the entry was a no-op.

Co-Authored-By: Claude <noreply@anthropic.com>
A configured coordinator that cannot be loaded is skipped, while a classpath from code that cannot be
imported raises ImportError.

Co-Authored-By: Claude <noreply@anthropic.com>
@jason810496
jason810496 merged commit e0202f4 into main Oct 5, 2026
108 checks passed
@jason810496
jason810496 deleted the jason/lang-sdk-e2e/03c-coordinator-dag-parsing branch October 5, 2026 09:12
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

area:coordinator Coordinator: The interface to spawn Lang-SDK subprocesses area:task-sdk

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants