Skip to content

Add SubprocessCoordinator - #67635

Merged
jason810496 merged 3 commits into
apache:mainfrom
jason810496:refactor/coordinator/add-socket-coordinator
Jun 3, 2026
Merged

Add SubprocessCoordinator#67635
jason810496 merged 3 commits into
apache:mainfrom
jason810496:refactor/coordinator/add-socket-coordinator

Conversation

@jason810496

@jason810496jason810496 commented May 28, 2026

Copy link
Copy Markdown
Member

Why

The upcoming ExecutableCoordinator (#67161,
launching any compiled-language executable as a subprocess) for the Go SDK starts its
subprocess the exact same way as JavaCoordinator. This introduces a common
SubprocessCoordinator base that both JavaCoordinator and ExecutableCoordinator
can inherit.

What

  • Add SubprocessCoordinator (in task-sdk/src/airflow/sdk/coordinators/_subprocess.py)
    and its unit tests.
    • Subclasses override
      def _build_execute_task_command(self, *, what: TaskInstanceDTO) -> tuple[list[str], str | None]:
    • The returned list[str] is the subprocess command args, without the
      --comm / --logs flags — the base appends those once the listening sockets
      are bound.
    • The returned str is the subprocess supervisor schema version (None disables
      schema migration).
  • Make JavaCoordinator inherit SubprocessCoordinator.

The base is named after the subprocess it launches rather than the TCP transport (an
implementation detail subclasses don't touch); the socket-specific helper is
_PopenActivitySubprocess.


Was generative AI tooling used to co-author this PR?
  • Yes (please specify the tool below) with help of Claude Opus 4.7 following the guidelines

@jason810496jason810496 added the area:coordinator Coordinator: The interface to spawn Lang-SDK subprocesses label May 28, 2026
@jason810496
jason810496 requested a review from CopilotMay 28, 2026 07:45
@jason810496jason810496 moved this from Backlog to In review in Java-SDKMay 28, 2026

CopilotAI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Pull request overview

This PR introduces a shared, socket-based coordinator scaffold in the Task SDK so multiple language runtimes (Java now, compiled executables next) can reuse the same subprocess + --comm/--logs TCP wiring instead of duplicating it in each coordinator.

Changes:

  • Added SocketCoordinator and _SocketActivitySubprocess to centralize TCP server setup, subprocess launch, connection accept/drain, and teardown behavior.
  • Refactored JavaCoordinator to inherit from SocketCoordinator and provide only the per-task command + schema version resolution.
  • Moved/added unit tests for the shared socket machinery and adjusted Java coordinator tests accordingly.

Reviewed changes

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

Show a summary per file
FileDescription
task-sdk/src/airflow/sdk/coordinators/socket/coordinator.pyAdds the shared socket-based coordinator and subprocess plumbing.
task-sdk/src/airflow/sdk/coordinators/socket/init.pyExposes the new socket coordinator package.
task-sdk/src/airflow/sdk/coordinators/java/coordinator.pyRefactors Java coordinator to subclass SocketCoordinator.
task-sdk/tests/task_sdk/coordinators/socket/test_coordinator.pyAdds unit tests covering socket server startup, accept/drain logic, and base coordinator behavior.
task-sdk/tests/task_sdk/coordinators/socket/init.pyIntroduces the socket coordinators test package.
task-sdk/tests/task_sdk/coordinators/java/test_coordinator.pyRemoves duplicated socket-plumbing tests and retargets patches to the shared socket module.

Comment threadtask-sdk/src/airflow/sdk/coordinators/socket/__init__.py Outdated
Comment threadtask-sdk/tests/task_sdk/coordinators/test_subprocess.py
Comment threadtask-sdk/tests/task_sdk/coordinators/test_subprocess.py
Comment threadtask-sdk/tests/task_sdk/coordinators/test_subprocess.py
@jason810496
jason810496 marked this pull request as ready for review May 28, 2026 08:05
@jason810496
jason810496 requested a review from uranusjrMay 28, 2026 08:05
@jason810496jason810496 self-assigned this May 28, 2026
Comment threadtask-sdk/src/airflow/sdk/coordinators/_subprocess.py
Comment threadtask-sdk/src/airflow/sdk/coordinators/socket/coordinator.py Outdated
Comment threadtask-sdk/src/airflow/sdk/coordinators/socket/coordinator.py Outdated
Comment threadtask-sdk/tests/task_sdk/coordinators/socket/test_coordinator.py Outdated
@jason810496
jason810496force-pushed the refactor/coordinator/add-socket-coordinator branch from 6e42582 to 5b4a5b7CompareMay 30, 2026 10:39
@jason810496jason810496 changed the title Add SocketCoordinatorAdd SubprocessCoordinatorMay 30, 2026
@jason810496

Copy link
Copy Markdown
MemberAuthor

Thanks TP and Kaxil for the review!

I just made the following changes, please feel free to give better naming suggestion, thanks.

  • task-sdk/src/airflow/sdk/coordinators/socket/coordinator.py -> task-sdk/src/airflow/sdk/coordinators/_subprocess.py
  • _SocketActivitySubprocess -> _PopenActivitySubprocess
  • SocketCoordinator -> SubprocessCoordinator

@jason810496
jason810496 requested review from kaxil and uranusjrMay 31, 2026 07:50
@jason810496
jason810496force-pushed the refactor/coordinator/add-socket-coordinator branch from 5b4a5b7 to 3650b2dCompareMay 31, 2026 09:06
…ransport
- Rename SocketCoordinator -> SubprocessCoordinator and _SocketActivitySubprocess
-> _PopenActivitySubprocess: the TCP transport is an implementation detail that
subclasses don't care about, while the subprocess they launch is the point.
- Flatten the coordinators/socket/ package into a single private
coordinators/_subprocess.py (also drops the stray subpackage __version__).
- Spec the mock_client test fixture with Client / TaskInstanceOperations so the
client call surface is checked.

@jason810496jason810496 left a comment

Copy link
Copy Markdown
MemberAuthor

Choose a reason for hiding this comment

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

Hi @uranusjr , @kaxil it would be nice to have review when you have a moment. Thanks!
Since the approved #67161 PR still depends on this one.

@kaxilkaxil 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.

LGTM. Clean extraction of the shared subprocess/socket machinery into a base class, with the Java-specific logic (JAR scan, manifest parse, classpath) kept in java/coordinator.py. Mock spec'ing on the test fixture is addressed.

@jason810496
jason810496 merged commit 23a3c33 into apache:mainJun 3, 2026
145 checks passed
@github-project-automationgithub-project-automationBot moved this from In review to Done in Java-SDKJun 3, 2026
@github-project-automationgithub-project-automationBot moved this from In review to Done in AIP-72 (addendum): Go-SDKJun 3, 2026
@phanikumvphanikumv added this to the Airflow 3.3.0 milestone Jun 5, 2026
Sign up for freeto join this conversation on GitHub. Already have an account? Sign in to comment

Labels

area:coordinatorCoordinator: The interface to spawn Lang-SDK subprocessesarea:task-sdk

Projects

Status: Done
Status: Done

Development

Successfully merging this pull request may close these issues.

6 participants

@jason810496@uranusjr@kaxil@vatsrahul1001@phanikumv