phanikumv commented on code in PR #74341: URL: https://github.com/apache/airflow/pull/74341#discussion_r4205661480
########## task-sdk/src/airflow/sdk/coordinators/executable/_dag_importer.py: ########## @@ -0,0 +1,106 @@ +# +# 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. +"""The Dag importer of :class:`~airflow.sdk.coordinators.executable.ExecutableCoordinator`.""" + +from __future__ import annotations + +import os +from pathlib import Path +from typing import TYPE_CHECKING, ClassVar, Final + +from airflow.sdk._shared.module_loading.file_discovery import find_path_from_directory +from airflow.sdk.configuration import conf +from airflow.sdk.coordinators._dag_importer import CoordinatorDagImporter +from airflow.sdk.coordinators.executable._bundle_reader import ( + read_bundle_entrypoint_source, + read_bundle_language, + read_bundle_source, +) +from airflow.sdk.coordinators.executable.coordinator import FOOTER_MAGIC +from airflow.sdk.importers.base import DagSourceCode, FilesystemDagDefinition + +if TYPE_CHECKING: + from collections.abc import Iterator + + from airflow.dag_processing.bundles.base import BaseDagBundle # noqa: SDK002 + from airflow.sdk.importers.base import DagDefinition + +_NO_SOURCE: Final = "// Source code is not available: the bundle embeds no source.\n" +_DEFAULT_LANGUAGE: Final = "text" + + +class ExecutableDagImporter(CoordinatorDagImporter): + """ + Claim the native Dags of executable bundles, such as the ones the Go SDK packs. + + An :class:`~airflow.sdk.coordinators.executable.ExecutableCoordinator` parses them. A bundle is + identified by the ``AFBNDL01`` magic that ends the file, not by its name, so it can have any extension + or none. + """ + + coordinator_classpath: ClassVar[str] = "airflow.sdk.coordinators.executable.ExecutableCoordinator" + artifact_suffix = "" + supported_extensions: list[str] = [] + + def can_handle(self, definition: DagDefinition | str | Path) -> bool: + try: + with open(str(definition), "rb") as bundle_file: + bundle_file.seek(-len(FOOTER_MAGIC), os.SEEK_END) + return bundle_file.read(len(FOOTER_MAGIC)) == FOOTER_MAGIC + except OSError: + return False Review Comment: This is the first importer whose `can_handle` reads the file, and `execute_task` calls `_is_native_dag_file(rel_path, ...)` with the bundle-relative `relative_fileloc` before resolving the bundle. `open("bin/orders")` resolves against the worker's cwd, so this returns False and the task goes down the dag_id-scan path Java/Node don't hit this because the base `can_handle` is suffix-only. I think the fix belongs in `SubprocessCoordinator.execute_task`: keep the cheap suffix check first (so the `mock_initialize.assert_not_called()` tests still hold), and if that says "not native" and the bundle has a content-claiming importer, resolve the task bundle and re-check `task_bundle.path / rel_path` (after `_is_file_in_bundle`), then reuse that bundle. Same idea as `task_runner.py`, which builds `Path(bundle_instance.path, rel_path)` before classifying. `dag_processing/manager.py:813` needs the same treatment -- 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]
