kaxil commented on code in PR #65958: URL: https://github.com/apache/airflow/pull/65958#discussion_r3295487550
########## task-sdk/src/airflow/sdk/coordinators/java/coordinator.py: ########## @@ -0,0 +1,261 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""Java runtime coordinator that launches a JVM subprocess for Dag file processing and task execution.""" + +from __future__ import annotations + +import email +import os +import pathlib +import selectors +import socket +import subprocess +import time +import zipfile +from typing import TYPE_CHECKING, cast + +import attrs +import psutil +import structlog + +from airflow.sdk.execution_time.coordinator import BaseCoordinator +from airflow.sdk.execution_time.supervisor import ActivitySubprocess + +if TYPE_CHECKING: + from collections.abc import Sequence + + from structlog.typing import FilteringBoundLogger + from typing_extensions import Self + + from airflow.sdk.api.client import Client + from airflow.sdk.api.datamodels._generated import BundleInfo + from airflow.sdk.execution_time.workloads.task import TaskInstanceDTO + +log: FilteringBoundLogger = structlog.get_logger(logger_name="coordinators.java") + + +def _start_server() -> socket.socket: + server = socket.socket() + server.bind(("127.0.0.1", 0)) + server.setblocking(True) + server.listen(1) # Just need to listen to the child process. + return server + + +def _calculate_classpath(jars_root: Sequence[pathlib.Path]) -> str: + jars = (p.as_posix() for root in jars_root for p in root.iterdir() if p.suffix == ".jar") + return os.pathsep.join(jars) + + [email protected] +class _MainJar: + path: pathlib.Path + main_class: str + schema_version: str | None + + @classmethod + def find(cls, jars_root: Sequence[pathlib.Path]) -> Self: + for root in jars_root: + for p in root.iterdir(): + if p.suffix != ".jar": + continue + with zipfile.ZipFile(p) as zf: + with zf.open("META-INF/MANIFEST.MF") as f: + manifest = email.message_from_binary_file(f) + if main_class := manifest["Main-Class"]: + return cls(p, main_class, manifest.get("Airflow-SDK-Supervisor-Schema-Version")) + resolved_paths = os.pathsep.join(str(p.resolve()) for p in jars_root) + raise FileNotFoundError(f"cannot fine main class in {resolved_paths}") + + +def _accept_connections( + servers: dict[str, socket.socket], + proc: subprocess.Popen, + *, + max_wait: float = 10.0, +) -> dict[str, socket.socket]: + """Block until the Java process connects to servers.""" + accepted: dict[str, socket.socket] = {} + with selectors.DefaultSelector() as sel: + for key, soc in servers.items(): + sel.register(soc, selectors.EVENT_READ, data=key) + deadline = time.monotonic() + max_wait + while len(accepted) < len(servers): + remaining = deadline - time.monotonic() + if remaining <= 0: + raise TimeoutError("process did not connect within timeout") + if proc.poll() is not None: + raise RuntimeError(f"process exited with {proc.returncode} before connecting") + for event, _ in sel.select(timeout=min(remaining, 1.0)): + log.debug("Accepting child process connection", key=(key := event.data)) + conn, _ = cast("socket.socket", event.fileobj).accept() + sel.unregister(servers[key]) + accepted[key] = conn + return accepted + + [email protected](kw_only=True) +class _JavaActivitySubprocess(ActivitySubprocess): + """Java task runner process.""" + + _comm_server: socket.socket + _logs_server: socket.socket + _subprocess: subprocess.Popen + + # Keep track of channels used to pipe subprocess stdout and stderr so we can + # close them on exit. The "read" side is handled by _register_pipe_readers + # callbacks so we don't need to worry about them. + _stdout_w: socket.socket + _stderr_w: socket.socket + + @classmethod + def start( # type: ignore[override] + cls, + *, + what: TaskInstanceDTO, + dag_rel_path: str | os.PathLike[str], + bundle_info, + logger: FilteringBoundLogger | None = None, + sentry_integration: str = "", + java_executable: str, + jvm_args: list[str], + jars_root: Sequence[pathlib.Path], + **kwargs, + ) -> Self: + jar = _MainJar.find(jars_root) + + comm_server = _start_server() + logs_server = _start_server() + + stdout_r, stdout_w = socket.socketpair() + stderr_r, stderr_w = socket.socketpair() + + comm_host, comm_port = comm_server.getsockname() + logs_host, logs_port = logs_server.getsockname() + + proc = subprocess.Popen( + [ + java_executable, + "-classpath", + _calculate_classpath(jars_root), + *jvm_args, + jar.main_class, + # Arguments to MainClass... + f"--comm={comm_host}:{comm_port}", + f"--logs={logs_host}:{logs_port}", + ], + stdout=stdout_w.makefile("wb", buffering=0).fileno(), + stderr=stderr_w.makefile("wb", buffering=0).fileno(), + ) + log.info("Starting subprocess", pid=proc.pid) + socks = _accept_connections({"comm": comm_server, "logs": logs_server}, proc) Review Comment: The accept loop in `_accept_connections` only watches `comm`/`logs` servers; `stdout_w`/`stderr_w` socketpairs aren't drained until `_register_pipe_readers` runs after this returns. A JVM that writes anything before connecting -- and many do (`WARNING: A restricted method in java.lang.System...`, `Picked up JAVA_TOOL_OPTIONS=...`, slf4j banner, log4j init) -- can fill the socketpair buffer (~16-64 KB depending on platform) and block in `write()` before reaching the `--comm=` connect. The accept loop then times out at 10s with an unhelpful "process did not connect within timeout". Fix options: register the pipe readers before the accept loop, or include the socketpair read-ends in the selector during accept. ########## task-sdk/src/airflow/sdk/coordinators/java/coordinator.py: ########## @@ -0,0 +1,261 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""Java runtime coordinator that launches a JVM subprocess for Dag file processing and task execution.""" + +from __future__ import annotations + +import email +import os +import pathlib +import selectors +import socket +import subprocess +import time +import zipfile +from typing import TYPE_CHECKING, cast + +import attrs +import psutil +import structlog + +from airflow.sdk.execution_time.coordinator import BaseCoordinator +from airflow.sdk.execution_time.supervisor import ActivitySubprocess + +if TYPE_CHECKING: + from collections.abc import Sequence + + from structlog.typing import FilteringBoundLogger + from typing_extensions import Self + + from airflow.sdk.api.client import Client + from airflow.sdk.api.datamodels._generated import BundleInfo + from airflow.sdk.execution_time.workloads.task import TaskInstanceDTO + +log: FilteringBoundLogger = structlog.get_logger(logger_name="coordinators.java") + + +def _start_server() -> socket.socket: + server = socket.socket() + server.bind(("127.0.0.1", 0)) + server.setblocking(True) + server.listen(1) # Just need to listen to the child process. + return server + + +def _calculate_classpath(jars_root: Sequence[pathlib.Path]) -> str: + jars = (p.as_posix() for root in jars_root for p in root.iterdir() if p.suffix == ".jar") + return os.pathsep.join(jars) + + [email protected] +class _MainJar: + path: pathlib.Path + main_class: str + schema_version: str | None + + @classmethod + def find(cls, jars_root: Sequence[pathlib.Path]) -> Self: + for root in jars_root: + for p in root.iterdir(): + if p.suffix != ".jar": + continue + with zipfile.ZipFile(p) as zf: + with zf.open("META-INF/MANIFEST.MF") as f: + manifest = email.message_from_binary_file(f) + if main_class := manifest["Main-Class"]: + return cls(p, main_class, manifest.get("Airflow-SDK-Supervisor-Schema-Version")) + resolved_paths = os.pathsep.join(str(p.resolve()) for p in jars_root) + raise FileNotFoundError(f"cannot fine main class in {resolved_paths}") + + +def _accept_connections( + servers: dict[str, socket.socket], + proc: subprocess.Popen, + *, + max_wait: float = 10.0, +) -> dict[str, socket.socket]: + """Block until the Java process connects to servers.""" + accepted: dict[str, socket.socket] = {} + with selectors.DefaultSelector() as sel: + for key, soc in servers.items(): + sel.register(soc, selectors.EVENT_READ, data=key) + deadline = time.monotonic() + max_wait + while len(accepted) < len(servers): + remaining = deadline - time.monotonic() + if remaining <= 0: + raise TimeoutError("process did not connect within timeout") + if proc.poll() is not None: + raise RuntimeError(f"process exited with {proc.returncode} before connecting") + for event, _ in sel.select(timeout=min(remaining, 1.0)): + log.debug("Accepting child process connection", key=(key := event.data)) + conn, _ = cast("socket.socket", event.fileobj).accept() + sel.unregister(servers[key]) + accepted[key] = conn + return accepted + + [email protected](kw_only=True) +class _JavaActivitySubprocess(ActivitySubprocess): + """Java task runner process.""" + + _comm_server: socket.socket + _logs_server: socket.socket + _subprocess: subprocess.Popen Review Comment: `_subprocess: subprocess.Popen` is declared on `_JavaActivitySubprocess` and assigned via `subprocess=proc` at line 175, but it isn't read anywhere in this file, and the parent `ActivitySubprocess` (in `supervisor.py`) doesn't declare a matching field either -- the parent tracks the child via the `process: psutil.Process` field, not `subprocess.Popen`. Drop the field and the kwarg unless it's intentionally reserved for something downstream -- or wire it into `wait()` cleanup (related to the orphan-JVM cleanup comment on `start()`). ########## task-sdk/src/airflow/sdk/coordinators/java/coordinator.py: ########## @@ -0,0 +1,261 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""Java runtime coordinator that launches a JVM subprocess for Dag file processing and task execution.""" + +from __future__ import annotations + +import email +import os +import pathlib +import selectors +import socket +import subprocess +import time +import zipfile +from typing import TYPE_CHECKING, cast + +import attrs +import psutil +import structlog + +from airflow.sdk.execution_time.coordinator import BaseCoordinator +from airflow.sdk.execution_time.supervisor import ActivitySubprocess + +if TYPE_CHECKING: + from collections.abc import Sequence + + from structlog.typing import FilteringBoundLogger + from typing_extensions import Self + + from airflow.sdk.api.client import Client + from airflow.sdk.api.datamodels._generated import BundleInfo + from airflow.sdk.execution_time.workloads.task import TaskInstanceDTO + +log: FilteringBoundLogger = structlog.get_logger(logger_name="coordinators.java") + + +def _start_server() -> socket.socket: + server = socket.socket() + server.bind(("127.0.0.1", 0)) + server.setblocking(True) + server.listen(1) # Just need to listen to the child process. + return server + + +def _calculate_classpath(jars_root: Sequence[pathlib.Path]) -> str: + jars = (p.as_posix() for root in jars_root for p in root.iterdir() if p.suffix == ".jar") + return os.pathsep.join(jars) + + [email protected] +class _MainJar: + path: pathlib.Path + main_class: str + schema_version: str | None + + @classmethod + def find(cls, jars_root: Sequence[pathlib.Path]) -> Self: + for root in jars_root: + for p in root.iterdir(): Review Comment: `pathlib.Path.iterdir()` order is filesystem-dependent. `_MainJar.find()` returns the *first* JAR with a `Main-Class` and uses *that* JAR's `Airflow-SDK-Supervisor-Schema-Version` (line 81). So filesystem ordering -- not anything Airflow controls -- decides which schema version drives the supervisor's frame migration. Pair this with the schema-version comment above: even if every JAR in `jars_root` carries the header, picking the wrong one still pins the migrator to the wrong version. Two suggestions: 1. Anchor selection on the Airflow header (e.g. "first JAR whose manifest has `Airflow-SDK-Supervisor-Schema-Version`") rather than `Main-Class`, since `Main-Class` is generic to any executable JAR and a third-party utility JAR in the same dir can win. 2. `sorted(root.iterdir())` so classpath ordering and discovery are reproducible across machines. `_calculate_classpath` (line 61) has the same ordering issue. ########## task-sdk/src/airflow/sdk/coordinators/java/coordinator.py: ########## @@ -0,0 +1,261 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""Java runtime coordinator that launches a JVM subprocess for Dag file processing and task execution.""" + +from __future__ import annotations + +import email +import os +import pathlib +import selectors +import socket +import subprocess +import time +import zipfile +from typing import TYPE_CHECKING, cast + +import attrs +import psutil +import structlog + +from airflow.sdk.execution_time.coordinator import BaseCoordinator +from airflow.sdk.execution_time.supervisor import ActivitySubprocess + +if TYPE_CHECKING: + from collections.abc import Sequence + + from structlog.typing import FilteringBoundLogger + from typing_extensions import Self + + from airflow.sdk.api.client import Client + from airflow.sdk.api.datamodels._generated import BundleInfo + from airflow.sdk.execution_time.workloads.task import TaskInstanceDTO + +log: FilteringBoundLogger = structlog.get_logger(logger_name="coordinators.java") + + +def _start_server() -> socket.socket: + server = socket.socket() + server.bind(("127.0.0.1", 0)) + server.setblocking(True) + server.listen(1) # Just need to listen to the child process. + return server + + +def _calculate_classpath(jars_root: Sequence[pathlib.Path]) -> str: + jars = (p.as_posix() for root in jars_root for p in root.iterdir() if p.suffix == ".jar") + return os.pathsep.join(jars) + + [email protected] +class _MainJar: + path: pathlib.Path + main_class: str + schema_version: str | None + + @classmethod + def find(cls, jars_root: Sequence[pathlib.Path]) -> Self: + for root in jars_root: + for p in root.iterdir(): + if p.suffix != ".jar": + continue + with zipfile.ZipFile(p) as zf: + with zf.open("META-INF/MANIFEST.MF") as f: Review Comment: `zf.open("META-INF/MANIFEST.MF")` raises `KeyError` on JARs without a manifest, and `zipfile.ZipFile(p)` raises `BadZipFile` on truncated/corrupt files. One non-standard dependency JAR earlier in `iterdir()` order aborts the whole walk before reaching the valid coordinator JAR. Catch per-JAR (`KeyError`, `zipfile.BadZipFile`) with path context and continue scanning. Optional: log the skipped JAR so it's diagnosable when discovery later raises `FileNotFoundError`. ########## task-sdk/src/airflow/sdk/coordinators/java/coordinator.py: ########## @@ -0,0 +1,261 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""Java runtime coordinator that launches a JVM subprocess for Dag file processing and task execution.""" + +from __future__ import annotations + +import email +import os +import pathlib +import selectors +import socket +import subprocess +import time +import zipfile +from typing import TYPE_CHECKING, cast + +import attrs +import psutil +import structlog + +from airflow.sdk.execution_time.coordinator import BaseCoordinator +from airflow.sdk.execution_time.supervisor import ActivitySubprocess + +if TYPE_CHECKING: + from collections.abc import Sequence + + from structlog.typing import FilteringBoundLogger + from typing_extensions import Self + + from airflow.sdk.api.client import Client + from airflow.sdk.api.datamodels._generated import BundleInfo + from airflow.sdk.execution_time.workloads.task import TaskInstanceDTO + +log: FilteringBoundLogger = structlog.get_logger(logger_name="coordinators.java") + + +def _start_server() -> socket.socket: + server = socket.socket() + server.bind(("127.0.0.1", 0)) + server.setblocking(True) + server.listen(1) # Just need to listen to the child process. + return server + + +def _calculate_classpath(jars_root: Sequence[pathlib.Path]) -> str: + jars = (p.as_posix() for root in jars_root for p in root.iterdir() if p.suffix == ".jar") + return os.pathsep.join(jars) + + [email protected] +class _MainJar: + path: pathlib.Path + main_class: str + schema_version: str | None + + @classmethod + def find(cls, jars_root: Sequence[pathlib.Path]) -> Self: + for root in jars_root: + for p in root.iterdir(): + if p.suffix != ".jar": + continue + with zipfile.ZipFile(p) as zf: + with zf.open("META-INF/MANIFEST.MF") as f: + manifest = email.message_from_binary_file(f) + if main_class := manifest["Main-Class"]: + return cls(p, main_class, manifest.get("Airflow-SDK-Supervisor-Schema-Version")) Review Comment: `manifest.get("Airflow-SDK-Supervisor-Schema-Version")` returns `None` when the header is missing. That `None` flows through `subprocess_schema_version=` (line 176) into `WatchedSubprocess._subprocess_schema_version`, and the supervisor (`supervisor.py` lines 765 and 792) short-circuits Cadwyn migration on `is None` -- so a JAR built without the header silently runs un-migrated frames against a versioned supervisor. No error, just protocol drift. Suggest failing closed: require the header in `_MainJar.find()` and raise with the JAR path if absent. Bonus: the same applies to a `Schema-Version` whose value isn't in the registered version bundle -- catch that here too instead of letting it explode mid-migration. ########## task-sdk/src/airflow/sdk/coordinators/java/coordinator.py: ########## @@ -0,0 +1,261 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""Java runtime coordinator that launches a JVM subprocess for Dag file processing and task execution.""" + +from __future__ import annotations + +import email +import os +import pathlib +import selectors +import socket +import subprocess +import time +import zipfile +from typing import TYPE_CHECKING, cast + +import attrs +import psutil +import structlog + +from airflow.sdk.execution_time.coordinator import BaseCoordinator +from airflow.sdk.execution_time.supervisor import ActivitySubprocess + +if TYPE_CHECKING: + from collections.abc import Sequence + + from structlog.typing import FilteringBoundLogger + from typing_extensions import Self + + from airflow.sdk.api.client import Client + from airflow.sdk.api.datamodels._generated import BundleInfo + from airflow.sdk.execution_time.workloads.task import TaskInstanceDTO + +log: FilteringBoundLogger = structlog.get_logger(logger_name="coordinators.java") + + +def _start_server() -> socket.socket: + server = socket.socket() + server.bind(("127.0.0.1", 0)) + server.setblocking(True) + server.listen(1) # Just need to listen to the child process. + return server + + +def _calculate_classpath(jars_root: Sequence[pathlib.Path]) -> str: + jars = (p.as_posix() for root in jars_root for p in root.iterdir() if p.suffix == ".jar") + return os.pathsep.join(jars) + + [email protected] +class _MainJar: + path: pathlib.Path + main_class: str + schema_version: str | None + + @classmethod + def find(cls, jars_root: Sequence[pathlib.Path]) -> Self: + for root in jars_root: + for p in root.iterdir(): + if p.suffix != ".jar": + continue + with zipfile.ZipFile(p) as zf: + with zf.open("META-INF/MANIFEST.MF") as f: + manifest = email.message_from_binary_file(f) + if main_class := manifest["Main-Class"]: + return cls(p, main_class, manifest.get("Airflow-SDK-Supervisor-Schema-Version")) + resolved_paths = os.pathsep.join(str(p.resolve()) for p in jars_root) + raise FileNotFoundError(f"cannot fine main class in {resolved_paths}") + + +def _accept_connections( + servers: dict[str, socket.socket], + proc: subprocess.Popen, + *, + max_wait: float = 10.0, +) -> dict[str, socket.socket]: + """Block until the Java process connects to servers.""" + accepted: dict[str, socket.socket] = {} + with selectors.DefaultSelector() as sel: + for key, soc in servers.items(): + sel.register(soc, selectors.EVENT_READ, data=key) + deadline = time.monotonic() + max_wait + while len(accepted) < len(servers): + remaining = deadline - time.monotonic() + if remaining <= 0: + raise TimeoutError("process did not connect within timeout") + if proc.poll() is not None: + raise RuntimeError(f"process exited with {proc.returncode} before connecting") + for event, _ in sel.select(timeout=min(remaining, 1.0)): + log.debug("Accepting child process connection", key=(key := event.data)) + conn, _ = cast("socket.socket", event.fileobj).accept() + sel.unregister(servers[key]) + accepted[key] = conn + return accepted + + [email protected](kw_only=True) +class _JavaActivitySubprocess(ActivitySubprocess): + """Java task runner process.""" + + _comm_server: socket.socket + _logs_server: socket.socket + _subprocess: subprocess.Popen + + # Keep track of channels used to pipe subprocess stdout and stderr so we can + # close them on exit. The "read" side is handled by _register_pipe_readers + # callbacks so we don't need to worry about them. + _stdout_w: socket.socket + _stderr_w: socket.socket + + @classmethod + def start( # type: ignore[override] + cls, + *, + what: TaskInstanceDTO, + dag_rel_path: str | os.PathLike[str], + bundle_info, + logger: FilteringBoundLogger | None = None, + sentry_integration: str = "", + java_executable: str, + jvm_args: list[str], + jars_root: Sequence[pathlib.Path], + **kwargs, + ) -> Self: + jar = _MainJar.find(jars_root) + + comm_server = _start_server() Review Comment: The launch path opens listening sockets and socketpairs, then has multiple raising points before any cleanup path exists: - `_MainJar.find()` `FileNotFoundError` - `_accept_connections` `TimeoutError` / `RuntimeError` - `cls(...)` `TypeError` from attrs construction - `_on_child_started()` failing On any of those, `proc` keeps running (orphan JVM) and all opened fds (`comm_server`, `logs_server`, the four socketpair ends) leak. Wrap the body in try/except that calls `proc.kill()` + `proc.wait()` if `proc` is bound, and closes every socket created so far. ########## task-sdk/src/airflow/sdk/coordinators/java/coordinator.py: ########## @@ -0,0 +1,261 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""Java runtime coordinator that launches a JVM subprocess for Dag file processing and task execution.""" + +from __future__ import annotations + +import email +import os +import pathlib +import selectors +import socket +import subprocess +import time +import zipfile +from typing import TYPE_CHECKING, cast + +import attrs +import psutil +import structlog + +from airflow.sdk.execution_time.coordinator import BaseCoordinator +from airflow.sdk.execution_time.supervisor import ActivitySubprocess + +if TYPE_CHECKING: + from collections.abc import Sequence + + from structlog.typing import FilteringBoundLogger + from typing_extensions import Self + + from airflow.sdk.api.client import Client + from airflow.sdk.api.datamodels._generated import BundleInfo + from airflow.sdk.execution_time.workloads.task import TaskInstanceDTO + +log: FilteringBoundLogger = structlog.get_logger(logger_name="coordinators.java") + + +def _start_server() -> socket.socket: + server = socket.socket() + server.bind(("127.0.0.1", 0)) + server.setblocking(True) + server.listen(1) # Just need to listen to the child process. + return server + + +def _calculate_classpath(jars_root: Sequence[pathlib.Path]) -> str: + jars = (p.as_posix() for root in jars_root for p in root.iterdir() if p.suffix == ".jar") + return os.pathsep.join(jars) + + [email protected] +class _MainJar: + path: pathlib.Path + main_class: str + schema_version: str | None + + @classmethod + def find(cls, jars_root: Sequence[pathlib.Path]) -> Self: + for root in jars_root: + for p in root.iterdir(): + if p.suffix != ".jar": + continue + with zipfile.ZipFile(p) as zf: + with zf.open("META-INF/MANIFEST.MF") as f: + manifest = email.message_from_binary_file(f) + if main_class := manifest["Main-Class"]: + return cls(p, main_class, manifest.get("Airflow-SDK-Supervisor-Schema-Version")) + resolved_paths = os.pathsep.join(str(p.resolve()) for p in jars_root) + raise FileNotFoundError(f"cannot fine main class in {resolved_paths}") Review Comment: Typo: "cannot fine main class" -> "cannot find main class". ########## task-sdk/src/airflow/sdk/execution_time/coordinator.py: ########## @@ -0,0 +1,227 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +""" +Runtime coordinator for non-Python DAG file processing and task execution. + +Provides :class:`BaseCoordinator`, the base class for +SDK-specific coordinators that bridge subprocess I/O between the +Airflow supervisor and an external-SDK runtime (Java, Go, Rust, etc.), +and :class:`CoordinatorManager`, the registry that loads coordinator +instances from the ``[sdk] coordinators`` configuration. + +The coordinator's :meth:`~BaseCoordinator.run_task_execution` handles the full +lifecycle: + +1. Creates TCP servers for comm and logs channels, and a socketpair for stderr. +2. Calls :meth:`~BaseCoordinator.task_execution_cmd` (provided by the subclass) + to obtain the subprocess command. +3. Spawns the subprocess and accepts TCP connections from it. +4. Runs a selector-based bridge that transparently forwards bytes + between fd 0 (supervisor) and the subprocess comm socket, and + re-emits the subprocess's log and stderr output through structlog. +""" + +from __future__ import annotations + +import contextlib +import functools +from typing import TYPE_CHECKING, Any + +import attrs +import pydantic + +from airflow.sdk._shared.module_loading import import_string +from airflow.sdk.configuration import conf + +if TYPE_CHECKING: + from collections.abc import Mapping + from os import PathLike + + from structlog.typing import FilteringBoundLogger + from typing_extensions import Self + + from airflow.sdk.api.client import Client + from airflow.sdk.execution_time.workloads.task import TaskInstanceDTO + +__all__ = [ + "BaseCoordinator", + "CoordinatorManager", + "get_coordinator_manager", + "reset_coordinator_manager", +] + + +class BaseCoordinator: + """ + Base coordinator for runtime-specific DAG file processing and task execution. + + Coordinators are instantiated from the ``[sdk] coordinators`` configuration + (see :class:`CoordinatorManager`) — each entry's ``classpath`` is resolved + via :func:`~airflow.sdk._shared.module_loading.import_string` and + constructed with the entry's ``kwargs``. + """ + + @attrs.define(slots=True) + class ExecutionResult: + """Return value for :meth:`BaseCoordinator.execute_task`.""" + + exit_code: Any + final_state: str + + def execute_task( + self, + *, + what: TaskInstanceDTO, + dag_rel_path: str | PathLike[str], + bundle_info, + client: Client, + logger: FilteringBoundLogger | None = None, + sentry_integration: str = "", + subprocess_logs_to_stdout: bool, + **kwargs, + ) -> ExecutionResult: + """ + Start task execution. + + This should execute the task and return a result. + """ + raise NotImplementedError + + +class _CoordinatorSpec(pydantic.BaseModel): + classpath: str + kwargs: dict[str, Any] Review Comment: `_CoordinatorSpec.kwargs` is required, but `config.yml` documents it as optional: > Each value is an object with `classpath` and **optional** `kwargs`. Result: a minimal entry like `{"jdk-17": {"classpath": "airflow.sdk.coordinators.java.JavaCoordinator"}}` fails pydantic validation at startup instead of constructing with defaults. Suggest: `kwargs: dict[str, Any] = pydantic.Field(default_factory=dict)`. ########## task-sdk/src/airflow/sdk/coordinators/java/coordinator.py: ########## @@ -0,0 +1,261 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""Java runtime coordinator that launches a JVM subprocess for Dag file processing and task execution.""" + +from __future__ import annotations + +import email +import os +import pathlib +import selectors +import socket +import subprocess +import time +import zipfile +from typing import TYPE_CHECKING, cast + +import attrs +import psutil +import structlog + +from airflow.sdk.execution_time.coordinator import BaseCoordinator +from airflow.sdk.execution_time.supervisor import ActivitySubprocess + +if TYPE_CHECKING: + from collections.abc import Sequence + + from structlog.typing import FilteringBoundLogger + from typing_extensions import Self + + from airflow.sdk.api.client import Client + from airflow.sdk.api.datamodels._generated import BundleInfo + from airflow.sdk.execution_time.workloads.task import TaskInstanceDTO + +log: FilteringBoundLogger = structlog.get_logger(logger_name="coordinators.java") + + +def _start_server() -> socket.socket: + server = socket.socket() + server.bind(("127.0.0.1", 0)) + server.setblocking(True) + server.listen(1) # Just need to listen to the child process. + return server + + +def _calculate_classpath(jars_root: Sequence[pathlib.Path]) -> str: + jars = (p.as_posix() for root in jars_root for p in root.iterdir() if p.suffix == ".jar") + return os.pathsep.join(jars) + + [email protected] +class _MainJar: + path: pathlib.Path + main_class: str + schema_version: str | None + + @classmethod + def find(cls, jars_root: Sequence[pathlib.Path]) -> Self: + for root in jars_root: + for p in root.iterdir(): + if p.suffix != ".jar": + continue + with zipfile.ZipFile(p) as zf: + with zf.open("META-INF/MANIFEST.MF") as f: + manifest = email.message_from_binary_file(f) + if main_class := manifest["Main-Class"]: + return cls(p, main_class, manifest.get("Airflow-SDK-Supervisor-Schema-Version")) + resolved_paths = os.pathsep.join(str(p.resolve()) for p in jars_root) + raise FileNotFoundError(f"cannot fine main class in {resolved_paths}") + + +def _accept_connections( + servers: dict[str, socket.socket], + proc: subprocess.Popen, + *, + max_wait: float = 10.0, +) -> dict[str, socket.socket]: + """Block until the Java process connects to servers.""" + accepted: dict[str, socket.socket] = {} + with selectors.DefaultSelector() as sel: + for key, soc in servers.items(): + sel.register(soc, selectors.EVENT_READ, data=key) + deadline = time.monotonic() + max_wait + while len(accepted) < len(servers): + remaining = deadline - time.monotonic() + if remaining <= 0: + raise TimeoutError("process did not connect within timeout") + if proc.poll() is not None: + raise RuntimeError(f"process exited with {proc.returncode} before connecting") + for event, _ in sel.select(timeout=min(remaining, 1.0)): + log.debug("Accepting child process connection", key=(key := event.data)) + conn, _ = cast("socket.socket", event.fileobj).accept() + sel.unregister(servers[key]) + accepted[key] = conn + return accepted + + [email protected](kw_only=True) +class _JavaActivitySubprocess(ActivitySubprocess): + """Java task runner process.""" + + _comm_server: socket.socket + _logs_server: socket.socket + _subprocess: subprocess.Popen + + # Keep track of channels used to pipe subprocess stdout and stderr so we can + # close them on exit. The "read" side is handled by _register_pipe_readers + # callbacks so we don't need to worry about them. + _stdout_w: socket.socket + _stderr_w: socket.socket + + @classmethod + def start( # type: ignore[override] + cls, + *, + what: TaskInstanceDTO, + dag_rel_path: str | os.PathLike[str], + bundle_info, + logger: FilteringBoundLogger | None = None, + sentry_integration: str = "", + java_executable: str, + jvm_args: list[str], + jars_root: Sequence[pathlib.Path], + **kwargs, + ) -> Self: + jar = _MainJar.find(jars_root) + + comm_server = _start_server() + logs_server = _start_server() + + stdout_r, stdout_w = socket.socketpair() + stderr_r, stderr_w = socket.socketpair() + + comm_host, comm_port = comm_server.getsockname() + logs_host, logs_port = logs_server.getsockname() + + proc = subprocess.Popen( + [ + java_executable, + "-classpath", + _calculate_classpath(jars_root), + *jvm_args, + jar.main_class, + # Arguments to MainClass... + f"--comm={comm_host}:{comm_port}", + f"--logs={logs_host}:{logs_port}", + ], + stdout=stdout_w.makefile("wb", buffering=0).fileno(), + stderr=stderr_w.makefile("wb", buffering=0).fileno(), + ) + log.info("Starting subprocess", pid=proc.pid) + socks = _accept_connections({"comm": comm_server, "logs": logs_server}, proc) + + self = cls( + id=what.id, + pid=proc.pid, + process=psutil.Process(proc.pid), + process_log=logger or structlog.get_logger(logger_name="task").bind(), + start_time=time.monotonic(), + stdin=socks["comm"], + subprocess=proc, + subprocess_schema_version=jar.schema_version, + comm_server=comm_server, + logs_server=logs_server, + stdout_w=stdout_w, + stderr_w=stderr_w, + **kwargs, + ) + self._register_pipe_readers(stdout_r, stderr_r, socks["comm"], socks["logs"]) + self._on_child_started( + ti=what, + dag_rel_path=dag_rel_path, + bundle_info=bundle_info, + sentry_integration=sentry_integration, + ) + return self + + def wait(self) -> int: + code = super().wait() + self._close_unused_sockets(self._comm_server, self._logs_server, self._stdout_w, self._stderr_w) Review Comment: `stdout_w` / `stderr_w` are kept open on the parent until after `super().wait()` returns. But the parent's `wait()` watches the read-ends and waits for EOF, and EOF on a socketpair only arrives when *all* write-end fds close -- including the parent's. Net effect: even after the JVM exits cleanly, tasks sit waiting for the socket-cleanup timeout because the parent is still holding the write ends open. Fix: close `stdout_w` / `stderr_w` in `start()` immediately after the `subprocess.Popen` call -- the kernel has already dup'd those fds into the child by then, so the parent's copies are no longer needed. ########## task-sdk/src/airflow/sdk/execution_time/coordinator.py: ########## @@ -0,0 +1,227 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +""" +Runtime coordinator for non-Python DAG file processing and task execution. + +Provides :class:`BaseCoordinator`, the base class for +SDK-specific coordinators that bridge subprocess I/O between the +Airflow supervisor and an external-SDK runtime (Java, Go, Rust, etc.), +and :class:`CoordinatorManager`, the registry that loads coordinator +instances from the ``[sdk] coordinators`` configuration. + +The coordinator's :meth:`~BaseCoordinator.run_task_execution` handles the full +lifecycle: + +1. Creates TCP servers for comm and logs channels, and a socketpair for stderr. +2. Calls :meth:`~BaseCoordinator.task_execution_cmd` (provided by the subclass) + to obtain the subprocess command. +3. Spawns the subprocess and accepts TCP connections from it. +4. Runs a selector-based bridge that transparently forwards bytes + between fd 0 (supervisor) and the subprocess comm socket, and + re-emits the subprocess's log and stderr output through structlog. +""" + +from __future__ import annotations + +import contextlib +import functools +from typing import TYPE_CHECKING, Any + +import attrs +import pydantic + +from airflow.sdk._shared.module_loading import import_string +from airflow.sdk.configuration import conf + +if TYPE_CHECKING: + from collections.abc import Mapping + from os import PathLike + + from structlog.typing import FilteringBoundLogger + from typing_extensions import Self + + from airflow.sdk.api.client import Client + from airflow.sdk.execution_time.workloads.task import TaskInstanceDTO + +__all__ = [ + "BaseCoordinator", + "CoordinatorManager", + "get_coordinator_manager", + "reset_coordinator_manager", +] + + +class BaseCoordinator: + """ + Base coordinator for runtime-specific DAG file processing and task execution. + + Coordinators are instantiated from the ``[sdk] coordinators`` configuration + (see :class:`CoordinatorManager`) — each entry's ``classpath`` is resolved + via :func:`~airflow.sdk._shared.module_loading.import_string` and + constructed with the entry's ``kwargs``. + """ + + @attrs.define(slots=True) + class ExecutionResult: + """Return value for :meth:`BaseCoordinator.execute_task`.""" + + exit_code: Any + final_state: str + + def execute_task( + self, + *, + what: TaskInstanceDTO, + dag_rel_path: str | PathLike[str], + bundle_info, + client: Client, + logger: FilteringBoundLogger | None = None, + sentry_integration: str = "", + subprocess_logs_to_stdout: bool, + **kwargs, + ) -> ExecutionResult: + """ + Start task execution. + + This should execute the task and return a result. + """ + raise NotImplementedError + + +class _CoordinatorSpec(pydantic.BaseModel): + classpath: str + kwargs: dict[str, Any] + + +class _PythonCoordinator(BaseCoordinator): + """ + Coordinator implementation to execute Python tasks. + + This is not supposed to be specified by users directly, but the fallback + used by default when nothing is specified. + """ + + def execute_task( + self, + *, + what: TaskInstanceDTO, + dag_rel_path: str | PathLike[str], + bundle_info, + client: Client, + logger: FilteringBoundLogger | None = None, + sentry_integration: str = "", + subprocess_logs_to_stdout: bool, + **kwargs, + ) -> BaseCoordinator.ExecutionResult: + # TODO: Move this to somewhere that makes more sense. + from airflow.sdk.execution_time.supervisor import ActivitySubprocess + + process = ActivitySubprocess.start( + dag_rel_path=dag_rel_path, + what=what, + client=client, + logger=logger, + bundle_info=bundle_info, + subprocess_logs_to_stdout=subprocess_logs_to_stdout, + sentry_integration=sentry_integration, + ) + exit_code = process.wait() + return self.ExecutionResult(exit_code, process.final_state) + + [email protected] +def _build_python_coordinator() -> _PythonCoordinator: + return _PythonCoordinator() + + [email protected](kw_only=True) +class CoordinatorManager: + """ + Registry of coordinator instances loaded from ``[sdk]`` configurations. + + The ``[sdk] coordinators`` value is a JSON object keyed by coordinator name:: + + { + "jdk-11": { + "classpath": "airflow.sdk.coordinators.java.JavaCoordinator", + "kwargs": {"java_executable": "/usr/lib/jvm/jdk-11/bin/java", ...}, + } + } + + The ``classpath`` is resolved via + :func:`~airflow.sdk._shared.module_loading.import_string` (no + :class:`ProvidersManager` involvement) and constructed with ``kwargs`` on + first use. A coordinator entry that is never looked up incurs no startup + cost. At most one coordinator object can be created from each entry. + + The ``[sdk] queue_to_coordinator`` config maps queue names to a key in the + object, which lets users reuse existing queue assignments to route tasks to + a specific coordinator instance (for example, a ``"legacy-java"`` queue + routed to a JDK 11 coordinator, and a ``"modern-java"`` queue routed to a + JDK 17 coordinator). + + :meta private: + """ + + _coordinator_specs: Mapping[str, _CoordinatorSpec] + _queue_to_coordinator: Mapping[str, str] + + _created_coordinators: dict[str, BaseCoordinator] = attrs.field(init=False, factory=dict) + + @classmethod + def from_config(cls) -> Self: + """Load coordinator specs from configuration without initialization.""" + coordinator_specs = { + k: _CoordinatorSpec.model_validate(v) + for k, v in conf.getjson("sdk", "coordinators", fallback={}).items() + } + queue_to_coordinator = conf.getjson("sdk", "queue_to_coordinator", fallback={}) + for key in queue_to_coordinator.values(): + if key not in coordinator_specs: + raise ValueError(f"[sdk] queue_to_coordinator references invalid coordinator key: {key!r}") + return cls(coordinator_specs=coordinator_specs, queue_to_coordinator=queue_to_coordinator) + + def _for_queue_internal(self, queue: str) -> BaseCoordinator: + key = self._queue_to_coordinator[queue] + with contextlib.suppress(KeyError): + return self._created_coordinators[key] + spec = self._coordinator_specs[key] + coordinator = self._created_coordinators[key] = import_string(spec.classpath)(**spec.kwargs) Review Comment: `import_string(spec.classpath)(**spec.kwargs)` errors are bare: - Wrong dotted path -> `ImportError` with no coordinator-key context. - Unknown kwarg on an attrs `kw_only=True` class (like `JavaCoordinator`) -> `TypeError: __init__() got an unexpected keyword argument 'foo'` with no hint which `coordinators` entry produced it. - The result isn't checked to be a `BaseCoordinator` -- a typo'd classpath pointing at some other importable class will only fail at `coordinator.execute_task(...)` later. Suggest: catch `ImportError`/`TypeError` here and re-raise with the offending coordinator key + classpath, and `isinstance(coordinator, BaseCoordinator)` check the result before caching. ########## task-sdk/src/airflow/sdk/coordinators/java/coordinator.py: ########## @@ -0,0 +1,261 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""Java runtime coordinator that launches a JVM subprocess for Dag file processing and task execution.""" + +from __future__ import annotations + +import email +import os +import pathlib +import selectors +import socket +import subprocess +import time +import zipfile +from typing import TYPE_CHECKING, cast + +import attrs +import psutil +import structlog + +from airflow.sdk.execution_time.coordinator import BaseCoordinator +from airflow.sdk.execution_time.supervisor import ActivitySubprocess + +if TYPE_CHECKING: + from collections.abc import Sequence + + from structlog.typing import FilteringBoundLogger + from typing_extensions import Self + + from airflow.sdk.api.client import Client + from airflow.sdk.api.datamodels._generated import BundleInfo + from airflow.sdk.execution_time.workloads.task import TaskInstanceDTO + +log: FilteringBoundLogger = structlog.get_logger(logger_name="coordinators.java") + + +def _start_server() -> socket.socket: + server = socket.socket() + server.bind(("127.0.0.1", 0)) + server.setblocking(True) + server.listen(1) # Just need to listen to the child process. + return server + + +def _calculate_classpath(jars_root: Sequence[pathlib.Path]) -> str: + jars = (p.as_posix() for root in jars_root for p in root.iterdir() if p.suffix == ".jar") + return os.pathsep.join(jars) + + [email protected] +class _MainJar: + path: pathlib.Path + main_class: str + schema_version: str | None + + @classmethod + def find(cls, jars_root: Sequence[pathlib.Path]) -> Self: + for root in jars_root: + for p in root.iterdir(): + if p.suffix != ".jar": + continue + with zipfile.ZipFile(p) as zf: + with zf.open("META-INF/MANIFEST.MF") as f: + manifest = email.message_from_binary_file(f) + if main_class := manifest["Main-Class"]: + return cls(p, main_class, manifest.get("Airflow-SDK-Supervisor-Schema-Version")) + resolved_paths = os.pathsep.join(str(p.resolve()) for p in jars_root) + raise FileNotFoundError(f"cannot fine main class in {resolved_paths}") + + +def _accept_connections( + servers: dict[str, socket.socket], + proc: subprocess.Popen, + *, + max_wait: float = 10.0, +) -> dict[str, socket.socket]: + """Block until the Java process connects to servers.""" + accepted: dict[str, socket.socket] = {} + with selectors.DefaultSelector() as sel: + for key, soc in servers.items(): + sel.register(soc, selectors.EVENT_READ, data=key) + deadline = time.monotonic() + max_wait + while len(accepted) < len(servers): + remaining = deadline - time.monotonic() + if remaining <= 0: + raise TimeoutError("process did not connect within timeout") + if proc.poll() is not None: + raise RuntimeError(f"process exited with {proc.returncode} before connecting") + for event, _ in sel.select(timeout=min(remaining, 1.0)): + log.debug("Accepting child process connection", key=(key := event.data)) + conn, _ = cast("socket.socket", event.fileobj).accept() + sel.unregister(servers[key]) + accepted[key] = conn + return accepted + + [email protected](kw_only=True) +class _JavaActivitySubprocess(ActivitySubprocess): + """Java task runner process.""" + + _comm_server: socket.socket + _logs_server: socket.socket + _subprocess: subprocess.Popen + + # Keep track of channels used to pipe subprocess stdout and stderr so we can + # close them on exit. The "read" side is handled by _register_pipe_readers + # callbacks so we don't need to worry about them. + _stdout_w: socket.socket + _stderr_w: socket.socket + + @classmethod + def start( # type: ignore[override] + cls, + *, + what: TaskInstanceDTO, + dag_rel_path: str | os.PathLike[str], + bundle_info, + logger: FilteringBoundLogger | None = None, + sentry_integration: str = "", + java_executable: str, + jvm_args: list[str], + jars_root: Sequence[pathlib.Path], + **kwargs, + ) -> Self: + jar = _MainJar.find(jars_root) + + comm_server = _start_server() + logs_server = _start_server() + + stdout_r, stdout_w = socket.socketpair() + stderr_r, stderr_w = socket.socketpair() + + comm_host, comm_port = comm_server.getsockname() + logs_host, logs_port = logs_server.getsockname() + + proc = subprocess.Popen( + [ + java_executable, + "-classpath", + _calculate_classpath(jars_root), + *jvm_args, + jar.main_class, + # Arguments to MainClass... + f"--comm={comm_host}:{comm_port}", + f"--logs={logs_host}:{logs_port}", + ], + stdout=stdout_w.makefile("wb", buffering=0).fileno(), + stderr=stderr_w.makefile("wb", buffering=0).fileno(), + ) + log.info("Starting subprocess", pid=proc.pid) + socks = _accept_connections({"comm": comm_server, "logs": logs_server}, proc) + + self = cls( + id=what.id, + pid=proc.pid, + process=psutil.Process(proc.pid), + process_log=logger or structlog.get_logger(logger_name="task").bind(), + start_time=time.monotonic(), + stdin=socks["comm"], + subprocess=proc, + subprocess_schema_version=jar.schema_version, + comm_server=comm_server, + logs_server=logs_server, + stdout_w=stdout_w, + stderr_w=stderr_w, + **kwargs, + ) + self._register_pipe_readers(stdout_r, stderr_r, socks["comm"], socks["logs"]) + self._on_child_started( + ti=what, + dag_rel_path=dag_rel_path, + bundle_info=bundle_info, + sentry_integration=sentry_integration, + ) + return self + + def wait(self) -> int: + code = super().wait() + self._close_unused_sockets(self._comm_server, self._logs_server, self._stdout_w, self._stderr_w) + return code + + +def _convert_jars_root( + value: None | os.PathLike[str] | pathlib.Path | list[os.PathLike[str] | pathlib.Path], +) -> list[pathlib.Path]: + if value is None: + return [] + if isinstance(value, (str, os.PathLike, pathlib.Path)): + return [pathlib.Path(value)] + return [pathlib.Path(v) for v in value] Review Comment: `_convert_jars_root` wraps each value in `pathlib.Path(v)` without `.expanduser()`. The class docstring example shows `"jars_root": ["~/airflow/jars"]` -- that path will `FileNotFoundError` at `iterdir()` because the tilde is never expanded. Either call `.expanduser()` in the converter (preferred -- it matches the docstring's stated affordance), or drop the tilde from the docstring example. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
