potiuk commented on code in PR #30255:
URL: https://github.com/apache/airflow/pull/30255#discussion_r1156758977
##########
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:
Just a thought @uranusjr -> but maybe indeed it is a good idea to just
review
https://github.com/apache/airflow/pull/30376 ?
This is a merge of those 4 PRS which I tried to keep separate and address
one part of the refactor at a time. But the final stage is "WAY" cleaner than
any of the intermediate steps. Those intermediate steps were only really needed
to make sure that there is a "surgical" change at each of the steps while the
ORM -> logic relation had been unentangled and to keep the tests green (and to
test each step separately)
WDYT?
--
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]
