uranusjr commented on code in PR #65956: URL: https://github.com/apache/airflow/pull/65956#discussion_r3158935545
########## java-sdk/adr/0002-dag-parsing.md: ########## @@ -0,0 +1,370 @@ +<!-- + 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. + --> + +# ADR-0002: DAG Parsing — Language-Specific DAG File Processing + +## Status + +Accepted + +## Context + +Airflow's standard DAG file processor only understands Python files. To support DAGs defined in other languages (Java, Go, Rust, etc.), the pipeline needs an extension point where a language-specific processor can intercept the parsing request, delegate to an external runtime, and return a result in the same format the Airflow scheduler expects. + +This ADR details the DAG parsing side of the coordinator architecture described in [ADR-0001](0001-java-sdk-airflow-integration.md). It starts with the generic model — the abstract contracts and expected behavior that any language must implement — then walks through Java as a concrete example. + +## Decision + +### Extension Point: `BaseCoordinator` + +A single abstract base class — `BaseCoordinator` — handles both DAG parsing and task execution. It is registered in `provider.yaml` under `coordinators`. For DAG parsing, a subclass must implement two methods: + +| Method | Signature | Responsibility | +|---|---|---| +| `can_handle_dag_file` | `(bundle_name, path) -> bool` | Return `True` if this coordinator should handle the given file. Default returns `False`; subclasses add language-specific checks (e.g., "is this a JAR with a Main-Class?"). | +| `dag_parsing_cmd` | `(dag_file_path, bundle_name, bundle_path, comm_addr, logs_addr) -> list[str]` | Return the full command to launch the language runtime. `comm_addr` and `logs_addr` are `host:port` strings the process must connect to. | + +### Registration + +In the provider's `provider.yaml`: + +```yaml +process-coordinators: + - airflow.providers.sdk.<lang>.coordinator.<LangCoordinator> +``` + +A single registration covers both DAG parsing and task execution — there are no separate `dag-file-processors` or `task-coordinators` keys. + +### Discovery: `_resolve_processor_target()` + +When `DagFileProcessorProcess.start()` needs to parse a file: + +``` +_resolve_processor_target(path, bundle_name, bundle_path) + for each class_path in ProvidersManager().coordinators: + coordinator_cls = import_string(class_path) + if coordinator_cls.can_handle_dag_file(bundle_name, path): + return functools.partial(coordinator_cls.run_dag_parsing, path=..., bundle_name=..., bundle_path=...) + return None # fall back to default Python parser +``` + +The first coordinator whose `can_handle_dag_file()` returns `True` wins. If none match, the default Python `_parse_file_entrypoint` runs. + +### What the Base Class Handles Automatically + +The matched coordinator's `run_dag_parsing()` (a concrete method on `BaseCoordinator`) delegates to `_runtime_subprocess_entrypoint()`, which handles all the TCP/process plumbing: + +1. Creates two TCP servers on `127.0.0.1` with random ports (comm + logs) +2. Creates a stderr socketpair +3. Calls `dag_parsing_cmd()` to get the command +4. Spawns the subprocess with `stdin=DEVNULL` (does NOT inherit fd 0) +5. Accepts TCP connections from the subprocess +6. Wraps fd 0 as `supervisor_comm` via `os.dup(0)` +7. Runs `_bridge()` — a raw byte forwarder between fd 0 and the TCP comm socket + +### Expected E2E Flow + +``` +Airflow Dag-Processor + │ + ▼ +DagFileProcessorProcess.start(path, bundle_name, bundle_path) + │ + ├─ _resolve_processor_target() + │ └─ iterates process-coordinators from provider.yaml + │ └─ first can_handle_dag_file() == True wins + │ + ▼ +WatchedSubprocess.start(target=coordinator.run_dag_parsing) + │ + [fork — child process gets fd 0 as Unix domain socket to supervisor] + │ + ▼ (in child) +<Lang>Coordinator.run_dag_parsing(path, bundle_name, bundle_path) + │ + ▼ +BaseCoordinator._runtime_subprocess_entrypoint(DagParsingInfo) + │ + ├─ 1. Create TCP comm_server + logs_server on 127.0.0.1:random + ├─ 2. Create stderr socketpair + ├─ 3. Call dag_parsing_cmd() → get launch command + ├─ 4. Popen(cmd, stdin=DEVNULL, stderr=child_stderr) + ├─ 5. Accept TCP connections from the language runtime + ├─ 6. supervisor_comm = socket(fileno=os.dup(0)) + └─ 7. _bridge() — raw byte forwarding until process exits +``` + +### Expected Message Sequence + +Once the bridge is running, the Airflow supervisor and the language runtime communicate directly through the bridge (raw bytes, no re-encoding): + +``` +Airflow Supervisor Bridge Language Runtime + │ │ │ + ├── DagFileParseRequest ──────────┼──────────────────────►│ + │ [4-byte len][msgpack frame] │ raw byte forward │ + │ │ │ + │ │ ├── parse DAGs from + │ │ │ bundle/file + │ │ │ + │◄── DagFileParsingResult ────────┼───────────────────────┤ + │ [4-byte len][msgpack frame] │ raw byte forward │ + │ │ │ + │ │ └── exit(0) + │ │ + │ └── drain remaining bytes (5s deadline) + │ close all sockets +``` + +### DagFileParsingResult Format + +The language runtime must produce a `DagFileParsingResult` that matches Python Airflow's DagSerialization format exactly. The Airflow scheduler deserializes this into its internal model — any divergence causes parsing failures. Review Comment: The internal representation is actually defined in `airflow-core/src/airflow/serialization/schema.json`, although I don’t know if there’s an established tool that generates POJO from a schema (reverse tools exist). We can probably use a pre-commit check to make sure the serialized DAG definition in Java matches the JSON schema? Regarding DagFileParsingResult (and also StartupDetails really), I don’t think they are currently documented anywhere. Now that they are no longer used for Python-to-Python, we should probably generate schema files for them too. This probably should be done separately since it’s not really in-scope of this feature specifically. -- 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]
