potiuk commented on code in PR #30255:
URL: https://github.com/apache/airflow/pull/30255#discussion_r1156750330
##########
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:
I actually removed it later in the following PRs it was only needed
temporaryily to fix some of the tests (final version does not need it). I will
remove it from here as well - I think I can do it now after some other changes
- that could have been a remnant that is not really needed.
--
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]
