uranusjr commented on code in PR #30255:
URL: https://github.com/apache/airflow/pull/30255#discussion_r1156463101
##########
airflow/cli/commands/dag_processor_command.py:
##########
@@ -19,30 +19,40 @@
import logging
from datetime import timedelta
+from typing import Any
import daemon
from daemon.pidfile import TimeoutPIDLockFile
from airflow import settings
from airflow.configuration import conf
-from airflow.jobs.dag_processor_job import DagProcessorJob
+from airflow.jobs.base_job import BaseJob
+from airflow.jobs.dag_processor_job import DagProcessorJobRunner
from airflow.utils import cli as cli_utils
from airflow.utils.cli import setup_locations, setup_logging
log = logging.getLogger(__name__)
+dag_processor_job_runner: DagProcessorJobRunner | None = None
-def _create_dag_processor_job(args) -> DagProcessorJob:
+
+def _create_dag_processor_job(args: Any) -> BaseJob:
"""Creates DagFileProcessorProcess instance."""
+ from airflow.dag_processing.manager import DagFileProcessorManager
+
processor_timeout_seconds: int = conf.getint("core",
"dag_file_processor_timeout")
processor_timeout = timedelta(seconds=processor_timeout_seconds)
- return DagProcessorJob(
+ global dag_processor_job_runner
+ processor = DagFileProcessorManager(
processor_timeout=processor_timeout,
dag_directory=args.subdir,
max_runs=args.num_runs,
dag_ids=[],
pickle_dags=args.do_pickle,
)
+ return BaseJob(
+ job_runner=processor.job_runner,
+ )
Review Comment:
Code here looks wrong? `dag_processor_job_runner` is never assigned anywhere.
FWIW I also dislike the global mutable variable design. A
`get_dag_processor` function with `lru_cache` would be significantly better.
--
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]
